NIO框架之MINA源码解析（一）：背景

Dear 丶 2022-08-05 08:10 203阅读 0赞

**“你们的agent占了好多系统的端口，把我们的很多业务系统都给整死了，给我们造成了很大的损失，要求你们的相关领导下周过来道歉**” \-- 来自我们的一个客户。

怎么可能呢，我们都不相信，我们的agent只占一个端口啊！

事实胜过雄辩，经过查证，确实是由于我们的agent占了好多系统的端口，我看了一下日志，基本把系统可用的端口占完了！

![Center][]

为什么呢？MINA框架私自开的！

由于我们的agent端使用了NIO通信框架MINA，但并没有使用好，造成了这一几乎毁灭行的灾难。

还是先看代码吧。

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

1.  /\*\*
2.   \* 异步发送消息
3.   \* @param agent
4.   \* @param request
5.   \*/
6.  **public****void** sendMessageToAgent(Agent agent, HyRequest request) \{ 
7.   IoSession session = **null**; 
8.   IoConnector connector=**null**; 
9.  **long** startTime = System.currentTimeMillis(); 
10. **try** \{ 
11. // 创建一个非阻塞的客户端程序
12.  connector = **new** NioSocketConnector(); 
13. // 设置链接超时时间
14.  connector.setConnectTimeoutMillis(connectTimeoutMillis); 
15. 
16.  ObjectSerializationCodecFactory objsCodec = **new** ObjectSerializationCodecFactory(); 
17.  objsCodec.setDecoderMaxObjectSize(DEFAULTDECODER); 
18.  objsCodec.setEncoderMaxObjectSize(DEFAULTDECODER); 
19.  ProtocolCodecFilter codecFilter = **new** ProtocolCodecFilter( 
20.  objsCodec); 
21. // 数据转换，编码设置
22.  connector.getFilterChain() 
23.  .addLast("codec", codecFilter); 
24. // 消息
25.  connector.setHandler(clientHandler); 
26. 
27.  SocketAddress socketAddress = **new** InetSocketAddress( 
28.  agent.getIpAddr(), agent.getAgentPort()); 
29.  ConnectFuture future = connector.connect(socketAddress); 
30.  future.awaitUninterruptibly(); 
31.  session = future.getSession(); 
32.  String json = mapper.writeValueAsString(request); 
33.  session.write(json); 
34. 
35. **long** endTime = System.currentTimeMillis(); 
36. 
37.  logerr.debug("send-time:" \+ (endTime - startTime)); 
38. 
39.  \} **catch** (Exception e) \{ 
40.  logerr.error("host:" \+ agent.getIpAddr() + ", AgentPORT:" \+ agent.getAgentPort() 
41.  \+ ", 连接异常..."\+e.getMessage()); 
42.  clientHandler.handlerConnectError(agent, request); 
43. 
44.  \} 
45. \}

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

1.  **public****class** MinaClientHandler **extends** IoHandlerAdapter \{ 
2.  // 日志
3.  **private** Logger log = Logger.getLogger(getClass()); 
4.  
5.  **private** MinaResponseProcesser minaResponseProcesser; 
6.  
7.   ObjectMapper mapper=**null**; 
8.  
9.  @Override
10. **public****void** messageReceived(IoSession session, Object message) 
11. **throws** Exception \{ 
12.  String msg = message.toString(); 
13.  log.info("receive message from " \+ session.getRemoteAddress().toString() + ",message:" \+ message); 
14. **if**(**null** == mapper)\{ 
15.  mapper = **new** ObjectMapper(); 
16.  \} 
17. //请求消息转换为HyResponse对象
18.  HyResponse response = mapper.readValue(msg, HyResponse.**class**); 
19.  String remoteIp= ((InetSocketAddress)session.getRemoteAddress()).getAddress().getHostAddress(); 
20.  response.setRemoteIp(remoteIp); 
21.  HyRequest request = minaResponseProcesser.processResponse(response); 
22. **if**(request == **null**)\{ 
23. //关闭当前session
24.  closeSessionByServer(session,response); 
25.  \}**else**\{ 
26.  session.write(mapper.writeValueAsString(request)); 
27.  \} 
28.  \} 
29. \}

上面的逻辑就是，当要发送一个消息时，创建一个新的connector，并获取一个session发送消息后直接返回，在MinaClientHandler类的messageReceived里面处理接受到的响应数据，并进行业务处理，最后如果不需要再次发送请求，则关闭当前session。

其实出现本文一开始的问题就是在这里造成的。

在出现我们的agent占用大量端口后，我们这边的工程人员就迅速定位到了这个问题，并很快修复了，但修复并不理想，但修复过后的代码。

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

1.  /\*\*
2.   \* 异步发送消息
3.   \* @param agent
4.   \* @param request
5.   \*/
6.  **public****void** sendMessageToAgent(Agent agent, HyRequest request) \{ 
7.   IoSession session = **null**; 
8.   IoConnector connector=**null**; 
9.  **long** startTime = System.currentTimeMillis(); 
10. **try** \{ 
11. // 创建一个非阻塞的客户端程序
12.  connector = **new** NioSocketConnector(); 
13. // 设置链接超时时间
14.  connector.setConnectTimeoutMillis(connectTimeoutMillis); 
15. 
16.  ObjectSerializationCodecFactory objsCodec = **new** ObjectSerializationCodecFactory(); 
17.  objsCodec.setDecoderMaxObjectSize(DEFAULTDECODER); 
18.  objsCodec.setEncoderMaxObjectSize(DEFAULTDECODER); 
19.  ProtocolCodecFilter codecFilter = **new** ProtocolCodecFilter( 
20.  objsCodec); 
21. // 数据转换，编码设置
22.  connector.getFilterChain() 
23.  .addLast("codec", codecFilter); 
24. // 消息
25.  connector.setHandler(clientHandler); 
26. 
27.  SocketAddress socketAddress = **new** InetSocketAddress( 
28.  agent.getIpAddr(), agent.getAgentPort()); 
29.  ConnectFuture future = connector.connect(socketAddress); 
30.  future.awaitUninterruptibly(); 
31.  session = future.getSession(); 
32.  String json = mapper.writeValueAsString(request); 
33.  session.write(json); 
34. // 等待断开连接
35.  session.getCloseFuture().awaitUninterruptibly(); 
36. **long** endTime = System.currentTimeMillis(); 
37. 
38.  logerr.debug("send-time:" \+ (endTime - startTime)); 
39. //connector.dispose();
40.  \} **catch** (Exception e) \{ 
41.  logerr.error("host:" \+ agent.getIpAddr() + ", AgentPORT:" \+ agent.getAgentPort() 
42.  \+ ", 连接异常..."\+e.getMessage()); 
43.  clientHandler.handlerConnectError(agent, request); 
44. 
45.  \}**finally**\{ 
46. **if**(**null**!=session)\{ 
47.  session.close(**true**); 
48.  session=**null**; 
49.  \} 
50. **if**(**null** !=connector)\{ 
51.  connector.dispose(); 
52.  \} 
53.  \} 
54. \}

只改了一个地方，就是在发送完消息后，加了一个等待断开连接语句和finally语句块-关闭session和connector。

虽然不会出现程序占用大量的系统端口这个问题，但会造成另外一个问题-当有一个消息队列需要异步调用上面语句发送消息时，有原来的异步（发送完直接返回，相当于快速并发发送）变成伪异步（发送完消息后并等待消息返回处理后返回，相当于顺序处理队列里面的消息）。

上面的修改并不是我们想要的结果，但至少修复了占用大量端口的问题。

由于怀着想彻底修复这个问题的想法，我想还是深入了解一下MINA源码吧。

[Center]: /images/20220731/41579a5a2a714108bccd75bb2475533d.png
[view plain]: http://blog.csdn.net/chaofanwei/article/details/38848085#