Support concurrent clients

Author: chenshuo

protorpc/echo/EchoClient.java

@@ -1,6 +1,13 @@
 package echo;
 
 import java.net.InetSocketAddress;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+
+import org.jboss.netty.channel.ChannelFactory;
+import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
+
+import com.google.protobuf.ServiceException;
 
 import muduo.rpc.RpcChannel;
 import muduo.rpc.RpcClient;
@@ -10,27 +17,65 @@
 import echo.EchoProto.EchoService.BlockingInterface;
 
 public class EchoClient {
+    static final int kRequests = 20000;
+
+    public static class Client implements Runnable {
+        private ChannelFactory channelFactory;
+        private InetSocketAddress serverAddr;
+        private CountDownLatch latch;
+
+        public Client(ChannelFactory channelFactory, InetSocketAddress server, CountDownLatch latch) {
+            this.channelFactory = channelFactory;
+            this.serverAddr = server;
+            this.latch = latch;
+        }
+
+        @Override
+        public void run() {
+            System.out.println(Thread.currentThread());
+            RpcClient client = new RpcClient(channelFactory);
+            RpcChannel channel = client.blockingConnect(serverAddr);
+            BlockingInterface remoteService = EchoService.newBlockingStub(channel);
+            String payload = new String(new byte[100]);
+            payload = "Hello";
+            EchoRequest request = EchoRequest.newBuilder().setPayload(payload).build();
+
+            for (int i = 0; i < kRequests; ++i) {
+                EchoResponse response;
+                try {
+                    response = remoteService.echo(null, request);
+                    assert response.getPayload().equals(payload);
+                } catch (ServiceException e) {
+                    // TODO Auto-generated catch block
+                    e.printStackTrace();
+                }
+                // System.out.println(response);
+            }
+            latch.countDown();
+            System.out.println(Thread.currentThread());
+            // System.out.println(response);
+            channel.disconnect();
+            // client.stop();
+        }
+    }
 
     public static void main(String[] args) throws Exception {
-        RpcClient client = new RpcClient();
-        RpcChannel channel = client.blockingConnect(new InetSocketAddress(args[0], 8888));
-        BlockingInterface remoteService = EchoService.newBlockingStub(channel);
-        String payload = new String(new byte[100]);
-        payload = "Hello";
-        EchoRequest request = EchoRequest.newBuilder().setPayload(payload).build();
+        ChannelFactory channelFactory = new NioClientSocketChannelFactory(
+                Executors.newCachedThreadPool(),
+                Executors.newCachedThreadPool());
+        InetSocketAddress server = new InetSocketAddress(args[0], 8888);
+        int N = 4;
+        CountDownLatch latch = new CountDownLatch(N);
         long start = System.currentTimeMillis();
-        int N = 20000;
+        Thread[] threads = new Thread[N];
         for (int i = 0; i < N; ++i) {
-            EchoResponse response = remoteService.echo(null, request);
-            assert response.getPayload().equals(payload);
-            //System.out.println(response);
+            threads[i] = new Thread(new Client(channelFactory, server, latch));
+            threads[i].start();
         }
+        latch.await();
         long end = System.currentTimeMillis();
-        System.err.println(end-start);
-        System.err.println(N*1000L/(end-start));
-        // System.out.println(response);
-        channel.disconnect();
-        client.stop();
+        System.err.println(end - start);
+        System.err.println(N * kRequests * 1000L / (end - start));
     }
 
 }

protorpc/muduo/rpc/RpcClient.java

@@ -26,6 +26,11 @@ public RpcClient() {
         bootstrap.setPipelineFactory(new RpcChannelPiplineFactory(this));
     }
 
+    public RpcClient(ChannelFactory channelFactory) {
+        bootstrap = new ClientBootstrap(channelFactory);
+        bootstrap.setPipelineFactory(new RpcChannelPiplineFactory(this));
+    }
+
     public RpcChannel blockingConnect(SocketAddress addr) {
         final CountDownLatch latch = new CountDownLatch(1);
         startConnect(addr, new NewChannelCallback() {