首页 > 代码库 > NIO框架之MINA源码解析(一):背景
NIO框架之MINA源码解析(一):背景
“你们的agent占了好多系统的端口,把我们的很多业务系统都给整死了,给我们造成了很大的损失,要求你们的相关领导下周过来道歉” -- 来自我们的一个客户。
怎么可能呢,我们都不相信,我们的agent只占一个端口啊!
事实胜过雄辩,经过查证,确实是由于我们的agent占了好多系统的端口,我看了一下日志,基本把系统可用的端口占完了!
为什么呢?MINA框架私自开的!
由于我们的agent端使用了NIO通信框架MINA,但并没有使用好,造成了这一几乎毁灭行的灾难。
还是先看代码吧。
/** * 异步发送消息 * @param agent * @param request */ public void sendMessageToAgent(Agent agent, HyRequest request) { IoSession session = null; IoConnector connector=null; long startTime = System.currentTimeMillis(); try { // 创建一个非阻塞的客户端程序 connector = new NioSocketConnector(); // 设置链接超时时间 connector.setConnectTimeoutMillis(connectTimeoutMillis); ObjectSerializationCodecFactory objsCodec = new ObjectSerializationCodecFactory(); objsCodec.setDecoderMaxObjectSize(DEFAULTDECODER); objsCodec.setEncoderMaxObjectSize(DEFAULTDECODER); ProtocolCodecFilter codecFilter = new ProtocolCodecFilter( objsCodec); // 数据转换,编码设置 connector.getFilterChain() .addLast("codec", codecFilter); // 消息 connector.setHandler(clientHandler); SocketAddress socketAddress = new InetSocketAddress( agent.getIpAddr(), agent.getAgentPort()); ConnectFuture future = connector.connect(socketAddress); future.awaitUninterruptibly(); session = future.getSession(); String json = mapper.writeValueAsString(request); session.write(json); long endTime = System.currentTimeMillis(); logerr.debug("send-time:" + (endTime - startTime)); } catch (Exception e) { logerr.error("host:" + agent.getIpAddr() + ", AgentPORT:" + agent.getAgentPort() + ", 连接异常..."+e.getMessage()); clientHandler.handlerConnectError(agent, request); } }
public class MinaClientHandler extends IoHandlerAdapter { // 日志 private Logger log = Logger.getLogger(getClass()); private MinaResponseProcesser minaResponseProcesser; ObjectMapper mapper=null; @Override public void messageReceived(IoSession session, Object message) throws Exception { String msg = message.toString(); log.info("receive message from " + session.getRemoteAddress().toString() + ",message:" + message); if(null == mapper){ mapper = new ObjectMapper(); } //请求消息转换为HyResponse对象 HyResponse response = mapper.readValue(msg, HyResponse.class); String remoteIp= ((InetSocketAddress)session.getRemoteAddress()).getAddress().getHostAddress(); response.setRemoteIp(remoteIp); HyRequest request = minaResponseProcesser.processResponse(response); if(request == null){ //关闭当前session closeSessionByServer(session,response); }else{ session.write(mapper.writeValueAsString(request)); } } }
上面的逻辑就是,当要发送一个消息时,创建一个新的connector,并获取一个session发送消息后直接返回,在MinaClientHandler类的messageReceived里面处理接受到的响应数据,并进行业务处理,最后如果不需要再次发送请求,则关闭当前session。
其实出现本文一开始的问题就是在这里造成的。
在出现我们的agent占用大量端口后,我们这边的工程人员就迅速定位到了这个问题,并很快修复了,但修复并不理想,但修复过后的代码。
/** * 异步发送消息 * @param agent * @param request */ public void sendMessageToAgent(Agent agent, HyRequest request) { IoSession session = null; IoConnector connector=null; long startTime = System.currentTimeMillis(); try { // 创建一个非阻塞的客户端程序 connector = new NioSocketConnector(); // 设置链接超时时间 connector.setConnectTimeoutMillis(connectTimeoutMillis); ObjectSerializationCodecFactory objsCodec = new ObjectSerializationCodecFactory(); objsCodec.setDecoderMaxObjectSize(DEFAULTDECODER); objsCodec.setEncoderMaxObjectSize(DEFAULTDECODER); ProtocolCodecFilter codecFilter = new ProtocolCodecFilter( objsCodec); // 数据转换,编码设置 connector.getFilterChain() .addLast("codec", codecFilter); // 消息 connector.setHandler(clientHandler); SocketAddress socketAddress = new InetSocketAddress( agent.getIpAddr(), agent.getAgentPort()); ConnectFuture future = connector.connect(socketAddress); future.awaitUninterruptibly(); session = future.getSession(); String json = mapper.writeValueAsString(request); session.write(json); // 等待断开连接 session.getCloseFuture().awaitUninterruptibly(); long endTime = System.currentTimeMillis(); logerr.debug("send-time:" + (endTime - startTime)); //connector.dispose(); } catch (Exception e) { logerr.error("host:" + agent.getIpAddr() + ", AgentPORT:" + agent.getAgentPort() + ", 连接异常..."+e.getMessage()); clientHandler.handlerConnectError(agent, request); }finally{ if(null!=session){ session.close(true); session=null; } if(null !=connector){ connector.dispose(); } } }
只改了一个地方,就是在发送完消息后,加了一个等待断开连接语句和finally语句块-关闭session和connector。
虽然不会出现程序占用大量的系统端口这个问题,但会造成另外一个问题-当有一个消息队列需要异步调用上面语句发送消息时,有原来的异步(发送完直接返回,相当于快速并发发送)变成伪异步(发送完消息后并等待消息返回处理后返回,相当于顺序处理队列里面的消息)。
上面的修改并不是我们想要的结果,但至少修复了占用大量端口的问题。
由于怀着想彻底修复这个问题的想法,我想还是深入了解一下MINA源码吧。
NIO框架之MINA源码解析(一):背景