几种NIO程序写法（一）kafka的SocketServer

女爷i 2022-01-16 05:29 205阅读 0赞

[2019独角兽企业重金招聘Python工程师标准>>> ][2019_Python_] ![hot3.png][]

总结一点常见的NIO程序的写法，虽然没有一个固定的格式，也没有特别大的差别，但是多总结总结各位大师的写法，多见点儿组合还是对自身代码的质量有很大提高的。这一篇我想通过对kafka network包下的源码进行分析，然后抽取第一种比较常见的写法。之前已经写过一篇关于kafka通信源码的解读，可参见[http://my.oschina.net/ielts0909/blog/102336][http_my.oschina.net_ielts0909_blog_102336]，最近再改写一些源码，所以还是有必要把更完整的认识更新上来。

当然这不是最常见的写法，最常见，也是大多数人都能接受的代码可以参考[http://my.oschina.net/ielts0909/blog/89849][http_my.oschina.net_ielts0909_blog_89849]文中的代码，当然在read或者write的时，更多的会采用多线程来处理。

各种写法的本质都是一样的，可能我们要更注重比较不同细节的写法，kafkaSocketServer的代码主要在kafka broker端使用，是最基础的服务端通信类。在kafka代码中，与常见写法区别比较大的就是将选择器键的接收部分（acceptor）单独拿出来写了。再通过接收器（acceptor）将读写的任务交付给更多个处理器（processor）。在处理器中采用队列的形式按先进先出的顺序对读写进行操作。

![220055_X4JH_589742.png][]

我特意将图画的跟reactor模式的图差不多的形式，其实两者的本质也没什么差别，主要看怎么去实现acceptor与多线程处理读写操作。

![220129_3rA3_589742.png][]

我用java改写了部分kafka的代码，这样便于阅读，我们看AbstractServerThread中的一些细节，这个类主要定义了startup、shutdown相关的一系列方法，这些方法通过闭锁来同步多线程执行的顺序。这个类直接向子类提供了selector对象。

package org.gfg.inforq.network;
    
    import java.io.IOException;
    import java.nio.channels.Selector;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.atomic.AtomicBoolean;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    /**
     * A base class for server thread with several variables and methods
     * 
     * @author Chen.Hui
     * @since 0.1
     */
    public abstract class AbstractServerThread implements Runnable {
    
    	protected Selector selector;
    	private static final Logger LOG = LoggerFactory
    			.getLogger(AbstractServerThread.class);
    	private CountDownLatch startupLatch = new CountDownLatch(1);
    	private CountDownLatch shutdownLatch = new CountDownLatch(1);
    	private AtomicBoolean alive = new AtomicBoolean(false);
    
    	protected AbstractServerThread() throws IOException {
    		this.selector = Selector.open();
    		LOG.info("selector is opening");
    	}
    
    	/**
    	 * shutdown the running therad
    	 * @throws InterruptedException
    	 * 
    	 */
    	protected void shutdown() throws InterruptedException {
    		alive.set(false);
    		selector.wakeup();
    		shutdownLatch.await();
    	}
    
    	/**
    	 * Causes the current thread to wait until the latch has counted down to
    	 * zero, unless the thread is interrupted.
    	 * @throws InterruptedException
    	 */
    	protected void awaitStartup() throws InterruptedException {
    		startupLatch.await();
    	}
    
    	/**
    	 *  releasing all waiting threads if the count reaches zero.
    	 */
    	protected void startupComplete() {
    		alive.set(true);
    		startupLatch.countDown();
    	}
    
    	/**
    	 * 
    	 */
    	protected void shutdownComplete() {
    		shutdownLatch.countDown();
    	}
    	
    	/**
    	 * 
    	 * @return true only this thread is startup complete
    	 */
    	protected boolean isRunning(){
    		return alive.get();
    	}
    	
    }

Acceptor的作用也仅仅就是accept。我们可以注意到Acceptor引用了一系列的Processor，然后为每个processor与一个key绑定。并将相应的通道传递给processor。所以总结起来acceptor做了两件事，绑定key和processor，并将通道传递给相应的processor。

package org.gfg.inforq.network;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.ServerSocketChannel;
    import java.nio.channels.SocketChannel;
    import java.util.Iterator;
    import java.util.Set;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    /**
     * 
     * @author Chen.Hui
     * 
     */
    public class Acceptor extends AbstractServerThread {
    
    	protected int port;
    	private Processor[] processors;
    	protected int sendBufferSize;
    	protected int receiveBufferSize;
    	
    	private static final Logger LOG=LoggerFactory.getLogger(Acceptor.class);
    	
    	
    	public Acceptor(int port, Processor[] processors, int sendBufferSize,
    			int receiveBufferSize) throws IOException {
    		//super();
    		this.port=port;
    		this.processors=processors;
    		this.sendBufferSize=sendBufferSize;
    		this.receiveBufferSize=receiveBufferSize;
    	}
    
    	public void run() {
    		try {
    			ServerSocketChannel serverChannel = ServerSocketChannel.open();
    			serverChannel.configureBlocking(false);
    			serverChannel.socket().bind(new InetSocketAddress(port));
    			serverChannel.register(selector, SelectionKey.OP_ACCEPT);
    			LOG.info("Awaiting connections on port "+port);
    			startupComplete();
    			
    			int currentProcessor=0;
    			while(isRunning()){
    				int ready=selector.select(500);
    				if(ready>0){
    				 Set<SelectionKey> keys=selector.selectedKeys();
    				 Iterator<SelectionKey> iter=keys.iterator();
    				 while(iter.hasNext()&&isRunning()){
    					 SelectionKey key=null;
    					 try{
    						 key=iter.next();
    						 iter.remove();
    						 if(key.isAcceptable()){
    							 accept(key,processors[currentProcessor]);
    						 }else{
    							 throw new IllegalStateException("Unrecognized key state for acceptor thread");
    						 }
    						 currentProcessor=(currentProcessor+1)%processors.length;
    					 }catch (Exception e) {
    						LOG.error(" ",e);
    					}
    				 }
    				}
    			}
    			serverChannel.close();
    			selector.close();
    			shutdownComplete();
    		} catch (IOException e) {
    			e.printStackTrace();
    		}
    	}
    
    	private void accept(SelectionKey key, Processor processor) throws IOException {
    		ServerSocketChannel ssc=(ServerSocketChannel) key.channel();
    		ssc.socket().setReceiveBufferSize(receiveBufferSize);
    		
    		SocketChannel socketChannel=ssc.accept();
    		socketChannel.configureBlocking(false);
    		socketChannel.socket().setTcpNoDelay(true);
    		socketChannel.socket().setSendBufferSize(sendBufferSize);
    		
    		if(LOG.isDebugEnabled()){
    			LOG.debug("sendBufferSize:["+socketChannel.socket().getSendBufferSize()+"] receiveBufferSize:["+socketChannel.socket().getReceiveBufferSize()+"]" );
    		}
    		processor.accept(socketChannel);
    	}
    }

Acceptor里也没做什么策略直接将任务按照currentProcessor=(currentProcessor+1)%processors.length;的形式分配，如果做的更好的话，可以按照任务的权重、执行时间等重新分配。另外要注意这里调用的一些AbstractServerThread里的方法。

Processor类中主要做的就是读写操作，这部分没有什么特别突出的不同，在类里引用了队列来存储由acceptor分发过来的通道，并按顺序处理。

package org.gfg.inforq.network;
    
    import java.io.IOException;
    import java.nio.channels.ClosedChannelException;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.SocketChannel;
    import java.util.Iterator;
    import java.util.Set;
    import java.util.concurrent.ConcurrentLinkedQueue;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    public class Processor extends AbstractServerThread {
    
    	private final int maxRequestSize;
    	
    	public Processor(int maxRequestSize) throws IOException {
    		super();
    		this.maxRequestSize=maxRequestSize;
    	}
    
    	private ConcurrentLinkedQueue<SocketChannel> newConnections = new ConcurrentLinkedQueue<SocketChannel>();
    	private static final Logger LOG = LoggerFactory.getLogger(Processor.class);
    
    	public void run() {
    		/** make sure the selector is open completely */
    		startupComplete();
    
    		while (isRunning()) {
    			try {
    				// setup any new connections that have been queued up
    				configureNewConnections();
    				int ready = selector.select();
    				if (ready > 0) {
    					Set<SelectionKey> keys = selector.selectedKeys();
    					Iterator<SelectionKey> iter = keys.iterator();
    					while (iter.hasNext() && isRunning()) {
    						SelectionKey key = null;
    						try {
    							key = iter.next();
    							iter.remove();
    
    							if (key.isReadable()) {
    								//
    								read(key);
    							} else if (key.isWritable()) {
    								//
    								write(key);
    							} else if (key.isValid()) {
    								//close
    								close(key);
    							} else {
    								throw new IllegalStateException("Unrecognized key state for processor thread");
    							}
    						} catch (Exception e) {
    							if (e instanceof IOException) {
    
    							}
    						}
    					}
    				}
    
    			} catch (Exception e) {
    				// TODO: handle exception
    			}
    		}
    	}
    
    	private void read(SelectionKey key) {
    		SocketChannel socketChannel=channelFor(key);
    		Receive request=(Receive) key.attachment();
    		
    		if(key.attachment()==null){
    			request=new BoundedByteBufferReceive(maxRequestSize);
    			key.attach(request);
    		}
    		
    		int read=request.readFrom(socketChannel);
    		
    		if(LOG.isTraceEnabled()){
    			LOG.trace(read+" bytes read from "+socketChannel.socket().getRemoteSocketAddress());
    		}
    		
    		if(read<0){
    			close(key);
    			return;
    		}else if(request.complete){
    			handle(key,request);
    		}
    		
    	}
    
    	private void handle(SelectionKey key, Receive request) {
    		// TODO Auto-generated method stub
    		
    	}
    
    	private void write(SelectionKey key) {	
    		Send response= (Send) key.attachment();
    		SocketChannel socketChannel=channelFor(key);
    		int written =response.writeTo(socketChannel);
    		
    		if(LOG.isTraceEnabled()){
    			LOG.trace(written+" bytes written to "+socketChannel.socket().getRemoteSocketAddress());
    		}
    		if(response.complete){
    			key.attach(null);
    			key.interestOps(SelectionKey.OP_READ);
    		}else{
    			key.interestOps(SelectionKey.OP_WRITE);
    			selector.wakeup();
    		}
    	}
    	
    	private SocketChannel channelFor(SelectionKey key){
    		return (SocketChannel) key.channel(); 
    	}
    
    	/**
    	 * close the channel
    	 * @param key
    	 */
    	private void close(SelectionKey key) {
    		SocketChannel channel=(SocketChannel) key.channel();
    		if(LOG.isDebugEnabled()){
    			LOG.debug("Closing the connection from"+channel.socket().getRemoteSocketAddress());
    		}
    		if(channel.isOpen()&&channel.socket().isConnected()){
    			try {
    				//care the sequence of closing
    				channel.socket().close();
    				channel.close();
    				key.attach(null);
    				key.cancel();
    			} catch (IOException e) {
    				LOG.info(e.getMessage(),e);
    			}
    		}
    	}
    
    	private void configureNewConnections() {
    		while (newConnections.size() > 0) {
    			SocketChannel channel = newConnections.poll();
    			if (LOG.isDebugEnabled()) {
    				LOG.debug("linstening to a new connection from "
    						+ channel.socket().getRemoteSocketAddress());
    			}
    			try {
    				channel.register(selector, SelectionKey.OP_READ);
    			} catch (ClosedChannelException e) {
    				LOG.info("the channel has been closed..");
    				continue;
    			}
    		}
    	}
    
    	public void accept(SocketChannel socketChannel) {
    		newConnections.add(socketChannel);
    		selector.wakeup();
    	}
    }

我也无法说出这种将接收器单独拿出来写的方式到底好处在哪里，但是这样的写法结构清晰，各个组件分工明确，还是很值得学习的。总结的目的不就是为了学习各种优秀的东西么。

转载于:https://my.oschina.net/ielts0909/blog/122182

[2019_Python_]: https://my.oschina.net/u/2663968/blog/3061697
[hot3.png]: /images/20220114/787371c6593d40f789ca701d8955fd89.png
[http_my.oschina.net_ielts0909_blog_102336]: http://my.oschina.net/ielts0909/blog/102336
[http_my.oschina.net_ielts0909_blog_89849]: http://my.oschina.net/ielts0909/blog/89849
[220055_X4JH_589742.png]: /images/20220114/eafc74d966e64d7798efb09cd4f49dd1.png
[220129_3rA3_589742.png]: /images/20220114/9bdbeb5d1fc04bb9912ba7aad6362a2e.png