功能:群聊+私發(fā)+上線提醒+下線提醒+查詢在線用戶

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

<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 */@Slf4jpublic class Utils { /** * 將二進(jìn)制數(shù)據(jù)轉(zhuǎn)為對象 * * @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(); } /** * 將對象轉(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; /** * 客戶端請求在線人員 */ public static final int MSG_ONLINE = 3; /** * 客戶端將用戶名稱發(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 */@Slf4jpublic class NioServer { private Selector selector; private ServerSocketChannel ssc; public NioServer() {try { // 創(chuàng)建 selector, 管理多個 channel selector = Selector.open(); //打開ServerSocketChannel，用于監(jiān)聽客戶端的連接，它是所有客戶端連接的父通道 ssc = ServerSocketChannel.open(); ssc.bind(new InetSocketAddress(8888)); //設(shè)置連接為非堵塞模式 ssc.configureBlocking(false); // 2. 建立 selector 和 channel 的聯(lián)系（注冊） // SelectionKey 就是將來事件發(fā)生后，通過它可以知道事件和哪個channel的事件 //將ServerSocketChannel注冊到Reactor線程的多路復(fù)用器Selector上，監(jiān)聽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('發(fā)生了一些問題');} } /** * 監(jiān)聽用戶的連接 * * @throws Exception */ private void listen() throws Exception {while (true) { // select 方法, 沒有事件發(fā)生，線程阻塞，有事件，線程才會恢復(fù)運行, 通過Selector的select（）方法可以選擇已經(jīng)準(zhǔn)備就緒的通道 （這些通道包含你感興趣的的事件） //通過Selector的select（）方法可以選擇已經(jīng)準(zhǔn)備就緒的通道 （這些通道包含你感興趣的的事件） // select 在事件未處理時，它不會阻塞, 事件發(fā)生后要么處理，要么取消，不能置之不理 selector.select(); // 處理事件, selectedKeys 內(nèi)部包含了所有發(fā)生的事件 Iterator<SelectionKey> iterator = selector.selectedKeys().iterator(); while (iterator.hasNext()) {SelectionKey key = iterator.next();// 處理key 時，要從 selectedKeys 集合中刪除，否則下次處理就會有問題iterator.remove();// 區(qū)分事件類型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); // 如果是正常斷開，read 的方法的返回值是 -1 if (read == -1) {//cancel 會取消注冊在 selector 上的 channel，并從 keys 集合中刪除 key 后續(xù)不會再監(jiān)聽事件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(); //關(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]+'用戶不存在,請重新輸入！！！'), 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)了一些問題');} } /** * 獲取所有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 */@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('歡迎{}來到聊天系統(tǒng)', username); } public static void main(String[] args) throws IOException {System.out.println('tips: n1. 直接發(fā)送消息會發(fā)給當(dāng)前的所有用戶 n2. @用戶名:消息 會私發(fā)給你要發(fā)送的用戶 n3. 輸入 查詢在線用戶 會顯示當(dāng)前的在線用戶');NioClient client = new NioClient();//啟動一個子線程接受服務(wù)器發(fā)送過來的消息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 ('查詢在線用戶'.equals(msgStr)) {msg = new Message(MSG_ONLINE, '請求在線人數(shù)'); } else {msg = new Message(MSG_GROUP, msgStr); } byte[] bytes1 = Utils.encode(msg); socketChannel.write(ByteBuffer.wrap(bytes1));} } /** * 接受從服務(wù)器發(fā)送過來的消息 */ 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實現(xiàn)NIO聊天室的示例代碼(群聊+私聊)的文章就介紹到這了,更多相關(guān)Java NIO聊天室內(nèi)容請搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持好吧啦網(wǎng)！