Java怎么实现NIO聊天室

2023-06-15 01:06:01 474人浏览薄情痞子

摘要

这篇文章给大家分享的是有关Java怎么实现NIO聊天室的内容。小编觉得挺实用的，因此分享给大家做个参考，一起跟随小编过来看看吧。功能介绍功能:群聊+私发+上线提醒+下线提醒+查询在线用户文件Utils需要用maven导入下面两个包

这篇文章给大家分享的是有关Java怎么实现NIO聊天室的内容。小编觉得挺实用的，因此分享给大家做个参考，一起跟随小编过来看看吧。

功能介绍

功能:群聊+私发+上线提醒+下线提醒+查询在线用户

Java怎么实现NIO聊天室

文件

Utils

需要用maven导入下面两个包

 <dependency>            <groupId>org.projectlombok</groupId>            <artifactId>lombok</artifactId>            <version>1.16.18</version>        </dependency>        <dependency>            <groupId>ch.qos.logback</groupId>            <artifactId>logback-classic</artifactId>            <version>1.2.3</version>        </dependency>

package moremorechat_nio;import lombok.extern.slf4j.Slf4j;import java.io.*;@Slf4jpublic class Utils {        public static Message decode(byte[] buf) throws IOException, ClassNotFoundException {        ByteArrayInputStream bais = new ByteArrayInputStream(buf);        ObjectInputStream ois = new ObjectInputStream(bais);        return (Message) ois.readObject();    }        public static byte[] encode(Message message) throws IOException {        ByteArrayOutputStream baos = new ByteArrayOutputStream();        ObjectOutputStream oos = new ObjectOutputStream(baos);        oos.writeObject(message);        oos.flush();        return baos.toByteArray();    }}

FinalValue

package moremorechat_nio;public final class FinalValue {        public static final int MSG_SYSTEM = 0;        public static final int MSG_GROUP = 1;        public static final int MSG_PRIVATE = 2;        public static final int MSG_ONLINE = 3;        public static final int MSG_NAME = 4;}

Message

package moremorechat_nio;import java.io.Serializable;public class Message implements Serializable {    public int type;    public String message;    public Message() {    }    public Message(String message) {        this.message = message;    }    public Message(int type, String message) {        this.type = type;        this.message = message;    }    @Override    public String toString() {        return "Message{" +                "type=" + type +                ", message='" + message + '\'' +                '}';    }}

NiOServer

package moremorechat_nio;import lombok.extern.slf4j.Slf4j;import java.io.*;import java.net.InetSocketAddress;import java.nio.ByteBuffer;import java.nio.channels.*;import java.util.ArrayList;import java.util.Iterator;import java.util.Set;import java.util.concurrent.atomic.AtomicBoolean;import java.util.stream.Collectors;import static moremorechat_nio.FinalValue.*;@Slf4jpublic class NioServer {    private Selector selector;    private ServerSocketChannel ssc;    public NioServer() {        try {            // 创建 selector, 管理多个 channel            selector = Selector.open();            //打开ServerSocketChannel，用于监听客户端的连接，它是所有客户端连接的父通道            ssc = ServerSocketChannel.open();            ssc.bind(new InetSocketAddress(8888));            //设置连接为非堵塞模式            ssc.configureBlocking(false);            // 2. 建立 selector 和 channel 的联系（注册）            // SelectionKey 就是将来事件发生后，通过它可以知道事件和哪个channel的事件            //将ServerSocketChannel注册到Reactor线程的多路复用器Selector上，监听ACCEPT事件            ssc.reGISter(selector, SelectionKey.OP_ACCEPT);        } catch (IOException e) {            e.printStackTrace();        }    }    public static void main(String[] args) {        NioServer server = new NioServer();        log.debug("server启动完成，等待用户连接...");        try {            server.listen();        } catch (Exception e) {            log.debug("发生了一些问题");        }    }        private void listen() throws Exception {        while (true) {            // select 方法, 没有事件发生，线程阻塞，有事件，线程才会恢复运行, 通过Selector的select（）方法可以选择已经准备就绪的通道 （这些通道包含你感兴趣的的事件）            //通过Selector的select（）方法可以选择已经准备就绪的通道 （这些通道包含你感兴趣的的事件）            // select 在事件未处理时，它不会阻塞, 事件发生后要么处理，要么取消，不能置之不理            selector.select();            // 处理事件, selectedKeys 内部包含了所有发生的事件            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();            while (iterator.hasNext()) {                SelectionKey key = iterator.next();                // 处理key 时，要从 selectedKeys 集合中删除，否则下次处理就会有问题                iterator.remove();                // 区分事件类型                if (key.isAcceptable()) {                    ServerSocketChannel channel = (ServerSocketChannel) key.channel();                    SocketChannel sc = channel.accept();                    sc.configureBlocking(false);                    sc.register(selector, SelectionKey.OP_READ);                } else if (key.isReadable()) {                    dealReadEvent(key);                }            }        }    }        private void dealReadEvent(SelectionKey key) {        SocketChannel channel = null;        try {            channel = (SocketChannel) key.channel();            ByteBuffer buffer = ByteBuffer.allocate(1024);            int read = channel.read(buffer);            // 如果是正常断开，read 的方法的返回值是 -1            if (read == -1) {                //cancel 会取消注册在 selector 上的 channel，并从 keys 集合中删除 key 后续不会再监听事件                key.cancel();            } else {                buffer.flip();                Message msg = Utils.decode(buffer.array());                log.debug(msg.toString());                dealMessage(msg, key, channel);            }        } catch (IOException | ClassNotFoundException e) {            System.out.println((key.attachment() == null ? "匿名用户" : key.attachment()) + " 离线了..");            dealMessage(new Message(MSG_SYSTEM, key.attachment() + " 离线了.."), key, channel);            //取消注册            key.cancel();            //关闭通道            try {                channel.close();            } catch (IOException ioException) {                ioException.printStackTrace();            }        }    }        private void dealMessage(Message msg, SelectionKey key, SocketChannel channel) {        switch (msg.type) {            case MSG_NAME:                key.attach(msg.message);                log.debug("用户{}已上线", msg.message);                getConnectedChannel(channel).forEach(selectionKey -> {                    SocketChannel sc = (SocketChannel) selectionKey.channel();                    sendMsgToClient(new Message("收到一条系统消息: " + msg.message + "已上线"), sc);                });                break;            case MSG_GROUP:                getConnectedChannel(channel).forEach(selectionKey -> {                    SocketChannel sc = (SocketChannel) selectionKey.channel();                    sendMsgToClient(new Message(key.attachment() + "给大家发送了一条消息: " + msg.message), sc);                });                break;            case MSG_PRIVATE:                String[] s = msg.message.split("_");                AtomicBoolean flag = new AtomicBoolean(false);                getConnectedChannel(channel).stream().filter(sk -> s[0].equals(sk.attachment())).forEach(selectionKey -> {                    SocketChannel sc = (SocketChannel) selectionKey.channel();                    sendMsgToClient(new Message(key.attachment() + "给你发送了一条消息: " + s[1]), sc);                    flag.set(true);                });                if (!flag.get()){                    sendMsgToClient(new Message(s[1]+"用户不存在,请重新输入！！！"), channel);                }                break;            case MSG_ONLINE:                ArrayList<String> onlineList = new ArrayList<>();                onlineList.add((String) key.attachment());                getConnectedChannel(channel).forEach(selectionKey -> onlineList.add((String) selectionKey.attachment()));                sendMsgToClient(new Message(onlineList.toString()), channel);                break;            case MSG_SYSTEM:                getConnectedChannel(channel).forEach(selectionKey -> {                    SocketChannel sc = (SocketChannel) selectionKey.channel();                    sendMsgToClient(new Message("收到一条系统消息: " + msg.message), sc);                });                break;            default:                break;        }    }        private void sendMsgToClient(Message msg, SocketChannel sc) {        try {            byte[] bytes = Utils.encode(msg);            sc.write(ByteBuffer.wrap(bytes));        } catch (IOException e) {            log.debug("sendMsgToClient出现了一些问题");        }    }        private Set<SelectionKey> getConnectedChannel(SocketChannel channel) {        return selector.keys().stream()                .filter(item -> item.channel() instanceof SocketChannel && item.channel().isOpen() && item.channel() != channel)                .collect(Collectors.toSet());    }}

Nioclient

package moremorechat_nio;import lombok.extern.slf4j.Slf4j;import java.io.*;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;import static moremorechat_nio.FinalValue.*;@Slf4jpublic class NioClient {    private Selector selector;    private SocketChannel socketChannel;    private String username;    private static Scanner input;    public NioClient() throws IOException {        selector = Selector.open();        socketChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 8888));        socketChannel.configureBlocking(false);        socketChannel.register(selector, SelectionKey.OP_READ);        log.debug("client启动完成......");        log.debug("请输入你的名字完成注册");        input = new Scanner(System.in);        username = input.next();        log.debug("欢迎{}来到聊天系统", username);    }    public static void main(String[] args) throws IOException {        System.out.println("tips: \n1. 直接发送消息会发给当前的所有用户 \n2. @用户名:消息  会私发给你要发送的用户 \n3. 输入  查询在线用户  会显示当前的在线用户");        NioClient client = new NioClient();        //启动一个子线程接受服务器发送过来的消息        new Thread(() -> {            try {                client.acceptMessageFromServer();            } catch (Exception e) {                e.printStackTrace();            }        }, "receiveClientThread").start();        //调用sendMessageToServer,发送消息到服务端        client.sendMessageToServer();    }        private void sendMessageToServer() throws IOException {        //先把用户名发给客户端        Message message = new Message(MSG_NAME, username);        byte[] bytes = Utils.encode(message);        socketChannel.write(ByteBuffer.wrap(bytes));        while (input.hasNextLine()) {            String msgStr = input.next();            Message msg;            boolean isPrivate = msgStr.startsWith("@");            if (isPrivate) {                int idx = msgStr.indexOf(":");                String targetName = msgStr.substring(1, idx);                msgStr = msgStr.substring(idx + 1);                msg = new Message(MSG_PRIVATE, targetName + "_" + msgStr);            } else if ("查询在线用户".equals(msgStr)) {                msg = new Message(MSG_ONLINE, "请求在线人数");            } else {                msg = new Message(MSG_GROUP, msgStr);            }            byte[] bytes1 = Utils.encode(msg);            socketChannel.write(ByteBuffer.wrap(bytes1));        }    }        private void acceptMessageFromServer() throws Exception {        while (selector.select() > 0) {            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();            while (iterator.hasNext()) {                SelectionKey key = iterator.next();                iterator.remove();                if (key.isReadable()) {                    SocketChannel sc = (SocketChannel) key.channel();                    ByteBuffer buffer = ByteBuffer.allocate(1024);                    sc.read(buffer);                    Message message = Utils.decode(buffer.array());                    log.debug(String.valueOf(message.message));                }            }        }    }}