基於Apache Mina實現的TCP長鏈接和短鏈接實例

一、前言

Apache MINA是Apache組織的一個優秀的項目。MINA是Multipurpose Infrastructure for NetworkApplications的縮寫。它是一個網絡應用程序框架，用來幫助用戶很是方便地開發高性能和高可靠性的網絡應用程序。在本文中介紹瞭如何經過Apache Mina2.0來實現TCP協議長鏈接和短鏈接應用。

二、系統介紹

2.1系統框架

整個系統由兩個服務端程序和兩個客戶端程序組成。分別實現TCP長鏈接和短鏈接通訊。

系統業務邏輯是一個客戶端與服務端創建長鏈接，一個客戶端與服務端創建短鏈接。數據從短鏈接客戶端通過服務端發送到長鏈接客戶端，並從長鏈接客戶端接收響應數據。當收到響應數據後斷開鏈接。

系統架構圖以下：

2.2處理流程

系統處理流程以下：

1）       啓動服務端程序，監聽8001和8002端口。

2）       長鏈接客戶端向服務端8002端口創建鏈接，服務端將鏈接對象保存到共享內存中。因爲採用長鏈接方式，鏈接對象是惟一的。

3）       短鏈接客戶端向服務端8001端口創建鏈接。創建鏈接後建立一個鏈接對象。

4）       短鏈接客戶端鏈接成功後發送數據。服務端接收到數據後從共享內存中獲得長鏈接方式的鏈接對象，使用此對象向長鏈接客戶端發送數據。發送前將短鏈接對象設爲長鏈接對象的屬性值。

5）       長鏈接客戶端接收到數據後返回響應數據。服務端從長鏈接對象的屬性中取得短鏈接對象，經過此對象將響應數據發送給短鏈接客戶端。

6）       短鏈接客戶端收到響應數據後，關閉鏈接。

三、服務端程序

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.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);

    }

}

五、總結

經過本文中的例子，Apache Mina在服務端可實現TCP協議長鏈接和短鏈接。在客戶端只實現了短鏈接模式，長鏈接模式也是能夠實現的（在本文中仍是採用傳統的Java Socket方式）。兩個服務端之間經過共享內存的方式來傳遞鏈接對象也許有更好的實現方式。

 

工程下載地址： http://download.csdn.net/detail/peterwanghao/5306972