基于Apache Mina实现的TCP长连接和短连接实例

1、前言

Apache MINA是Apache组织的一个优秀的项目。MINA是Multipurpose Infrastructure for NetworkApplications的缩写。它是一个网络应用程序框架，用来帮助用户非常方便地开发高性能和高可靠性的网络应用程序。在本文中介绍了 如何通过Apache Mina2.0来实现TCP协议长连接和短连接应用。

2、系统介绍

2.1系统框架

整个系统由两个服务端程序和两个客户端程序组成。分别实现TCP长连接和短连接通信。

系统业务逻辑是一个客户端与服务端建立长连接，一个客户端与服务端建立短连接。数据从短连接客户端经过服务端发送到长连接客户端，并从长连接客户端接收响应数据。当收到响应数据后断开连接。

系统架构图如下：

2.2处理流程

系统处理流程如下：

1）       启动服务端程序，监听8001和8002端口。

2）       长连接客户端向服务端8002端口建立连接，服务端将连接对象保存到共享内存中。由于采用长连接方式，连接对象是唯一的。

3）       短连接客户端向服务端8001端口建立连接。建立连接后创建一个连接对象。

4）       短连接客户端连接成功后发送数据。服务端接收到数据后从共享内存中得到长连接方式的连接对象，使用此对象向长连接客户端发送数据。发送前将短连接对象设为长连接对象的属性值。

5）       长连接客户端接收到数据后返回响应数据。服务端从长连接对象的属性中取得短连接对象，通过此对象将响应数据发送给短连接客户端。

6）       短连接客户端收到响应数据后，关闭连接。

3、服务端程序

3.1长连接服务端

服务启动

public class MinaLongConnServer {

private static final int PORT = 8002;

   

    public void start()throws IOException{

       IoAcceptor acceptor = new NioSocketAcceptor();

 

       acceptor.getFilterChain().addLast("logger", new LoggingFilter());

       acceptor.getFilterChain().addLast("codec", newProtocolCodecFilter(new TextLineCodecFactory(Charset.forName("UTF-8"))));

 

       acceptor.setHandler(new MinaLongConnServerHandler());

       acceptor.getSessionConfig().setReadBufferSize(2048);

       acceptor.bind(new InetSocketAddress(PORT));

       System.out.println("Listeningon port " + PORT);

    }

}

消息处理

public class MinaLongConnServerHandler extends IoHandlerAdapter {

    private final Logger logger = (Logger) LoggerFactory.getLogger(getClass());

 

    @Override

    public void sessionOpened(IoSession session) {

       InetSocketAddress remoteAddress = (InetSocketAddress)session.getRemoteAddress();

       String clientIp = remoteAddress.getAddress().getHostAddress();

       logger.info("LongConnect Server opened Session ID ="+String.valueOf(session.getId()));

       logger.info("接收来自客户端 :" + clientIp + "的连接.");

       Initialization init = Initialization.getInstance();

       HashMap<String, IoSession> clientMap =init.getClientMap();

       clientMap.put(clientIp, session);

    }

 

    @Override

    public void messageReceived(IoSession session, Object message) {

       logger.info("Messagereceived in the long connect server..");

       String expression = message.toString();

       logger.info("Message is:" + expression);

       IoSession shortConnSession =(IoSession) session.getAttribute("shortConnSession");

       logger.info("ShortConnect Server Session ID ="+String.valueOf(shortConnSession.getId()));

       shortConnSession.write(expression);

    }

 

    @Override

    public void sessionIdle(IoSession session, IdleStatus status) {

       logger.info("Disconnectingthe idle.");

       // disconnect an idle client

       session.close(true);

    }

 

    @Override

    public void exceptionCaught(IoSession session, Throwable cause) {

       // close the connection onexceptional situation

       logger.warn(cause.getMessage(), cause);

       session.close(true);

    }

}

3.2短连接服务端

服务启动

public class MinaShortConnServer {

    private static final int PORT = 8001;

   

    public void start()throws IOException{

       IoAcceptor acceptor = new NioSocketAcceptor();

 

       acceptor.getFilterChain().addLast("logger", new LoggingFilter());

       acceptor.getFilterChain().addLast("codec", newProtocolCodecFilter(new TextLineCodecFactory(Charset.forName("UTF-8"))));

 

       acceptor.setHandler(new MinaShortConnServerHandler());

       acceptor.getSessionConfig().setReadBufferSize(2048);

       acceptor.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, 3);

       acceptor.bind(new InetSocketAddress(PORT));

       System.out.println("Listeningon port " + PORT);

    }

}

消息处理

public class MinaShortConnServerHandler extends IoHandlerAdapter {

    private final Logger logger = (Logger) LoggerFactory.getLogger(getClass());

 

    @Override

    public void sessionOpened(IoSession session) {

       InetSocketAddress remoteAddress = (InetSocketAddress)session.getRemoteAddress();

       logger.info(remoteAddress.getAddress().getHostAddress());

       logger.info(String.valueOf(session.getId()));

    }

 

    @Override

    public void messageReceived(IoSession session, Object message) {

       logger.info("Messagereceived in the short connect server...");

       String expression = message.toString();

       Initialization init = Initialization.getInstance();

       HashMap<String, IoSession> clientMap =init.getClientMap();

       if (clientMap == null || clientMap.size() == 0) {

           session.write("error");

       } else {

           IoSession longConnSession = null;

           Iterator<String> iterator =clientMap.keySet().iterator();

           String key = "";

           while (iterator.hasNext()) {

              key = iterator.next();

              longConnSession = clientMap.get(key);

           }

           logger.info("ShortConnect Server Session ID :"+String.valueOf(session.getId()));

           logger.info("LongConnect Server Session ID :"+String.valueOf(longConnSession.getId()));

           longConnSession.setAttribute("shortConnSession",session);

           longConnSession.write(expression);

       }

    }

 

    @Override

    public void sessionIdle(IoSession session, IdleStatus status) {

       logger.info("Disconnectingthe idle.");

       // disconnect an idle client

       session.close(true);

    }

 

    @Override

    public void exceptionCaught(IoSession session, Throwable cause) {

       // close the connection onexceptional situation

       logger.warn(cause.getMessage(), cause);

       session.close(true);

    }

}

 

4、客户端程序

4.1长连接客户端

使用java.net.Socket来实现向服务端建立连接。Socket建立后一直保持连接，从服务端接收到数据包后直接将原文返回。

public class TcpKeepAliveClient {

    private String ip;

    private int port;

    private static Socket socket = null;

    private static int timeout = 50 * 1000;

 

    public TcpKeepAliveClient(String ip, int port) {

       this.ip = ip;

       this.port = port;

    }

 

    public void receiveAndSend() throws IOException {

       InputStream input = null;

       OutputStream output = null;

 

       try {

           if (socket == null || socket.isClosed() || !socket.isConnected()) {

              socket = new Socket();

              InetSocketAddress addr = new InetSocketAddress(ip, port);

              socket.connect(addr, timeout);

              socket.setSoTimeout(timeout);

              System.out.println("TcpKeepAliveClientnew ");

           }

 

           input = socket.getInputStream();

           output = socket.getOutputStream();

 

           // read body

           byte[] receiveBytes = {};// 收到的包字节数组

           while (true) {

              if (input.available() > 0) {

                  receiveBytes = new byte[input.available()];

                  input.read(receiveBytes);

 

                  // send

                  System.out.println("TcpKeepAliveClientsend date :" + new String(receiveBytes));

                  output.write(receiveBytes, 0, receiveBytes.length);

                  output.flush();

              }

           }

 

       } catch (Exception e) {

           e.printStackTrace();

           System.out.println("TcpClientnew socket error");

       }

    }

 

    public static void main(String[] args) throws Exception {

       TcpKeepAliveClient client = new TcpKeepAliveClient("127.0.0.1", 8002);

       client.receiveAndSend();

    }

 

}

4.2短连接客户端

服务启动

public class MinaShortClient {

    private static final int PORT = 8001;

 

    public static void main(String[] args) throws IOException,InterruptedException {

       IoConnector connector = new NioSocketConnector();

       connector.getSessionConfig().setReadBufferSize(2048);

 

       connector.getFilterChain().addLast("logger", new LoggingFilter());

       connector.getFilterChain().addLast("codec", newProtocolCodecFilter(new TextLineCodecFactory(Charset.forName("UTF-8"))));

 

       connector.setHandler(new MinaShortClientHandler());

       for (int i = 1; i <= 10; i++) {

           ConnectFuture future = connector.connect(new InetSocketAddress("127.0.0.1", PORT));

           future.awaitUninterruptibly();

           IoSession session =future.getSession();

           session.write(i);

           session.getCloseFuture().awaitUninterruptibly();

 

           System.out.println("result=" + session.getAttribute("result"));

       }

       connector.dispose();

 

    }

}

消息处理

public class MinaShortClientHandler extends IoHandlerAdapter{

    private final Logger logger = (Logger) LoggerFactory.getLogger(getClass());

 

    public MinaShortClientHandler() {

      

    }

 

    @Override

    public void sessionOpened(IoSession session) {

    }

 

    @Override

    public void messageReceived(IoSession session, Object message) {

       logger.info("Messagereceived in the client..");

       logger.info("Message is:" + message.toString());

       session.setAttribute("result", message.toString());

       session.close(true);

    }

 

    @Override

    public void exceptionCaught(IoSession session, Throwable cause) {

       session.close(true);

    }

}

5、总结

通过本文中的例子，Apache Mina在服务端可实现TCP协议长连接和短连接。在客户端只实现了短连接模式，长连接模式也是可以实现的（在本文中还是采用传统的java Socket方式）。两个服务端之间通过共享内存的方式来传递连接对象也许有更好的实现方式。

评论

13 楼 smallbee 2013-08-13  

brad2309 写道

lz
你的两个messageReceived方法是不是错乱了

?

12 楼 brad2309 2013-08-13  

lz
你的两个messageReceived方法是不是错乱了

11 楼 sund201006 2012-06-01  

sund201006 写道

可否提供一个客户端长连接的例子，以参考学习

可否提供一个利用mina建立长连接的客户端的例子，以参考学习

10 楼 sund201006 2012-06-01  

可否提供一个客户端长连接的例子，以参考学习

9 楼 红发programmer 2012-04-17  

很感谢你耐心的回复和指导，希望以后能多交流知识。像您学习。

8 楼 smallbee 2012-04-17  

mina自带的Text的那个过滤器

红发programmer 写道

smallbee 写道

红发programmer 写道

smallbee 写道

          

红发programmer 写道

smallbee 写道

红发programmer 写道

请教下，如何利用mina建立长连接的客户端。
我尝试了很多次，但都有很大的问题。。。

思路就是连接上来，然后把iosession保留在后台单例下 ，这样就可以实现长连接。

可是，我使用短连接的时候，每次它都会阻塞在session.getCloseFuture().awaitUninterruptibly();
不再往下运行，很苦恼。。。

你不是要实现长连接么？如果是长连接，就是阻塞的。

我按照您的socket长连接客户端的例子，但是你每次只是发送接收到的按个byte[]。有没有办法发送自己定义的byte[]，我试了几次，服务器可以接受，但是无法处理自己发送的byte[]。

无法处理时什么意思？服务器可以接收到，说明通信是没有问题的，注意你的编码方式。

很感谢你耐心的回答，现在我遇到另一个问题，就是当我的字符串长度为113的时候，发送的消息是可以被服务器处理的，超过这个长度，服务器的Iohandler就无法处理了。这是怎么回事呢？我用了mina自带的Text的那个过滤器。

没用过mina自带的Text的那个过滤器，我们是自己写的。因为socket是无边界的，所以需要你在自己的报文结构中（一般在前面）指定我们传输的报文长度，然后往后面截取即可。

7 楼 红发programmer 2012-04-17  

smallbee 写道

红发programmer 写道

smallbee 写道

          

红发programmer 写道

smallbee 写道

红发programmer 写道

请教下，如何利用mina建立长连接的客户端。
我尝试了很多次，但都有很大的问题。。。

思路就是连接上来，然后把iosession保留在后台单例下 ，这样就可以实现长连接。

可是，我使用短连接的时候，每次它都会阻塞在session.getCloseFuture().awaitUninterruptibly();
不再往下运行，很苦恼。。。

你不是要实现长连接么？如果是长连接，就是阻塞的。

我按照您的socket长连接客户端的例子，但是你每次只是发送接收到的按个byte[]。有没有办法发送自己定义的byte[]，我试了几次，服务器可以接受，但是无法处理自己发送的byte[]。

无法处理时什么意思？服务器可以接收到，说明通信是没有问题的，注意你的编码方式。

很感谢你耐心的回答，现在我遇到另一个问题，就是当我的字符串长度为113的时候，发送的消息是可以被服务器处理的，超过这个长度，服务器的Iohandler就无法处理了。这是怎么回事呢？我用了mina自带的Text的那个过滤器。

6 楼 smallbee 2012-04-12  

红发programmer 写道

smallbee 写道

          

红发programmer 写道

smallbee 写道

红发programmer 写道

请教下，如何利用mina建立长连接的客户端。
我尝试了很多次，但都有很大的问题。。。

思路就是连接上来，然后把iosession保留在后台单例下 ，这样就可以实现长连接。

可是，我使用短连接的时候，每次它都会阻塞在session.getCloseFuture().awaitUninterruptibly();
不再往下运行，很苦恼。。。

你不是要实现长连接么？如果是长连接，就是阻塞的。

我按照您的socket长连接客户端的例子，但是你每次只是发送接收到的按个byte[]。有没有办法发送自己定义的byte[]，我试了几次，服务器可以接受，但是无法处理自己发送的byte[]。

无法处理时什么意思？服务器可以接收到，说明通信是没有问题的，注意你的编码方式。

5 楼 红发programmer 2012-04-11  

smallbee 写道

          

红发programmer 写道

smallbee 写道

红发programmer 写道

请教下，如何利用mina建立长连接的客户端。
我尝试了很多次，但都有很大的问题。。。

思路就是连接上来，然后把iosession保留在后台单例下 ，这样就可以实现长连接。

可是，我使用短连接的时候，每次它都会阻塞在session.getCloseFuture().awaitUninterruptibly();
不再往下运行，很苦恼。。。

你不是要实现长连接么？如果是长连接，就是阻塞的。

我按照您的socket长连接客户端的例子，但是你每次只是发送接收到的按个byte[]。有没有办法发送自己定义的byte[]，我试了几次，服务器可以接受，但是无法处理自己发送的byte[]。

4 楼 smallbee 2012-04-11  

          

红发programmer 写道

smallbee 写道

红发programmer 写道

请教下，如何利用mina建立长连接的客户端。
我尝试了很多次，但都有很大的问题。。。

思路就是连接上来，然后把iosession保留在后台单例下 ，这样就可以实现长连接。

可是，我使用短连接的时候，每次它都会阻塞在session.getCloseFuture().awaitUninterruptibly();
不再往下运行，很苦恼。。。

你不是要实现长连接么？如果是长连接，就是阻塞的。

3 楼 红发programmer 2012-04-11  

smallbee 写道

红发programmer 写道

请教下，如何利用mina建立长连接的客户端。
我尝试了很多次，但都有很大的问题。。。

思路就是连接上来，然后把iosession保留在后台单例下 ，这样就可以实现长连接。

可是，我使用短连接的时候，每次它都会阻塞在session.getCloseFuture().awaitUninterruptibly();
不再往下运行，很苦恼。。。

2 楼 smallbee 2012-04-11  

红发programmer 写道

请教下，如何利用mina建立长连接的客户端。
我尝试了很多次，但都有很大的问题。。。

思路就是连接上来，然后把iosession保留在后台单例下 ，这样就可以实现长连接。

1 楼 红发programmer 2012-04-11  

请教下，如何利用mina建立长连接的客户端。
我尝试了很多次，但都有很大的问题。。。