现在的位置: 首页 > 综合 > 正文

xSocket multiplexed介绍

2013年12月12日 ⁄ 综合 ⁄ 共 4839字 ⁄ 字号 评论关闭
 xSocket是一个轻量级的基于nio的服务器框架用于开发高性能、可扩展、多线程的服务器。该框架封装了线程处理、异步读/写等方面。 

相关socket介绍请看 
xsocket documentation 

xsocket Multiplexed是xSocket的一个扩展模块,xsocket Multiplexed的定义是支持一个物理TCP连接之上运行的多个逻辑连接。 
说白了其实就是长连接的利用,所谓长连接就是Client方与Server方先建立通讯连接,连接建立后不断开, 然后再进行报文发送和接收。对于点对点的通讯,比较适合。使用xsocket Multiplexed进行持久连接通讯,对业务开发更加友好。 

下面是一个同步客户端的例子: 

Java代码 复制代码 收藏代码
  1. // 建立一个客户端的 multiplexed connection,   
  2. //这个连接里面套了一个NonBlockingConnection,后面所有的逻辑连接都通过这个连接传输数据   
  3. INonBlockingConnection nativeCon = NonBlockingConnection(hostname, 9011); 
      
  4. IMultiplexedConnection multiplexedCon = new MultiplexedConnection(nativeCon);  
      
  5.  // 建立一个逻辑连接 pipeline,然后所做的事情就是,向服务端写了一个header,默认实现是写了18个字节。   
  6. //VERSION byte(1) ;  PIPELINE_OPENED byte(1) ; UUID(16)   
  7.  //服务端收到header后就会调用注册handler的onPipelineOpend方法    
  8. String controlPipelineId = multiplexedCon.createPipeline();    
  9. IBlockingPipeline controlPipeline = multiplexedCon.getBlockingPipeline(controlPipelineId);   
  10.   
  11.   
  12. // 在发送数据时,会再向服务端写一个header    
  13. //VERSION byte(1) ;  PIPELINE_DATA byte(1) ; UUID(16)    
  14. //告诉服务端,我要写正文了,服务端会调用注册handler的onPipelineData方法,开始准备接收数据处理业务逻辑    
  15. //在处理过程中建立新的逻辑连接进行处理,多个逻辑连接之间不会有影响    
  16.   
  17. controlPipeline.write("NEW_MESSAGE_PIPELINE\r\n"); 
      
  18. String messagePipelineId = commandPipeline.readStringByDelimiter("\r\n"); 
      
  19. IBlockingPipeline messagePipeline = multiplexedCon.getBlockingPipeline(messagePipelineId);   
  20. messagePipeline.write(messsagepart_1);    
  21. messagePipeline.write(messsagepart_2); ...     
  22. controlPipeline.write("CLOSE_MESSAGE_PIPELINE" + messagePipelineId + "\r\n"); 
      
  23.   
  24. //最后关闭业务连接,向服务端发送一个header    
  25. //VERSION byte(1) ;  PIPELINE_CLOSED byte(1) ; UUID(16)   
  26. //服务端收到后就会调用注册handler的onPipelineClosed方法进行后续清理   
  27. //这个关闭仅仅是逻辑链接的关闭,并非主连接的关闭。   
  28. messagePipeline.close();   
[java] view
plain
copy

  1. // 建立一个客户端的 multiplexed connection,  
  2. //这个连接里面套了一个NonBlockingConnection,后面所有的逻辑连接都通过这个连接传输数据  
  3. INonBlockingConnection nativeCon = NonBlockingConnection(hostname, 9011);   
  4. IMultiplexedConnection multiplexedCon = new MultiplexedConnection(nativeCon);    
  5.  // 建立一个逻辑连接 pipeline,然后所做的事情就是,向服务端写了一个header,默认实现是写了18个字节。   
  6. //VERSION byte(1) ;  PIPELINE_OPENED byte(1) ; UUID(16)  
  7.  //服务端收到header后就会调用注册handler的onPipelineOpend方法   
  8. String controlPipelineId = multiplexedCon.createPipeline();   
  9. IBlockingPipeline controlPipeline = multiplexedCon.getBlockingPipeline(controlPipelineId);  
  10.   
  11.   
  12. // 在发送数据时,会再向服务端写一个header   
  13. //VERSION byte(1) ;  PIPELINE_DATA byte(1) ; UUID(16)   
  14. //告诉服务端,我要写正文了,服务端会调用注册handler的onPipelineData方法,开始准备接收数据处理业务逻辑   
  15. //在处理过程中建立新的逻辑连接进行处理,多个逻辑连接之间不会有影响   
  16.   
  17. controlPipeline.write("NEW_MESSAGE_PIPELINE\r\n");   
  18. String messagePipelineId = commandPipeline.readStringByDelimiter("\r\n");   
  19. IBlockingPipeline messagePipeline = multiplexedCon.getBlockingPipeline(messagePipelineId);  
  20. messagePipeline.write(messsagepart_1);   
  21. messagePipeline.write(messsagepart_2); ...    
  22. controlPipeline.write("CLOSE_MESSAGE_PIPELINE" + messagePipelineId + "\r\n");   
  23.   
  24. //最后关闭业务连接,向服务端发送一个header   
  25. //VERSION byte(1) ;  PIPELINE_CLOSED byte(1) ; UUID(16)  
  26. //服务端收到后就会调用注册handler的onPipelineClosed方法进行后续清理  
  27. //这个关闭仅仅是逻辑链接的关闭,并非主连接的关闭。  
  28. messagePipeline.close();   

服务端的处理 

对于一个新的IMultiplexedConnection服务端会调度到线程池执行,而对于IMultiplexedConnection的逻辑链接默认是在IMultiplexedConnection所在的线程中执行的,你也可以将逻辑链接中的业务处理自行调度到线程池中去执行。 

Java代码 复制代码 收藏代码
  1. IServer server = new Server(9011new MultiplexedProtocolAdapter(new CommandPipelineHandler()));
      
  2. ConnectionUtils.start(server);   
  3.     
  4.     
  5. // pipeline data handler which is assigned to each new pipeline   
  6. class CommandPipelineHandler implements IPipelineDataHandler {
      
  7.     
  8.     public boolean onData(INonBlockingPipeline pipeline) throws IOException {
      
  9.         String cmd = pipeline.readStringByDelimiter("\r\n")
      
  10.     
  11.         if (cmd.equals("NEW_MESSAGE_PIPELINE")) {
      
  12.             IMultiplexedConnection mplCon = pipeline.getMultiplexedConnection();   
  13.     
  14.             //这里定义了一个新的逻辑连接,并不和其他逻辑连接冲突   
  15.             String msgPipelineId = mplCon.createPipeline();   
  16.             INonBlockingPipeline msgPipeline = mplCon.getNonBlockingPipeline(msgPipelineId);   
  17.     
  18.             // replace the CommandPipelineHandler of the new pipeline   
  19.             // 为新的逻辑连接指派一个handler,   
  20.             // 和当前handler在一个线程中,即使设定@Execution(Execution.MULTITHREADED) 也是没用的   
  21.             msgPipeline.setHandler(new DataHandler());
      
  22.             pipeline.write(msgPipelineId + "\r\n");
      
  23.   
  24.             //耗时业务处理可以自行进行线程池调度,比如   
  25.             pipeline.getWorkerpool().execute(task..);   
  26.     
  27.                
  28.     
  29.         } ...   
  30.         ...   
  31.     }   
  32. }   
  33.     
  34. // A pipeline handler could also be a ordinary handler like IDataHandler   
  35. class DataHandler implements IDataHandler {
      
  36.     
  37.     private FileChannel fc = null;
      
  38.     
  39.     DataHandler() {   
  40.         File file = ...   
  41.         fc = new RandomAccessFile(file, "rw").getChannel();
      
  42.         ...   
  43.     }   
  44.     
  45.     
  46.     public boolean onData(INonBlockingConnection pipeline) throws IOException {
      
  47.         pipeline.transferTo(fc, pipeline.available());   
  48.         ...   

抱歉!评论已关闭.