《Apache MINA 2.0 用户指南》第九章：编解码器过滤器

比眉伴天荒 2023-10-16 21:19 8阅读 0赞

最近准备将Apache MINA 2.0 用户指南英文文档翻译给大家，但是我偶然一次百度，发现“[Defonds][]”这位大牛已经翻译大部分文档。

原文链接：[http://mina.apache.org/mina-project/userguide/ch11-ssl-filter/ch11-ssl-filter.html][http_mina.apache.org_mina-project_userguide_ch11-ssl-filter_ch11-ssl-filter.html]。

本文解释一下为什么以及如何使用一个 ProtocolCodecFilter。

为什么要使用一个 ProtocolCodecFilter？

*  TCP 担保以正确的顺序交付所有数据包。但是没有担保对于在发送端写操作时影响到接收端的读事件。参考[http://en.wikipedia.org/wiki/IPv4\#Fragmentation\_and\_reassembly][http_en.wikipedia.org_wiki_IPv4_Fragmentation_and_reassembly] 和 [http://en.wikipedia.org/wiki/Nagle%27s\_algorithm][http_en.wikipedia.org_wiki_Nagle_27s_algorithm]。在 MINA 术语中：如果没有 ProtocolCodecFilter 的话，一个由发送端对 IoSession.write(Object message) 的调用将导致多个接收端的 messageReceived(IoSession session, Object message) 事件，多个 IoSession.write(Object message) 的调用只会引起唯一的一个 messageReceived 事件。当你的客户端和服务器端运行在同一台主机上 (或者处于同一个本地网) 而你的应用能够应对这种情况时，你可能不会遭遇到这种情况。
 *  大多数网络应用需要一种方法以找出当前消息的结束位置和下一条消息的起始位置。
 *  你完全可以在你的 IoHandler 实现所有的这些逻辑，但是添加一个 ProtocolCodecFilter 可以让你的代码更加清晰并容易维护。
 *  它将你的协议逻辑从你的业务逻辑中 (IoHandler) 剥离开来。

如何使用？  
 你的应用接收到的基本上是一串字节，你需要将这些字节转换为消息 (高层对象)。  
 有三种常见的技术用来将字节流分割为消息：

*  使用固定长度的字节
 *  使用固定长度的报头来指示出报体的长度
 *  使用定界符。例如，许多基于文本的协议在每条消息 ([http://www.faqs.org/rfcs/rfc977.html][http_www.faqs.org_rfcs_rfc977.html]) 后面紧跟一个新的空行 (或者 CR LF 对)。

本文将使用第一种和第二种方法，因为它们很明显更容易实现。然后我们再看一下使用定界符的做法。

例子  
 我们将会开发一个 (毫无用处的) 图形字符发生器协议服务器来说明如何实现你自己协议的编解码器 (ProtocolEncoder、ProtocolDecoder 和 ProtocolCodecFactory)。这个协议是很简单的。这是请求消息的布局：

*  宽度：被请求图片的宽度 (网络字节顺序的一个整数)
 *  高度：被请求图片的高度 (网络字节顺序的一个整数)
 *  角色数量：要生成的字符数 (网络字节顺序的一个整数)

服务器以符合请求尺寸的两张图片进行响应，每张图片上打上请求的字符数。这是一个响应消息的布局：

<table style="color:rgb(0,0,0); font-family:Arial; font-size:14px; line-height:26px"> 
 <tbody> 
  <tr> 
   <th>4 个字节</th> 
   <th>有效报体长度</th> 
   <th>4 个字节</th> 
   <th>有效报体长度</th> 
  </tr> 
 </tbody> 
 <tbody> 
  <tr> 
   <td>length1</td> 
   <td>image1</td> 
   <td>length2</td> 
   <td>image2</td> 
  </tr> 
 </tbody> 
</table>

我们需要的用于请求和响应的编码和解码的类的概述：

*  **ImageRequest**：一个表示对我们的图片服务器请求的简单的 POJO
 *  **ImageRequestEncoder**：将 ImageRequest 对象编码为协议专用数据 (由客户端使用)
 *  **ImageRequestDecoder**：将协议专用数据解码为 ImageRequest 对象 (由服务器端使用)
 *  **ImageResponse**：一个表示来自我们图片服务器端的响应的简单的 POJO
 *  **ImageResponseEncoder**：服务器端用以对 ImageResponse 对象编码的类
 *  **ImageResponseDecoder**：客户端用以对 ImageResponse 对象解码的类
 *  **ImageCodecFactory**：这个类负责创建需要的编码器和解码器

ImageRequest 类源代码如下：

**\[java\]** [view plain][] [copy][view plain] [print][view plain] [?][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  **public****class** ImageRequest \{ 
2.  
3.  **private****int** width; 
4.  **private****int** height; 
5.  **private****int** numberOfCharacters; 
6.  
7.  **public** ImageRequest(**int** width, **int** height, **int** numberOfCharacters) \{ 
8.  **this**.width = width; 
9.  **this**.height = height; 
10. **this**.numberOfCharacters = numberOfCharacters; 
11.  \} 
12. 
13. **public****int** getWidth() \{ 
14. **return** width; 
15.  \} 
16. 
17. **public****int** getHeight() \{ 
18. **return** height; 
19.  \} 
20. 
21. **public****int** getNumberOfCharacters() \{ 
22. **return** numberOfCharacters; 
23.  \} 
24. \}

编码往往比解码容易，因此我们先从 ImageRequestEncoder 开始：

**\[java\]** [view plain][] [copy][view plain] [print][view plain] [?][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  **public****class** ImageRequestEncoder **implements** ProtocolEncoder \{ 
2.  
3.  **public****void** encode(IoSession session, Object message, ProtocolEncoderOutput out) **throws** Exception \{ 
4.   ImageRequest request = (ImageRequest) message; 
5.   IoBuffer buffer = IoBuffer.allocate(12, **false**); 
6.   buffer.putInt(request.getWidth()); 
7.   buffer.putInt(request.getHeight()); 
8.   buffer.putInt(request.getNumberOfCharacters()); 
9.   buffer.flip(); 
10.  out.write(buffer); 
11.  \} 
12. 
13. **public****void** dispose(IoSession session) **throws** Exception \{ 
14. // nothing to dispose
15.  \} 
16. \}

备注：

*  MINA 将会为所有在 IoSession 的写队列里的消息调用编码方法。既然我们的客户端只会写 ImageRequest 对象，我们可以安全地把消息放进 ImageRequest
 *  我们在堆空间分配了一个新的 IoBuffer。最好避免使用直接缓存，因为通常堆缓存表现的更好。参考 [http://issues.apache.org/jira/browse/DIRMINA-289][http_issues.apache.org_jira_browse_DIRMINA-289]
 *  不需要你去释放缓存，MINA 为你代劳了。参考[《Apache MINA 2.0 用户指南》第八章：字节缓存][Apache MINA 2.0]
 *  你应该在 dispose() 方法中释放所有在为特定会话编码时所获取的资源。如果没有任何事情要处理你可以让你的编码器继承自 ProtocolEncoderAdapter

现在我们来看一下解码器。CumulativeProtocolDecoder 绝对对写你自己的编码器有很大帮助：它将把你的解码器决定对连入数据可以做一些事情之前都缓存起来。在这种情况下消息具有固定大小，因此很容易等待所有的数据到齐以后再进行一些操作：

**\[java\]** [view plain][] [copy][view plain] [print][view plain] [?][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  **public****class** ImageRequestDecoder **extends** CumulativeProtocolDecoder \{ 
2.  
3.  **protected****boolean** doDecode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) **throws** Exception \{ 
4.  **if** (in.remaining() >= 12) \{ 
5.  **int** width = in.getInt(); 
6.  **int** height = in.getInt(); 
7.  **int** numberOfCharachters = in.getInt(); 
8.   ImageRequest request = **new** ImageRequest(width, height, numberOfCharachters); 
9.   out.write(request); 
10. **return****true**; 
11.  \} **else** \{ 
12. **return****false**; 
13.  \} 
14.  \} 
15. \}

备注：

*  每次在对一个完整的消息编码时，你应该将其写进 ProtocolDecoderOutput；这些消息将穿过过滤器链并最终到达你的 IoHandler.messageReceived 方法
 *  你不必负责释放 IoBuffer
 *  如果没有足够的可用数据用以解码一条消息，只需返回 false

响应也是一个非常简单的 POJO：

**\[java\]** [view plain][] [copy][view plain] [print][view plain] [?][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  **public****class** ImageResponse \{ 
2.  
3.  **private** BufferedImage image1; 
4.  
5.  **private** BufferedImage image2; 
6.  
7.  **public** ImageResponse(BufferedImage image1, BufferedImage image2) \{ 
8.  **this**.image1 = image1; 
9.  **this**.image2 = image2; 
10.  \} 
11. 
12. **public** BufferedImage getImage1() \{ 
13. **return** image1; 
14.  \} 
15. 
16. **public** BufferedImage getImage2() \{ 
17. **return** image2; 
18.  \} 
19. \}

响应的编码也很简单：

**\[java\]** [view plain][] [copy][view plain] [print][view plain] [?][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  **public****class** ImageResponseEncoder **extends** ProtocolEncoderAdapter \{ 
2.  
3.  **public****void** encode(IoSession session, Object message, ProtocolEncoderOutput out) **throws** Exception \{ 
4.   ImageResponse imageResponse = (ImageResponse) message; 
5.  **byte**\[\] bytes1 = getBytes(imageResponse.getImage1()); 
6.  **byte**\[\] bytes2 = getBytes(imageResponse.getImage2()); 
7.  **int** capacity = bytes1.length + bytes2.length + 8; 
8.   IoBuffer buffer = IoBuffer.allocate(capacity, **false**); 
9.   buffer.setAutoExpand(**true**); 
10.  buffer.putInt(bytes1.length); 
11.  buffer.put(bytes1); 
12.  buffer.putInt(bytes2.length); 
13.  buffer.put(bytes2); 
14.  buffer.flip(); 
15.  out.write(buffer); 
16.  \} 
17. 
18. **private****byte**\[\] getBytes(BufferedImage image) **throws** IOException \{ 
19.  ByteArrayOutputStream baos = **new** ByteArrayOutputStream(); 
20.  ImageIO.write(image, "PNG", baos); 
21. **return** baos.toByteArray(); 
22.  \} 
23. \}

备注：

*  当无法提前预测 IoBuffer 的长度时，你可以通过调用 buffer.setAutoExpand(true); 使用一个自动扩展缓存

现在我们来看一下响应的解码：

**\[java\]** [view plain][] [copy][view plain] [print][view plain] [?][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  **public****class** ImageResponseDecoder **extends** CumulativeProtocolDecoder \{ 
2.  
3.  **private****static****final** String DECODER\_STATE\_KEY = ImageResponseDecoder.**class**.getName() + ".STATE"; 
4.  
5.  **public****static****final****int** MAX\_IMAGE\_SIZE = 5 \* 1024 \* 1024; 
6.  
7.  **private****static****class** DecoderState \{ 
8.   BufferedImage image1; 
9.   \} 
10. 
11. **protected****boolean** doDecode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) **throws** Exception \{ 
12.  DecoderState decoderState = (DecoderState) session.getAttribute(DECODER\_STATE\_KEY); 
13. **if** (decoderState == **null**) \{ 
14.  decoderState = **new** DecoderState(); 
15.  session.setAttribute(DECODER\_STATE\_KEY, decoderState); 
16.  \} 
17. **if** (decoderState.image1 == **null**) \{ 
18. // try to read first image
19. **if** (in.prefixedDataAvailable(4, MAX\_IMAGE\_SIZE)) \{ 
20.  decoderState.image1 = readImage(in); 
21.  \} **else** \{ 
22. // not enough data available to read first image
23. **return****false**; 
24.  \} 
25.  \} 
26. **if** (decoderState.image1 != **null**) \{ 
27. // try to read second image
28. **if** (in.prefixedDataAvailable(4, MAX\_IMAGE\_SIZE)) \{ 
29.  BufferedImage image2 = readImage(in); 
30.  ImageResponse imageResponse = **new** ImageResponse(decoderState.image1, image2); 
31.  out.write(imageResponse); 
32.  decoderState.image1 = **null**; 
33. **return****true**; 
34.  \} **else** \{ 
35. // not enough data available to read second image
36. **return****false**; 
37.  \} 
38.  \} 
39. **return****false**; 
40.  \} 
41. 
42. **private** BufferedImage readImage(IoBuffer in) **throws** IOException \{ 
43. **int** length = in.getInt(); 
44. **byte**\[\] bytes = **new****byte**\[length\]; 
45.  in.get(bytes); 
46.  ByteArrayInputStream bais = **new** ByteArrayInputStream(bytes); 
47. **return** ImageIO.read(bais); 
48.  \} 
49. \}

备注：

*  我们将解码过程的状态保存为一个会话属性。也可以将这一状态保存在解码器对象自身中，但这么干有一些缺点：

每个 IoSession 需要自己的解码器实例  
 MINA 能够确保不会有多个线程同时为同一个 IoSession 执行 decode() 方法，但这并不能担保执行这个方法的总述同一个线程。假设第一块数据有线程 1 处理，线程 1 认为还不能解码，当下一块数据到达时，它可能会被另一个线程处理。为了避免可见性问题，你必须对这一解码状态进行适当加锁 (IoSession 的属性保存在一个 ConcurrentHashMap中，因此它们对于其他线程是自动可见的)  
 在邮件列表中的一个讨论得出了这一结论：关于是把状态保存在 IoSession 还是解码器实例中的选择是一个很有趣的问题。要确保不会有两个以上的线程为同一个 IoSession 运行解码方法，MINA 需要做某种形式的同步 => 这一同步也可以保证你不会遇到上面描述的可见性问题。(感谢 Adam Fisk 指出这一点) 参考  [http://www.nabble.com/Tutorial-on-ProtocolCodecFilter,-state-and-threads-t3965413.html][http_www.nabble.com_Tutorial-on-ProtocolCodecFilter_-state-and-threads-t3965413.html]

*  IoBuffer.prefixedDataAvailable() 在你的协议事宜一个长度前缀时很是便利；它支持 1、2 或 4 个字节的前缀
 *  在你解码一个响应的时候别忘了复位解码状态 (将该会话属性删除是解决这种问题的另一种方法)

如果响应只有单一的一个图片，我们就无需保存解码状态了：

**\[java\]** [view plain][] [copy][view plain] [print][view plain] [?][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  **protected****boolean** doDecode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) **throws** Exception \{ 
2.  **if** (in.prefixedDataAvailable(4)) \{ 
3.  **int** length = in.getInt(); 
4.  **byte**\[\] bytes = **new****byte**\[length\]; 
5.   in.get(bytes); 
6.   ByteArrayInputStream bais = **new** ByteArrayInputStream(bytes); 
7.   BufferedImage image = ImageIO.read(bais); 
8.   out.write(image); 
9.  **return****true**; 
10.  \} **else** \{ 
11. **return****false**; 
12.  \} 
13. \}

现在我们把它们都粘在一起：

**\[java\]** [view plain][] [copy][view plain] [print][view plain] [?][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  **public****class** ImageCodecFactory **implements** ProtocolCodecFactory \{ 
2.  **private** ProtocolEncoder encoder; 
3.  **private** ProtocolDecoder decoder; 
4.  
5.  **public** ImageCodecFactory(**boolean** client) \{ 
6.  **if** (client) \{ 
7.   encoder = **new** ImageRequestEncoder(); 
8.   decoder = **new** ImageResponseDecoder(); 
9.   \} **else** \{ 
10.  encoder = **new** ImageResponseEncoder(); 
11.  decoder = **new** ImageRequestDecoder(); 
12.  \} 
13.  \} 
14. 
15. **public** ProtocolEncoder getEncoder(IoSession ioSession) **throws** Exception \{ 
16. **return** encoder; 
17.  \} 
18. 
19. **public** ProtocolDecoder getDecoder(IoSession ioSession) **throws** Exception \{ 
20. **return** decoder; 
21.  \} 
22. \}

备注：

*  对于每个新会话，MINA 将会请求 ImageCodecFactory 以获取一个编码器或者一个解码器
 *  因为我们的编码器和解码器没有保存会话状态，因此让所有会话共享一个单一实例是安全的

这里是服务器端对 ProtocolCodecFactory 的使用：

**\[java\]** [view plain][] [copy][view plain] [print][view plain] [?][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  **public****class** ImageServer \{ 
2.  **public****static****final****int** PORT = 33789; 
3.  
4.  **public****static****void** main(String\[\] args) **throws** IOException \{ 
5.   ImageServerIoHandler handler = **new** ImageServerIoHandler(); 
6.   NioSocketAcceptor acceptor = **new** NioSocketAcceptor(); 
7.   acceptor.getFilterChain().addLast("protocol", **new** ProtocolCodecFilter(**new** ImageCodecFactory(**false**))); 
8.   acceptor.setLocalAddress(**new** InetSocketAddress(PORT)); 
9.   acceptor.setHandler(handler); 
10.  acceptor.bind(); 
11.  System.out.println("server is listenig at port " \+ PORT); 
12.  \} 
13. \}

客户端的使用完全一致：

**\[java\]** [view plain][] [copy][view plain] [print][view plain] [?][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  **public****class** ImageClient **extends** IoHandlerAdapter \{ 
2.  **public****static****final****int** CONNECT\_TIMEOUT = 3000; 
3.  
4.  **private** String host; 
5.  **private****int** port; 
6.  **private** SocketConnector connector; 
7.  **private** IoSession session; 
8.  **private** ImageListener imageListener; 
9.  
10. **public** ImageClient(String host, **int** port, ImageListener imageListener) \{ 
11. **this**.host = host; 
12. **this**.port = port; 
13. **this**.imageListener = imageListener; 
14.  connector = **new** NioSocketConnector(); 
15.  connector.getFilterChain().addLast("codec", **new** ProtocolCodecFilter(**new** ImageCodecFactory(**true**))); 
16.  connector.setHandler(**this**); 
17.  \} 
18. 
19. **public****void** messageReceived(IoSession session, Object message) **throws** Exception \{ 
20.  ImageResponse response = (ImageResponse) message; 
21.  imageListener.onImages(response.getImage1(), response.getImage2()); 
22.  \} 
23.  ...

完整性考虑，现在附加一个服务器端的 IoHandler 代码：

**\[java\]** [view plain][] [copy][view plain] [print][view plain] [?][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  **public****class** ImageServerIoHandler **extends** IoHandlerAdapter \{ 
2.  
3.  **private****final****static** String characters = "mina rocks abcdefghijklmnopqrstuvwxyz0123456789"; 
4.  
5.  **public****static****final** String INDEX\_KEY = ImageServerIoHandler.**class**.getName() + ".INDEX"; 
6.  
7.  **private** Logger logger = LoggerFactory.getLogger(**this**.getClass()); 
8.  
9.  **public****void** sessionOpened(IoSession session) **throws** Exception \{ 
10.  session.setAttribute(INDEX\_KEY, 0); 
11.  \} 
12. 
13. **public****void** exceptionCaught(IoSession session, Throwable cause) **throws** Exception \{ 
14.  IoSessionLogger sessionLogger = IoSessionLogger.getLogger(session, logger); 
15.  sessionLogger.warn(cause.getMessage(), cause); 
16.  \} 
17. 
18. **public****void** messageReceived(IoSession session, Object message) **throws** Exception \{ 
19.  ImageRequest request = (ImageRequest) message; 
20.  String text1 = generateString(session, request.getNumberOfCharacters()); 
21.  String text2 = generateString(session, request.getNumberOfCharacters()); 
22.  BufferedImage image1 = createImage(request, text1); 
23.  BufferedImage image2 = createImage(request, text2); 
24.  ImageResponse response = **new** ImageResponse(image1, image2); 
25.  session.write(response); 
26.  \} 
27. 
28. **private** BufferedImage createImage(ImageRequest request, String text) \{ 
29.  BufferedImage image = **new** BufferedImage(request.getWidth(), request.getHeight(), BufferedImage.TYPE\_BYTE\_INDEXED); 
30.  Graphics graphics = image.createGraphics(); 
31.  graphics.setColor(Color.YELLOW); 
32.  graphics.fillRect(0, 0, image.getWidth(), image.getHeight()); 
33.  Font serif = **new** Font("serif", Font.PLAIN, 30); 
34.  graphics.setFont(serif); 
35.  graphics.setColor(Color.BLUE); 
36.  graphics.drawString(text, 10, 50); 
37. **return** image; 
38.  \} 
39. 
40. **private** String generateString(IoSession session, **int** length) \{ 
41.  Integer index = (Integer) session.getAttribute(INDEX\_KEY); 
42.  StringBuffer buffer = **new** StringBuffer(length); 
43. 
44. **while** (buffer.length() < length) \{ 
45.  buffer.append(characters.charAt(index)); 
46.  index++; 
47. **if** (index >= characters.length()) \{ 
48.  index = 0; 
49.  \} 
50.  \} 
51.  session.setAttribute(INDEX\_KEY, index); 
52. **return** buffer.toString(); 
53.  \} 
54. \}

![codec-filter][]

[Defonds]: http://my.csdn.net/defonds
[http_mina.apache.org_mina-project_userguide_ch11-ssl-filter_ch11-ssl-filter.html]: http://mina.apache.org/mina-project/userguide/ch11-ssl-filter/ch11-ssl-filter.html
[http_en.wikipedia.org_wiki_IPv4_Fragmentation_and_reassembly]: http://en.wikipedia.org/wiki/IPv4#Fragmentation_and_reassembly
[http_en.wikipedia.org_wiki_Nagle_27s_algorithm]: http://en.wikipedia.org/wiki/Nagle%27s_algorithm
[http_www.faqs.org_rfcs_rfc977.html]: http://www.faqs.org/rfcs/rfc977.html
[view plain]: http://blog.csdn.net/defonds/article/details/18223463#
[CODE]: https://code.csdn.net/assets/CODE_ico.png
[CODE_CODE]: https://code.csdn.net/snippets/154650
[ico_fork.svg]: https://code.csdn.net/assets/ico_fork.svg
[ico_fork.svg 1]: https://code.csdn.net/snippets/154650/fork
[http_issues.apache.org_jira_browse_DIRMINA-289]: http://issues.apache.org/jira/browse/DIRMINA-289
[Apache MINA 2.0]: http://blog.csdn.net/defonds/article/details/18083377
[http_www.nabble.com_Tutorial-on-ProtocolCodecFilter_-state-and-threads-t3965413.html]: http://www.nabble.com/Tutorial-on-ProtocolCodecFilter,-state-and-threads-t3965413.html
[codec-filter]: https://img-blog.csdn.net/20140113161215437?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvZGVmb25kcw==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast