NIO socket 的简单连接池

      在最近的项目中，需要写一个socket 与 底层服务器通信的模块。在设计中，请求对象被封装 xxxRequest,消息返回被封装为 xxxResponse. 由于socket的编程开发经验少，一开始我使用了短连接的方式，每个请求建立一个socket通信，由于每个socket只进行一次读写，这大大浪费了系统资源。

      于是考虑使用长连接，系统公用一个client socket 并对send 操作进行加锁，结果在处理并发的时候，各种慢，各种等待。没有办法，考虑使用两节池，预先创建多个 client socket 放入 连接池，需要发送请求时从连接池获取一个socket，完成请求时放入连接池中。下面是一个简单的实现。

       

        private  static String IP=GlobalNames.industryIP;
 private  static int PORT =Integer.parseInt(GlobalNames.industryPort);
 
 private static  int CONNECTION_POOL_SIZE = 10;
 private static NIOConnectionPool self = null;
 private Hashtable<Integer, SocketChannel> socketPool = null; // 连接池
 private boolean[] socketStatusArray = null; // 连接的状态(true-被占用,false-空闲)
 private static Selector selector  = null;
 private static InetSocketAddress SERVER_ADDRESS = null;
 
 /**
  * 初始化连接池，最大TCP连接的数量为10
  *
  * @throws IOException
  */
 public static synchronized void init() throws Exception {
  self = new NIOConnectionPool();
  self.socketPool = new Hashtable<Integer, SocketChannel>();
  self.socketStatusArray = new boolean[CONNECTION_POOL_SIZE];
  buildConnectionPool();
 }

 /**
  * 建立连接池
  */
 public synchronized static void buildConnectionPool() throws Exception {
  if (self == null) {
   init();
  }
  for (int i = 0; i < CONNECTION_POOL_SIZE; i++) {
   SocketChannel client = allocateSocketChannel();
   self.socketPool.put(new Integer(i), client);
   self.socketStatusArray[i] = false;
  }
 }

 /**
  * 从连接池中获取一个空闲的Socket
  *
  * @return 获取的TCP连接
  */
 public static SocketChannel getConnection() throws Exception {
  if (self == null)
   init();
  int i = 0;
  for (i = 0; i < CONNECTION_POOL_SIZE; i++) {
   if (!self.socketStatusArray[i]) {
    self.socketStatusArray[i] = true;
    break;
   }
  }
  if (i < CONNECTION_POOL_SIZE) {
   return self.socketPool.get(new Integer(i));
   
  } else {

  //目前连接池无可用连接时只是简单的新建一个连接
   SocketChannel newClient = allocateSocketChannel();
   CONNECTION_POOL_SIZE++;
   self.socketPool.put(CONNECTION_POOL_SIZE, newClient);
   return newClient;
  }
 }

 /**
  * 当获得的socket不可用时，重新获得一个空闲的socket。
  *
  * @param socket
  *            不可用的socket
  * @return 新得到的socket
  * @throws Exception
  */
 public static SocketChannel rebuildConnection(SocketChannel socket)
   throws Exception {
  if (self == null) {
   init();
  }
  SocketChannel newSocket = null;
  try {
   for (int i = 0; i < CONNECTION_POOL_SIZE; i++) {
    if (self.socketPool.get(new Integer(i)) == socket) {
     newSocket = allocateSocketChannel();
     self.socketPool.put(new Integer(i), newSocket);
     self.socketStatusArray[i] = true;
    }
   }

  } catch (Exception e) {
   System.out.println("重建连接失败!");
   throw new RuntimeException(e);
  }
  return newSocket;
 }

 /**
  * 将用完的socket放回池中，调整为空闲状态。此时连接并没有断开。
  *
  * @param socket
  *            使用完的socket
  * @throws Exception
  */
 public static void releaseConnection(SocketChannel socket) throws Exception {
  if (self == null) {
   init();
  }
  for (int i = 0; i < CONNECTION_POOL_SIZE; i++) {
   if (self.socketPool.get(new Integer(i)) == socket) {
    self.socketStatusArray[i] = false;
    break;
   }
  }
 }

 /**
  * 断开池中所有连接
  *
  * @throws Exception
  */
 public synchronized static void releaseAllConnection() throws Exception {
  if (self == null)
   init();

  // 关闭所有连接
  SocketChannel socket = null;
  for (int i = 0; i < CONNECTION_POOL_SIZE; i++) {
   socket = self.socketPool.get(new Integer(i));
   try {
    socket.close();
    self.socketStatusArray[i] = false;
   } catch (Exception e) {
    e.printStackTrace();
   }
  }
 }
   
 
 public static SocketChannel allocateSocketChannel(){
  
   SERVER_ADDRESS = new InetSocketAddress(  
                IP, PORT);  
  SocketChannel socketChannel = null;
     SocketChannel client = null;
     try{
     socketChannel = SocketChannel.open();  
        socketChannel.configureBlocking(false);  
        selector = Selector.open();  
        socketChannel.register(selector, SelectionKey.OP_CONNECT);  
        socketChannel.connect(SERVER_ADDRESS);
        Set<SelectionKey> selectionKeys;  
        Iterator<SelectionKey> iterator;  
        SelectionKey selectionKey;
        selector.select();  
        selectionKeys = selector.selectedKeys();  
        iterator = selectionKeys.iterator();  
        while (iterator.hasNext()) {  
            selectionKey = iterator.next();  
            if (selectionKey.isConnectable()) {  
                client = (SocketChannel) selectionKey.channel();  
                if (client.isConnectionPending()) {  
                    client.finishConnect();
                    client.register(selector, SelectionKey.OP_WRITE);  
                    break;
                }
            }
        }
 }catch(Exception e){
  e.printStackTrace();
 }
 return client;
  }

 public static Selector getSelector() {
  return selector;
 }

 

使用连接池进行通信：

 /*缓冲区大小*/ 
     private static int BLOCK = 8*4096;  
     /*发送数据缓冲区*/ 
     private static ByteBuffer sendbuffer = ByteBuffer.allocate(BLOCK);  
   
     /*接受数据缓冲区*/
     private static ByteBuffer protocalNum = ByteBuffer.allocate(4);
     private static ByteBuffer functionNum = ByteBuffer.allocate(4);
     private static ByteBuffer messageLen = ByteBuffer.allocate(4);
     private static ByteBuffer receivebuffer = null;
     private  SocketChannel client = null;
     private Selector selector = null;
    
     private boolean readable = true;
     private boolean writable = true;
    
    
     public NIOSocketBackUp() throws Exception{
      client = NIOConnectionPool.getConnection();
      selector = NIOConnectionPool.getSelector();
     }
    
     public String send(ServiceRequest request) throws Exception {   
             
         Set<SelectionKey> selectionKeys;  
         Iterator<SelectionKey> iterator;  
         SelectionKey selectionKey;  
         int count=0;  
         boolean flag = true;
         String receiveText="";  
          while (flag) {  
             selector.select();  
             //返回此选择器的已选择键集。  
             selectionKeys = selector.selectedKeys();  
             iterator = selectionKeys.iterator();  
             while (iterator.hasNext()) {  
                 selectionKey = iterator.next();  
                 if (selectionKey.isWritable() && (writable)) {  
                         sendbuffer.clear();  
                         sendbuffer.put(request.getProtocalNum());
                         sendbuffer.put(request.getFunctionNum());
                         sendbuffer.put(request.getMessageLen());
                         sendbuffer.put(request.getXmlbytes());
                         sendbuffer.flip();  
                         client.write(sendbuffer);  
                         client.register(selector, SelectionKey.OP_READ);  
                         writable = false;
                 } else if (selectionKey.isReadable() && (readable) ) {  
                     protocalNum.clear();
                     functionNum.clear();
                     messageLen.clear();
                    
                    
                     count=client.read(protocalNum);  
                     count=client.read(functionNum);
                     count=client.read(messageLen);
                     messageLen.rewind();
                     int length = messageLen.asIntBuffer().get(0);
                     receivebuffer = ByteBuffer.allocate(length-12);
                     receivebuffer.clear();
                    
                     //read方式竟然不阻塞
                     int total=0;
                     while(total!=(length-12)){
                       count=client.read(receivebuffer);
                       total+=count;
                     }
                     client.register(selector, SelectionKey.OP_WRITE);  
                     receiveText = new String(receivebuffer.array(),"GBK");
                     flag = false;
                     readable = false;
                     break;
                 }
             }  
         }

      
      NIOConnectionPool.releaseConnection(client);
         return receiveText.trim();
     }  

 

 

 

 

评论

1 楼 zhaoqiubo 2015-08-03  

我理解长连接就是链路要维持着，怎么能一次写入就关闭了呢？你可以再次写入啊，搞不懂为啥要用连接池。望不吝赐教。