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源码解析（一）：背景

标签：nio   mina   端口   

原文地址：http://blog.csdn.net/chaofanwei/article/details/38848085