java socket的理解-从helo world开始

之前我以为java的socket很简单，不就是创建一个socketServet，然后不断的accept么？这种代码网上很多，基本流程是这样的：

ServerSocket server = new ServerSocket(8080);
while (true) {
	Socket socket = server.accept();
	InputStream input = socket.getInputStream();
	int length = input.available();
	if(length>0){
		byte[] data = new byte[length];
		input.read(data);
		String str = new String(data);
		String result = execute(str); 
		OutputStream out = socket.getOutputStream();
		out.write(("ok"+result+"\r\n").getBytes());
		out.flush();
	}
       socket.close();
 }

虽然在做轮询，但accept()会阻塞，直到有新的客户端请求过来才会产生一个新的socket，然后读取数据，处理数据，最后关闭。基本流程是 accept()--->read()--->hand()--->close()。

 

代码看上去很简单，也能成功运行，但这段代码存在2个问题：

1. 一个socket只能处理一次就被关闭了，对于局域网的RPC来说这是很不划算的，很多时候我们希望能是长链接。处理流程变成accept()-->read()-->hand()-->read()-->hand()-->close()

2. 如果一个client在一个socket中连续发送两次数据，client端代码会报错，因为socket在处理玩第一次read数据之后就被server关闭了。

 

所以为了解决上面两个问题，我把这段代码修改一下，每个socket一个线程来处理：

  class Channel implements Runnable {

        private Socket socket;
        private ServiceRegisterTester t ;
        public Channel(Socket socket,ServiceRegisterTester t ){
            super();
            this.socket = socket;
            this.t = t;
        }

        public void run() {
            try{
                Socket socket = this.socket;
                while(true){
                    InputStream input = socket.getInputStream();
                    int length = input.available();
                    if(length>0){
                        byte[] data = new byte[length];
                        input.read(data);
                        String str = new String(data);
                        String result = t.test(str);
                        OutputStream out = socket.getOutputStream();
                        out.write(("ok"+result+"\r\n").getBytes());
                        out.flush();
                    }
                }
            }catch(Exception e){
                e.printStackTrace();
            }
        }
    }

ServerSocket server = new ServerSocket(8080);
while (true) {
    Socket socket = server.accept();
    Channel chanenl = new Channel(socket,t);
    new Thread(chanenl).start();
}

 

 这样每个socket都有一个单独的线程来处理，前面2个问题解决了！

但通过top观测发现java进程cpu占用非常高，都在做空转，每个socket都有线程在做轮询，这太伤害性能了吧，怎么才能让

socket.getInputStream();

也阻塞，让他能智能的读取到可用的数据才唤醒呢？

 

 

我相信netty也会遇到这个问题看看netty是如何解决的吧：

        ChannelFactory factory =  
                new OioServerSocketChannelFactory (  
                        Executors.newCachedThreadPool(),  
                        Executors.newCachedThreadPool());  
        
        ServerBootstrap bootstrap = new ServerBootstrap (factory);  
        DiscardServerHandler handler = new DiscardServerHandler();
        ChannelPipeline pipeline = bootstrap.getPipeline();  
        pipeline.addLast("handler", handler);  
        bootstrap.setOption("child.tcpNoDelay", true);  
        bootstrap.setOption("child.keepAlive", true);  
        bootstrap.bind(new InetSocketAddress(8080));  

 ChannelFactory需要两个线程池，一个是boss，一个是worker，boss只需要一个线程即可，worker可以根据合适的情况配置。当在执行 bootstrap.bind()的时候会启动boss线程，代码如下：

class OioServerSocketPipelineSink{

	private void bind(
		        OioServerSocketChannel channel, ChannelFuture future,
		        SocketAddress localAddress) {


	 Executor bossExecutor =
		            ((OioServerSocketChannelFactory) channel.getFactory()).bossExecutor;
		        bossExecutor.execute(
		                new IoWorkerRunnable(
		                        new ThreadRenamingRunnable(
		                                new Boss(channel),
		                                "Old I/O server boss (channelId: " +
		                                channel.getId() + ", " + localAddress + ')')));
		        bossStarted = true;
	}
}

 

OioServerSocketPipelineSink&Boss 是一个Runnable,其run方法如下：

 

 while (channel.isBound()) {
        try {
            Socket acceptedSocket = channel.socket.accept();
            try {
                ChannelPipeline pipeline =
                    channel.getConfig().getPipelineFactory().getPipeline();
                final OioAcceptedSocketChannel acceptedChannel =
                    new OioAcceptedSocketChannel(
                            channel,
                            channel.getFactory(),
                            pipeline,
                            OioServerSocketPipelineSink.this,
                            acceptedSocket);
                workerExecutor.execute(
                        new IoWorkerRunnable(
                                new ThreadRenamingRunnable(
                                        new OioWorker(acceptedChannel),
                                        "Old I/O server worker (parentId: " +
                                        channel.getId() + ", channelId: " +
                                        acceptedChannel.getId() + ", " +
                                        channel.getRemoteAddress() + " => " +
                                        channel.getLocalAddress() + ')')));
            } catch (Exception e) {
                logger.warn(
                        "Failed to initialize an accepted socket.", e);
                try {
                    acceptedSocket.close();
                } catch (IOException e2) {
                    logger.warn(
                            "Failed to close a partially accepted socket.",
                            e2);
                }
            }
  }

 可以看到Boss的职责就是轮询获取每个新请求Socket，立即交给workerExecutor处理，workerExecutor的逻辑单元封装在OioWorker，OioWorker的run方法如下：

public void run(){
        final PushbackInputStream in = channel.getInputStream();
        while (channel.isOpen()) {
            synchronized (channel.interestOpsLock) {
                while (!channel.isReadable()) {
                    try {
                        // notify() is not called at all.
                        // close() and setInterestOps() calls Thread.interrupt()
                        channel.interestOpsLock.wait();
                    } catch (InterruptedException e) {
                        if (!channel.isOpen()) {
                            break;
                        }
                    }
                }
            }

            byte[] buf;
            int readBytes;
            try {
                int bytesToRead = in.available();
                if (bytesToRead > 0) {
                    buf = new byte[bytesToRead];
                    readBytes = in.read(buf);
                } else {
                    int b = in.read();
                    if (b < 0) {
                        break;
                    }
                    in.unread(b);
                    continue;
                }
            } catch (Throwable t) {
                if (!channel.socket.isClosed()) {
                    fireExceptionCaught(channel, t);
                }
                break;
            }

            ChannelBuffer buffer;
            if (readBytes == buf.length) {
                buffer = ChannelBuffers.wrappedBuffer(buf);
            } else {
                // A rare case, but it sometimes happen.
                buffer = ChannelBuffers.wrappedBuffer(buf, 0, readBytes);
            }

            fireMessageReceived(channel, buffer);
        }

        // Setting the workerThread to null will prevent any channel
        // operations from interrupting this thread from now on.
        channel.workerThread = null;

        // Clean up.
        close(channel, succeededFuture(channel));

}

 OioWorker里有一个非常重要的InputStream-PushbackInputStream，这个输入流能阻塞io,请看其read()方法的注释：“从此输入流中读取下一个数据字节。返回 0 到 255 范围内的 int 字节值。如果因流的末尾已到达而没有可用的字节，则返回值 -1。在输入数据可用、检测到流的末尾或者抛出异常前，此方法一直阻塞。”
netty就是利用这种方式来做轮询，而CPU又不至于空转。