Java Socket框架Apache MINA实现Socket服务器-客户端双向即时消息通信

不念不忘少年蓝@ 2024-02-19 18:31 163阅读 0赞

## Java Socket框架Apache MINA实现Socket服务器-客户端双向即时消息通信 ##

基于Java Socket框架Apache MINA，实现一对简单的基于Socket的服务-客户端双向即时消息通信。服务器绑定本地80端口，客户端连接服务器，建立连接后，双方开始互发消息服务器端可客户端均能在控制台收发消息（控制台输入字符串然后按enter键发送）。当任意一方发送消息字符串“bye”时，通信结束。

服务器端程序代码，用Apache MINA实现：

import java.io.BufferedReader;
    import java.io.InputStreamReader;
    import java.net.InetSocketAddress;
    import java.nio.charset.Charset;
    import java.util.concurrent.CompletableFuture;
    
    import org.apache.mina.core.service.IoAcceptor;
    import org.apache.mina.core.service.IoHandlerAdapter;
    import org.apache.mina.core.session.IdleStatus;
    import org.apache.mina.core.session.IoSession;
    import org.apache.mina.filter.codec.ProtocolCodecFilter;
    import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
    import org.apache.mina.filter.logging.LoggingFilter;
    import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
    
    public class MINATest {
    	private IoSession mIoSession;
    
    	public static void main(String[] args) {
    		try {
    			MINATest test = new MINATest();
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    	}
    
    	public MINATest() throws Exception {
    		IoAcceptor acceptor = new NioSocketAcceptor();
    
    		// 过滤链。
    		acceptor.getFilterChain().addLast("logger", new LoggingFilter());
    		acceptor.getFilterChain().addLast("codec",
    				new ProtocolCodecFilter(new TextLineCodecFactory(Charset.forName("UTF-8"))));
    
    		SocketServerHandler socketServerHandler = new SocketServerHandler();
    
    		// 处理。
    		acceptor.setHandler(socketServerHandler);
    
    		// 服务器端绑定80端口，等待客户端连接请求。
    		acceptor.bind(new InetSocketAddress(80));
    
    		waitServerInput();
    	}
    
    	// Apache MINA的Socket服务器端。
    	private class SocketServerHandler extends IoHandlerAdapter {
    
    		public SocketServerHandler() {
    
    		}
    
    		// 会话创建。
    		@Override
    		public void sessionCreated(IoSession session) throws Exception {
    			super.sessionCreated(session);
    			// System.out.println("#sessionCreated#");
    		}
    
    		// 会话打开时触发（第一次连接打开时先触发sessionCreated，后触发本函数）。
    		@Override
    		public void sessionOpened(IoSession session) throws Exception {
    			super.sessionOpened(session);
    			// System.out.println("#sessionOpened#");
    
    			mIoSession = session;
    		}
    
    		@Override
    		public void messageSent(IoSession session, Object message) throws Exception {
    			super.messageSent(session, message);
    			// System.out.println("\n#messageSent#");
    			// System.out.println(message.toString());
    			// System.out.println("=============");
    		}
    
    		// 接收到消息时触发。
    		@Override
    		public void messageReceived(IoSession session, Object message) throws Exception {
    			super.messageReceived(session, message);
    			// System.out.println("messageReceived");
    
    			System.out.println("客户端说:" + message.toString());
    		}
    
    		@Override
    		public void sessionClosed(IoSession session) throws Exception {
    			super.sessionClosed(session);
    			System.out.println("\n#sessionClosed#");
    			sessionInfoPrinter(session);
    		}
    
    		@Override
    		public void sessionIdle(IoSession session, IdleStatus status) throws Exception {
    			super.sessionIdle(session, status);
    			System.out.println("#sessionIdle#");
    		}
    
    		@Override
    		public void exceptionCaught(IoSession session, Throwable cause) throws Exception {
    			super.exceptionCaught(session, cause);
    			System.out.println("#exceptionCaught#");
    			sessionInfoPrinter(session);
    			cause.printStackTrace();
    		}
    	}
    
    	private void sessionInfoPrinter(IoSession session) {
    		StringBuffer sb = new StringBuffer();
    		sb.append("\n" + session.toString());
    		sb.append("\nId:" + session.getId());
    		sb.append("\nLocalAddress:" + session.getLocalAddress());
    		sb.append("\nRemoteAddress:" + session.getRemoteAddress());
    		System.out.println(sb.toString());
    	}
    
    	private void waitServerInput() {
    		CompletableFuture.runAsync(new Runnable() {
    			@Override
    			public void run() {
    				BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
    
    				String s = null;
    				while (true) {
    					try {
    						s = br.readLine();
    					} catch (Exception e) {
    						e.printStackTrace();
    					}
    
    					try {
    						mIoSession.write(s);
    					} catch (Exception e) {
    						e.printStackTrace();
    					}
    
    					if (s.equals("bye")) {
    						break;
    					}
    				}
    
    				try {
    					br.close();
    					mIoSession.closeOnFlush();
    				} catch (Exception e) {
    					e.printStackTrace();
    				}
    			}
    		});
    	}
    }

客户端代码，用Java自身的Socket实现：

import java.io.BufferedInputStream;
    import java.io.BufferedReader;
    import java.io.BufferedWriter;
    import java.io.IOException;
    import java.io.InputStreamReader;
    import java.io.OutputStreamWriter;
    import java.net.Socket;
    import java.util.concurrent.CompletableFuture;
    
    
    public class ClientTest {
    	private Socket socket;
    
    	public static void main(String[] args) {
    		try {
    			ClientTest test = new ClientTest();
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    	}
    
    	public ClientTest() throws Exception {
    		socket = new Socket("localhost", 80);
    		BufferedInputStream bis = new BufferedInputStream(socket.getInputStream());
    
    		waitClientInput();
    
    		int count = 0;
    		byte[] buffer = new byte[1024];
    
    		while (true) {
    			count = bis.read(buffer);
    			if (count == -1) {
    				break;
    			}
    
    			System.out.print("服务器说:" + new String(buffer, 0, count, "UTF-8"));
    		}
    
    		bis.close();
    		socket.close();
    	}
    
    	private void waitClientInput() {
    		CompletableFuture.runAsync(new Runnable() {
    			@Override
    			public void run() {
    				BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
    				BufferedWriter bw = null;
    				try {
    					bw = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()));
    				} catch (Exception e) {
    					e.printStackTrace();
    				}
    
    				String s = null;
    				while (true) {
    					try {
    						s = br.readLine();
    					} catch (IOException e) {
    						e.printStackTrace();
    					}
    
    					if (s.equals("bye")) {
    						try {
    							bw.write(s);
    							bw.flush();
    						} catch (Exception e) {
    							e.printStackTrace();
    						}
    
    						break;
    					}
    
    					try {
    						// 注意！此处如果不主动为输入的字符串加入\n换行符，则MINA服务器端收不到。
    						s = s + "\n";
    
    						bw.write(s);
    						bw.flush();
    					} catch (Exception e) {
    						e.printStackTrace();
    					}
    				}
    
    				try {
    					br.close();
    					socket.close();
    				} catch (Exception e) {
    					e.printStackTrace();
    				}
    			}
    		});
    	}
    }

先启动服务器端程序，程序运行后输出结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly96aGFuZ3BoaWwuYmxvZy5jc2RuLm5ldA_size_16_color_FFFFFF_t_70][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly96aGFuZ3BoaWwuYmxvZy5jc2RuLm5ldA_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/3453f3fa262b4283893e337f37b3181b.png