java NIO群聊系统

demo要求：

1）编写一个NIO群聊系统，实现服务器端和客户端之间的数据简单通讯（非阻塞）

2）实现多人群聊

3）服务器端：可以监测用户上线，离线，并实现消息转发功能。

4）客户端：通过channel可以无阻塞发送消息给其他所有用户(客户端)，同时可以接受其他用户发送的消息（由服务器转发得到）

5）目的：进一步理解NIO非阻塞网络编程机制。

从Netty的Reactor模式看demo是单Reactor单线程模式。其实Netty是在NIO1.0的基础上封装了复杂的调用操作，解决了JDK1.6的NIO臭名昭著的Epoll Bug，方便程序员进行网络编程，提升效率。

以下代码：

服务器端实现的代码：

package com.tfq.netty.nio.groupchat;import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;/*** @author: fqtang* @date: 2024/03/19/11:22* @description: 服务器端*/
public class GroupChatServer {//定义属性private ServerSocketChannel listenChannel;private Selector selector;private static final int PORT = 6667;//构造器public GroupChatServer() {try {//得到选择器this.selector = Selector.open();//获取监听通道this.listenChannel = ServerSocketChannel.open();//绑定端口this.listenChannel.socket().bind(new InetSocketAddress(PORT));//设置通道为非阻塞this.listenChannel.configureBlocking(false);//将该listenChannel注册到selectorthis.listenChannel.register(selector, SelectionKey.OP_ACCEPT);} catch(IOException e) {e.printStackTrace();}}/*** 监听*/public void listen(){try {//循环处理while(true) {int count = selector.select();if(count > 0) {//有事件处理//遍历得到selectionKey集合Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();while(iterator.hasNext()) {//取出selectionkeySelectionKey key = iterator.next();//监听到acceptif(key.isAcceptable()) {SocketChannel sc = listenChannel.accept();sc.configureBlocking(false);//将sc注册到Selectorsc.register(selector, SelectionKey.OP_READ);System.out.println(sc.getRemoteAddress() + " 上线");}if(key.isReadable()) {//通道发送read事件，即通道是可读的状态//处理读readDate(key);}//当前的Key删除，防止重复处理iterator.remove();}}}} catch(IOException e) {e.printStackTrace();} finally {try {listenChannel.close();} catch(IOException e) {throw new RuntimeException(e);}}}/*** 读取客户端的消息*/private void readDate(SelectionKey key) {//定义一个SocketChannelSocketChannel channel = null;try {//得到channelchannel = (SocketChannel) key.channel();//创建bufferByteBuffer buffer = ByteBuffer.allocate(1024);int count = channel.read(buffer);if(count > 0) {//把缓冲区的数据转成字符串String msg = new String(buffer.array());System.out.println("from 客户端发送的消息：" + msg);//向其他的客户端转发消息sendMsgToOtherClients(channel, msg);}} catch(Exception e) {try {System.out.println(channel.getRemoteAddress() + "已离线了");//取消注册key.cancel();//关闭通道channel.close();} catch(IOException ex) {throw new RuntimeException(ex);}}}/*** 转发消息给其他通道,排除自己* @param selfChannel* @param msg*/private void sendMsgToOtherClients(SocketChannel selfChannel,String msg) throws IOException {System.out.println("服务器转发消息中.....");//遍历 所有注册到selector 上的SocketChannel，并排除seflChannelfor(SelectionKey key: selector.keys()){//通过key 取出对应的SocketChannelChannel targetChannel = key.channel();//排除自己if(targetChannel instanceof  SocketChannel && targetChannel != selfChannel ){//转型SocketChannel dest = (SocketChannel) targetChannel;//将数据存储到buffer。写入bufferByteBuffer byteBuffer = ByteBuffer.wrap(msg.getBytes());//将buffer 的数据写入通道dest.write(byteBuffer);}}}public static void main(String[] args) {GroupChatServer groupChatServer = new GroupChatServer();groupChatServer.listen();}}

客户端代码如下：

package com.tfq.netty.nio.groupchat;import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Scanner;/*** @author: fqtang* @date: 2024/03/19/14:14* @description: 描述*/
public class GroupChatClient {//定义属性private final String HOST = "127.0.0.1";private final int PORT = 6667;private SocketChannel sc;private Selector selector;private String userName;public GroupChatClient() throws IOException {selector = Selector.open();//连接服务器sc = sc.open(new InetSocketAddress(HOST, PORT));//设置非阻塞sc.configureBlocking(false);//将channel注册到selectorsc.register(selector, SelectionKey.OP_READ);//得到usernameuserName = sc.getLocalAddress().toString().substring(1);System.out.println(userName + " is ok.....");}/*** 向服务器发送消息** @param info*/public void sendMsg(String info) {info = userName + " 说: " + info;try {sc.write(ByteBuffer.wrap(info.getBytes()));} catch(IOException e) {e.printStackTrace();}}/*** 读取服务器端的数据*/public void readMsg() {try {int readChannel = selector.select();if(readChannel > 0) {//有可用通道Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();while(iterator.hasNext()) {SelectionKey key = iterator.next();if(key.isReadable()) {//得到相关通道SocketChannel socketChannel = (SocketChannel) key.channel();//得到一个BufferByteBuffer buffer = ByteBuffer.allocate(1024);//读取socketChannel.read(buffer);System.out.println("读取数据：" + new String(buffer.array()));}}//删除当前的selectionKey,防止重复操作,若不清空，其他客户端收到不到最新消息数据iterator.remove();}} catch(IOException e) {e.printStackTrace();}}public static void main(String[] args) throws IOException {GroupChatClient groupChatClient = new GroupChatClient();//启动一个线程,每隔3秒读取发送的数据new Thread() {public void run() {while(true){groupChatClient.readMsg();try {Thread.sleep(3000);} catch(InterruptedException e) {e.printStackTrace();}}}}.start();//发送数据给服务器端Scanner scanner =  new Scanner(System.in);while(scanner.hasNextLine()){String s = scanner.nextLine();groupChatClient.sendMsg(s);}}}

通过idea运行GroupChatClient.java，多开几个客户端实现。实现如下截图：

总结：服务器端用一个线程通过多路复用器搞定所有的IO操作（包括连接、读、写等），编码简单，清晰明了，但是如果客户端连接数量较多，将无法支撑。要使用单Reactor多线程解决。

若有问题请留言。