xSocket multiplexed介绍

ゞ浴缸里的玫瑰 2021-09-15 06:14 289阅读 0赞

xSocket是一个轻量级的基于nio的服务器框架用于开发高性能、可扩展、多线程的服务器。该框架封装了线程处理、异步读/写等方面。

相关socket介绍请看   
[xsocket documentation][]   
  
xsocket Multiplexed是xSocket的一个扩展模块，xsocket Multiplexed的定义是支持一个物理TCP连接之上运行的多个逻辑连接。   
说白了其实就是长连接的利用，所谓长连接就是Client方与Server方先建立通讯连接，连接建立后不断开， 然后再进行报文发送和接收。对于点对点的通讯，比较适合。使用xsocket Multiplexed进行持久连接通讯，对业务开发更加友好。   
  
  
![MultiplexedConnection.gif][]  
  
下面是一个同步客户端的例子：

Java代码  [![复制代码][icon_copy.gif]][icon_copy.gif 1]  [![收藏代码][icon_star.png]![spinner.gif][]][icon_star.png_spinner.gif]

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][view plain]

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所在的线程中执行的，你也可以将逻辑链接中的业务处理自行调度到线程池中去执行。   
  
  
![PipelineHandler.gif][]

Java代码  [![复制代码][icon_copy.gif]][icon_copy.gif 1]  [![收藏代码][icon_star.png]![spinner.gif][]][icon_star.png_spinner.gif]

1.  IServer server = **new** Server(9011, **new** 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.         ...   
49.         **return** **true**;   
50.     \}   
51. \}

**\[java\]**  [view plain][] [copy][view plain]

1.  IServer server = **new** Server(9011, **new** 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.         ...  
49.         **return** **true**;  
50.     \}  
51. \}

Multiplexer的实现   
  
在Multiplexer可以定义发送header及如何读取数据的的各项细节，参见默认的实现可以了解的更多，基本上一目了然了。   
  
默认实现是org.xsocket.connection.multiplexed.multiplexer.SimpleMultiplexer   
  
  
  
  
你可以自定义个Multiplexer：

Java代码  [![复制代码][icon_copy.gif]][icon_copy.gif 1]  [![收藏代码][icon_star.png]![spinner.gif][]][icon_star.png_spinner.gif]

1.  // client-side custom multiplexed handling   
2.  IMultiplexedConnection multiplexedCon = **new** MultiplexedConnection(nativeCon, **new** MyMultiplexer());   
3.      
4.  // server-side custom multiplexed handling   
5.  IHandler adpater = **new** MultiplexedProtocolAdapter(**new** CommandPipelineHandler(), **new** MyMultiplexer()))   
6.  IServer server = **new** Server(9011, adpater);

**\[java\]**  [view plain][] [copy][view plain]

1.  // client-side custom multiplexed handling  
2.  IMultiplexedConnection multiplexedCon = **new** MultiplexedConnection(nativeCon, **new** MyMultiplexer());  
3.     
4.  // server-side custom multiplexed handling  
5.  IHandler adpater = **new** MultiplexedProtocolAdapter(**new** CommandPipelineHandler(), **new** MyMultiplexer()))  
6.  IServer server = **new** Server(9011, adpater);

Java代码  [![复制代码][icon_copy.gif]][icon_copy.gif 1]  [![收藏代码][icon_star.png]![spinner.gif][]][icon_star.png_spinner.gif]

1.  **class** MyMultiplexer **implements** IMultiplexer \{   
2.      
3.    
4.      //打开一个新的逻辑连接，向对端发送一个header   
5.      **public** String openPipeline(INonBlockingConnection nbc)    
6.               **throws** IOException,    
7.               ClosedException \{   
8.    
9.          ... create a unique pipelineId   
10.         ... create a network packet   
11.         ... and send it to the peer   
12.     
13.         **return** pipelineId;   
14.     \}   
15.     
16.     //关闭逻辑连接，向对端发送一个header   
17.     **public** **void** closePipeline(INonBlockingConnection nbc, String pipelineId)   
18.              **throws** IOException,    
19.              ClosedException \{   
20.   
21.         ... create a network packet   
22.         ... and send it to the peer   
23.     \}   
24.     
25.     //发送正文的之前执行，想对端发送一个header      
26.     **public** **void** multiplex(INonBlockingConnection nbc, String pipelineId, ByteBuffer\[\] dataToWrite)   
27.              **throws** IOException,    
28.              ClosedException \{   
29.   
30.         ... create a network packet based on the pipeline data   
31.         ... and send it to the peer   
32.     \}   
33.   
34.       
35.     **public** **void** multiplex(INonBlockingConnection nbc, String pipelineId, ByteBuffer\[\] dataToWrite, IWriteCompletionHandler completionHandler)   
36.              **throws** IOException,    
37.              ClosedChannelException \{   
38.   
39.         ... create a network packet based on the pipeline data   
40.         ... and send it to the peer   
41.    \}   
42.   
43.     
44.      //这个方法定义数据的接收处理   
45.     **public** **void** demultiplex(INonBlockingConnection nbc, IDemultiplexResultHandler rsltHdl)    
46.              **throws** IOException,    
47.              ClosedException \{   
48.   
49.         ... read the network packet   
50.         ... notify the specific pipeline message   
51.   
52.             //默认实现是：   
53.            **byte** dataType = nbc.readByte();   
54.            ...   
55.         **switch** (dataType) \{   
56.             **case** PIPELINE\_DATA:...rsltHdl.onPipelineData(pipelineId, data);   
57.             **case** PIPELINE\_OPENED:...rsltHdl.onPipelineOpend(pipelineId, data);   
58.             **case** PIPELINE\_CLOSED:...rsltHdl.onPipelineClosed(pipelineId, data);   
59.                         **default**:...   
60.                 \}   
61.            ...   
62.               
63.             
64.     \}   
65. \}

**\[java\]**  [view plain][] [copy][view plain]

1.  **class** MyMultiplexer **implements** IMultiplexer \{  
2.     
3.    
4.      //打开一个新的逻辑连接，向对端发送一个header  
5.      **public** String openPipeline(INonBlockingConnection nbc)   
6.               **throws** IOException,   
7.               ClosedException \{  
8.    
9.          ... create a unique pipelineId  
10.         ... create a network packet  
11.         ... and send it to the peer  
12.    
13.         **return** pipelineId;  
14.     \}  
15.    
16.     //关闭逻辑连接，向对端发送一个header  
17.     **public** **void** closePipeline(INonBlockingConnection nbc, String pipelineId)  
18.              **throws** IOException,   
19.              ClosedException \{  
20.   
21.         ... create a network packet  
22.         ... and send it to the peer  
23.     \}  
24.    
25.     //发送正文的之前执行，想对端发送一个header     
26.     **public** **void** multiplex(INonBlockingConnection nbc, String pipelineId, ByteBuffer\[\] dataToWrite)  
27.              **throws** IOException,   
28.              ClosedException \{  
29.   
30.         ... create a network packet based on the pipeline data  
31.         ... and send it to the peer  
32.     \}  
33.   
34.      
35.     **public** **void** multiplex(INonBlockingConnection nbc, String pipelineId, ByteBuffer\[\] dataToWrite, IWriteCompletionHandler completionHandler)  
36.              **throws** IOException,   
37.              ClosedChannelException \{  
38.   
39.         ... create a network packet based on the pipeline data  
40.         ... and send it to the peer  
41.    \}  
42.   
43.    
44.      //这个方法定义数据的接收处理  
45.     **public** **void** demultiplex(INonBlockingConnection nbc, IDemultiplexResultHandler rsltHdl)   
46.              **throws** IOException,   
47.              ClosedException \{  
48.   
49.         ... read the network packet  
50.         ... notify the specific pipeline message  
51.   
52.             //默认实现是：  
53.            **byte** dataType = nbc.readByte();  
54.            ...  
55.         **switch** (dataType) \{  
56.             **case** PIPELINE\_DATA:...rsltHdl.onPipelineData(pipelineId, data);  
57.             **case** PIPELINE\_OPENED:...rsltHdl.onPipelineOpend(pipelineId, data);  
58.             **case** PIPELINE\_CLOSED:...rsltHdl.onPipelineClosed(pipelineId, data);  
59.                         **default**:...  
60.                 \}  
61.            ...  
62.              
63.            
64.     \}  
65. \}

总的来说这个api还是非常轻巧实用的。

[xsocket documentation]: http://xsocket.sourceforge.net/documentation.htm
[MultiplexedConnection.gif]: /images/20210726/3b6697a2c95d476db6e1575624fd5b58.png
[icon_copy.gif]: http://www.iteye.com/images/icon_copy.gif
[icon_copy.gif 1]: http://www.iteye.com/topic/784857#
[icon_star.png]: http://www.iteye.com/images/icon_star.png
[spinner.gif]: /images/20210726/ea1024cc97ff474a8aad21ee9bc719b3.png
[icon_star.png_spinner.gif]: http://blog.csdn.net/lizhi200404520/article/details/6700532
[view plain]: http://blog.csdn.net/lizhi200404520/article/details/6700532#
[PipelineHandler.gif]: /images/20210726/1c92606a1306431297e599647930f54c.png