Java实现NIO聊天室的示例代码(群聊+私聊)
作者:马走日mazouri 发布时间:2023-11-29 00:57:45
标签: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.*;
/**
* @author mazouri
* @create 2021-05-09 22:26
*/
@Slf4j
public class Utils {
/**
* 将二进制数据转为对象
*
* @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();
}
/**
* 将对象转为二进制数据
*
* @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 {
/**
* 系统消息
*/
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;
/**
* @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 {
// 创建 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("发生了一些问题");
}
}
/**
* 监听用户的连接
*
* @throws Exception
*/
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);
}
}
}
}
/**
* 处理读事件
*
* @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 后续不会再监听事件
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();
}
}
}
/**
* 处理各种消息,并发送给客户端
*
* @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("收到一条系统消息: " + 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;
}
}
/**
* 发送消息给客户端
*
* @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出现了一些问题");
}
}
/**
* 获取所有channel,除去调用者
*
* @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启动完成......");
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();
}
/**
* 将消息发送到服务端
*
* @throws IOException
*/
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));
}
}
}
}
}
来源:https://blog.csdn.net/weixin_46215617/article/details/116613948
0
投稿
猜你喜欢
- 第1部分 ArrayList介绍ArrayList简介ArrayList 是一个数组队列,相当于 动态数组。与Java中的数组相比,它的容量
- 最近重新研究Java基础知识,发现以前太多知识知识略略带过了,比较说Java的线程机制,在Java中有两类线程:User Thread(用户
- 1 关于自动内存管理Java是由jvm来管理内存,包括自动分配以及自动回收,因此它不容易出现内存泄漏和内存溢出问题。C/C++,由程序员手动
- 如何将Object类型转换为int类型Object object = null;try{ Integer.pars
- 前言:文件的上传和下载在日常开发中很是常见,那么这一功能是如何实现的呢,下面我给大家介绍一下实现条件:1、需要一个form标签,method
- 对openfeign不清楚的同学可以参考下我的这篇文章:springboot~openfeign从此和httpClient说再见对于open
- 一、Java内存区域方法区(公有):用户存储已被虚拟机加载的类信息,常量,静态常量,即时编译器编译后的代码等数据。异常状态 OutOfMem
- 1. Ajax 概述Ajax 的英文全称是 ”Asynchronous JavaScript and XML&l
- 干java 开发这么多年, 之前一直没留意java 进程还区分守护进程和用户进程。守护进程这个概念最早还是在linux系统中接触的,直到近期
- 本文实例讲述了java中Object类用法。分享给大家供大家参考。具体如下:1、Object类是所有java类的基类如果在类的声明中未使用e
- 前言目前正在练手springboot+vue,因为很多步骤会遇到困难,当时查完资料解决,过一段时间就会忘记,所以决定建个系列记录下来。因为中
- 1. 编写索引内容节点解释:settings:配置信息"number_of_replicas": 0 不需要备份(单节点
- 一般而言,Android 应用在请求数据时都是以 Get 或 Post 等方式向远程服务器发起请求,那你有没有想过其实我们也可以在 Andr
- 如何配置 * step1: 自定义 * /** * 自定义 * */public class MyInterceptor implemen
- 概述:Flutter 标签类控件大全ChipFlutter内置了多个标签类控件,但本质上它们都是同一个控件,只不过是属性参数不同而已,在学习
- 以下摘自胖哥分享的 2022开工福利教程。在学习Spring Security的时候你有没有下面这两个疑问:Spring Security的
- 首先引入pom <!--SpringBoot 2.1.0--> <parent>  
- 一、前言在Spring中,事务有两种实现方式:编程式事务管理: 编程式事务管理使用TransactionTemplate可实现更细
- 一、简介Android的消息机制主要是指Handler的运行机制,那么什么是Handler的运行机制那?通俗的来讲就是,使用Handler将
- ErrorPageFilter的实际应用Spring框架错误页过滤器springboot提供了一个ErrorPageFilter,用来处理当