demo要求:
1)编写一个NIO群聊系统,实现服务器端和客户端之间的数据简单通讯(非阻塞)
2)实现多人群聊
3)服务器端:可以监测用户上线,离线,并实现消息转发功能。
4)客户端:通过channel可以无阻塞发送消息给其他所有用户(客户端),同时可以接受其他用户发送的消息(由服务器转发得到)
5)目的:进一步理解NIO非阻塞网络编程机制。
以下代码:
服务器端实现的代码:
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注册到selector
this.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()) {
//取出selectionkey
SelectionKey key = iterator.next();
//监听到accept
if(key.isAcceptable()) {
SocketChannel sc = listenChannel.accept();
sc.configureBlocking(false);
//将sc注册到Selector
sc.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) {
//定义一个SocketChannel
SocketChannel channel = null;
try {
//得到channel
channel = (SocketChannel) key.channel();
//创建buffer
ByteBuffer 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,并排除seflChannel
for(SelectionKey key: selector.keys()){
//通过key 取出对应的SocketChannel
Channel targetChannel = key.channel();
//排除自己
if(targetChannel instanceof SocketChannel && targetChannel != selfChannel ){
//转型
SocketChannel dest = (SocketChannel) targetChannel;
//将数据存储到buffer。写入buffer
ByteBuffer 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注册到selector
sc.register(selector, SelectionKey.OP_READ);
//得到username
userName = 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();
//得到一个Buffer
ByteBuffer 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,多开几个客户端实现。实现如下截图:
若有问题请留言。