快捷導(dǎo)航

Java實(shí)現(xiàn)NIO聊天室的示例代碼(群聊+私聊)

更新時(shí)間：2021年05月11日 14:13:19 作者：馬走日mazouri

這篇文章主要介紹了Java實(shí)現(xiàn)NIO聊天室的示例代碼(群聊+私聊)，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧

功能介紹

功能:群聊+私發(fā)+上線提醒+下線提醒+查詢(xún)?cè)诰€用戶

在這里插入圖片描述

文件

Utils

需要用maven導(dǎo)入下面兩個(gè)包

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

/**
 * @author mazouri
 * @create 2021-05-09 22:26
 */
@Slf4j
public class Utils {
    /**
     * 將二進(jìn)制數(shù)據(jù)轉(zhuǎn)為對(duì)象
     *
     * @param buf
     * @return
     * @throws IOException
     * @throws ClassNotFoundException
     */
    public static Message decode(byte[] buf) throws IOException, ClassNotFoundException {
        ByteArrayInputStream bais = new ByteArrayInputStream(buf);
        ObjectInputStream ois = new ObjectInputStream(bais);
        return (Message) ois.readObject();
    }

    /**
     * 將對(duì)象轉(zhuǎn)為二進(jìn)制數(shù)據(jù)
     *
     * @param message
     * @return
     */
    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;
/**
 * @author mazouri
 * @create 2021-05-05 21:00
 */
public final class FinalValue {
    /**
     * 系統(tǒng)消息
     */
    public static final int MSG_SYSTEM = 0;
    /**
     * 群發(fā)消息
     */
    public static final int MSG_GROUP = 1;
    /**
     * 私發(fā)消息
     */
    public static final int MSG_PRIVATE = 2;
    /**
     * 客戶端請(qǐng)求在線人員
     */
    public static final int MSG_ONLINE = 3;
    /**
     * 客戶端將用戶名稱(chēng)發(fā)送給服務(wù)端
     */
    public static final int MSG_NAME = 4;
}

Message

package moremorechat_nio;

import java.io.Serializable;

/**
 * @author mazouri
 * @create 2021-05-05 21:00
 */
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.*;

/**
 * ctrl+f12 方法
 * ctrl+alt+左鍵
 * @author mazouri
 * @create 2021-05-09 19:24
 */
@Slf4j
public class NioServer {
    private Selector selector;
    private ServerSocketChannel ssc;

    public NioServer() {
        try {
            // 創(chuàng)建 selector, 管理多個(gè) channel
            selector = Selector.open();
            //打開(kāi)ServerSocketChannel，用于監(jiān)聽(tīng)客戶端的連接，它是所有客戶端連接的父通道
            ssc = ServerSocketChannel.open();
            ssc.bind(new InetSocketAddress(8888));
            //設(shè)置連接為非堵塞模式
            ssc.configureBlocking(false);
            // 2. 建立 selector 和 channel 的聯(lián)系（注冊(cè)）
            // SelectionKey 就是將來(lái)事件發(fā)生后，通過(guò)它可以知道事件和哪個(gè)channel的事件
            //將ServerSocketChannel注冊(cè)到Reactor線程的多路復(fù)用器Selector上，監(jiān)聽(tīng)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啟動(dòng)完成，等待用戶連接...");
        try {
            server.listen();
        } catch (Exception e) {
            log.debug("發(fā)生了一些問(wèn)題");
        }
    }

    /**
     * 監(jiān)聽(tīng)用戶的連接
     *
     * @throws Exception
     */
    private void listen() throws Exception {
        while (true) {
            // select 方法, 沒(méi)有事件發(fā)生，線程阻塞，有事件，線程才會(huì)恢復(fù)運(yùn)行, 通過(guò)Selector的select（）方法可以選擇已經(jīng)準(zhǔn)備就緒的通道 （這些通道包含你感興趣的的事件）
            //通過(guò)Selector的select（）方法可以選擇已經(jīng)準(zhǔn)備就緒的通道 （這些通道包含你感興趣的的事件）
            // select 在事件未處理時(shí)，它不會(huì)阻塞, 事件發(fā)生后要么處理，要么取消，不能置之不理
            selector.select();
            // 處理事件, selectedKeys 內(nèi)部包含了所有發(fā)生的事件
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                // 處理key 時(shí)，要從 selectedKeys 集合中刪除，否則下次處理就會(huì)有問(wèn)題
                iterator.remove();
                // 區(qū)分事件類(lèi)型
                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);
                }
            }
        }
    }

    /**
     * 處理讀事件
     *
     * @param key
     */
    private void dealReadEvent(SelectionKey key) {
        SocketChannel channel = null;
        try {
            channel = (SocketChannel) key.channel();
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            int read = channel.read(buffer);
            // 如果是正常斷開(kāi)，read 的方法的返回值是 -1
            if (read == -1) {
                //cancel 會(huì)取消注冊(cè)在 selector 上的 channel，并從 keys 集合中刪除 key 后續(xù)不會(huì)再監(jiān)聽(tīng)事件
                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);
            //取消注冊(cè)
            key.cancel();
            //關(guān)閉通道
            try {
                channel.close();
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }
        }
    }

    /**
     * 處理各種消息，并發(fā)送給客戶端
     *
     * @param msg
     * @param key
     * @param channel
     */
    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("收到一條系統(tǒng)消息: " + msg.message + "已上線"), sc);
                });
                break;
            case MSG_GROUP:
                getConnectedChannel(channel).forEach(selectionKey -> {
                    SocketChannel sc = (SocketChannel) selectionKey.channel();
                    sendMsgToClient(new Message(key.attachment() + "給大家發(fā)送了一條消息: " + 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() + "給你發(fā)送了一條消息: " + s[1]), sc);
                    flag.set(true);
                });
                if (!flag.get()){
                    sendMsgToClient(new Message(s[1]+"用戶不存在,請(qǐng)重新輸入！??！"), 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("收到一條系統(tǒng)消息: " + msg.message), sc);
                });
                break;
            default:
                break;
        }
    }

    /**
     * 發(fā)送消息給客戶端
     *
     * @param msg
     * @param sc
     */
    private void sendMsgToClient(Message msg, SocketChannel sc) {
        try {
            byte[] bytes = Utils.encode(msg);
            sc.write(ByteBuffer.wrap(bytes));
        } catch (IOException e) {
            log.debug("sendMsgToClient出現(xiàn)了一些問(wèn)題");
        }
    }

    /**
     * 獲取所有channel,除去調(diào)用者
     *
     * @param channel
     * @return
     */
    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.*;

/**
 * @author mazouri
 * @create 2021-04-29 12:02
 */
@Slf4j
public 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啟動(dòng)完成......");
        log.debug("請(qǐng)輸入你的名字完成注冊(cè)");
        input = new Scanner(System.in);
        username = input.next();
        log.debug("歡迎{}來(lái)到聊天系統(tǒng)", username);
    }

    public static void main(String[] args) throws IOException {
        System.out.println("tips: \n1. 直接發(fā)送消息會(huì)發(fā)給當(dāng)前的所有用戶 \n2. @用戶名:消息  會(huì)私發(fā)給你要發(fā)送的用戶 \n3. 輸入  查詢(xún)?cè)诰€用戶  會(huì)顯示當(dāng)前的在線用戶");
        NioClient client = new NioClient();

        //啟動(dòng)一個(gè)子線程接受服務(wù)器發(fā)送過(guò)來(lái)的消息
        new Thread(() -> {
            try {
                client.acceptMessageFromServer();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "receiveClientThread").start();

        //調(diào)用sendMessageToServer,發(fā)送消息到服務(wù)端
        client.sendMessageToServer();
    }

    /**
     * 將消息發(fā)送到服務(wù)端
     *
     * @throws IOException
     */
    private void sendMessageToServer() throws IOException {
        //先把用戶名發(fā)給客戶端
        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 ("查詢(xún)?cè)诰€用戶".equals(msgStr)) {
                msg = new Message(MSG_ONLINE, "請(qǐng)求在線人數(shù)");
            } else {
                msg = new Message(MSG_GROUP, msgStr);
            }

            byte[] bytes1 = Utils.encode(msg);
            socketChannel.write(ByteBuffer.wrap(bytes1));
        }
    }

    /**
     * 接受從服務(wù)器發(fā)送過(guò)來(lái)的消息
     */
    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));
                }
            }
        }
    }
}

到此這篇關(guān)于Java實(shí)現(xiàn)NIO聊天室的示例代碼(群聊+私聊)的文章就介紹到這了,更多相關(guān)Java NIO聊天室內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: