package com.dashu.netty.group_chat;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
public class GroupChatServer {
/**
* 初始化选择器
*/
private Selector selector;
/**
* 初始化服务器网络通道
*/
private ServerSocketChannel serverSocketChannel;
/**
* 端口
*/
private static final int PORT = 6666;
/**
* 构造方法
*/
public GroupChatServer() {
try {
//获取选择器
selector = Selector.open();
//获取服务器网络通道
serverSocketChannel = ServerSocketChannel.open();
//网络地址
InetSocketAddress inetSocketAddress = new InetSocketAddress(PORT);
//服务器网络通道绑定网络地址
serverSocketChannel.socket().bind(inetSocketAddress);
//设置服务器网络通道非阻塞
serverSocketChannel.configureBlocking(false);
//将服务器网络通道注册到选择器上，绑定连接请求事件
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* 监听客户端请求事件
*/
public void listen() {
try {
//无限循环
while (true) {
//获取请求数
int count = selector.select();
//count大于0，则代表有请求进来
if (count > 0) {
//获取请求集
Iterator<SelectionKey> selectionKeyIterator = selector.selectedKeys().iterator();
//遍历请求集
while (selectionKeyIterator.hasNext()) {
//得到请求
SelectionKey selectionKey = selectionKeyIterator.next();
//连接请求
if (selectionKey.isAcceptable()) {
//获取客户端网络通道
SocketChannel socketChannel = serverSocketChannel.accept();
//设置客户端网络通道非阻塞
socketChannel.configureBlocking(false);
//将客户端网络通道注册到选择器上
socketChannel.register(selector, SelectionKey.OP_READ);
System.out.println(socketChannel.getRemoteAddress() + "上线了");
}
//信息读取请求
if (selectionKey.isReadable()) {
//客户端信息读取
readData(selectionKey);
}
//移除请求
selectionKeyIterator.remove();
}
} else {
System.out.println("等待...");
}
}
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* 客户端信息读取
*
* @param selectionKey
*/
private void readData(SelectionKey selectionKey) {
//初始化客户端网络通道
SocketChannel socketChannel = null;
try {
//获取客户端网络通道
socketChannel = (SocketChannel) selectionKey.channel();
//创建缓冲区
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
//读取客户端网络通道中的数据到缓冲区
int count = socketChannel.read(byteBuffer);
//判断缓冲区中是否有数据
if (count > 0) {
//将缓冲区的数据转换位字符串
String message = new String(byteBuffer.array());
System.out.println(message.trim());
//将信息群发到其他客户端
sendInfoToOtClients(message, socketChannel);
}
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* 将信息群发到其他客户端
*
* @param message
* @param socketChannel
*/
private void sendInfoToOtClients(String message, SocketChannel socketChannel) {
//获取所有注册到选择器的客户端，并遍历
for (SelectionKey selectionKey : selector.keys()) {
//获取通道
Channel channel = selectionKey.channel();
//判断通道是否属于SocketChannel，同时不等于发送信息的客户端
if (channel instanceof SocketChannel && channel != socketChannel) {
//通道转换
SocketChannel sc = (SocketChannel) channel;
//将信息写入缓冲区
ByteBuffer byteBuffer = ByteBuffer.wrap(message.getBytes(StandardCharsets.UTF_8));
try {
//将缓冲区的数据写入通道
sc.write(byteBuffer);
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
public static void main(String[] args) {
GroupChatServer groupChatServer = new GroupChatServer();
System.out.println("服务器启动，开始监听客户端请求...");
groupChatServer.listen();
}
}
package com.dashu.netty.group_chat;
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.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.Scanner;
public class GroupChatClient {
/**
* 网络连接地址
*/
private final String HOST = "127.0.0.1";
/**
* 端口
*/
private final int PORT = 6666;
/**
* 初始化选择器
*/
private Selector selector;
/**
* 初始化网络通道
*/
private SocketChannel socketChannel;
/**
* 用户名
*/
private String username;
public GroupChatClient() {
try {
//获取选择器
selector = Selector.open();
//获取服务器网络地址
InetSocketAddress inetSocketAddress = new InetSocketAddress(HOST, PORT);
//获取网络通道
socketChannel = SocketChannel.open(inetSocketAddress);
//设置网络通道非阻塞
socketChannel.configureBlocking(false);
//将网络通道注册到选择器
socketChannel.register(selector, SelectionKey.OP_READ);
//获取用户名
System.out.println("请输入用户名：");
Scanner scanner = new Scanner(System.in);
username = scanner.nextLine();
System.out.println(username + " 进入群聊...");
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* 向服务器发送信息
*
* @param message
*/
public void sendInfo(String message) {
message = username + ":" + message;
try {
//向通道写入数据
socketChannel.write(ByteBuffer.wrap(message.getBytes(StandardCharsets.UTF_8)));
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* 读取服务器发来的信息
*/
public void readInfo() {
try {
//获取请求数
int count = selector.select();
if (count > 0) {
//获取请求集
Iterator<SelectionKey> selectionKeyIterator = selector.selectedKeys().iterator();
//遍历请求集
while (selectionKeyIterator.hasNext()) {
//获取请求
SelectionKey selectionKey = selectionKeyIterator.next();
//判断位读请求
if (selectionKey.isReadable()) {
//获取通道
SocketChannel sc = (SocketChannel) selectionKey.channel();
//创建缓冲区
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
//读取通道的数据到缓冲区
sc.read(byteBuffer);
//缓冲区数据转字符串
String message = new String(byteBuffer.array());
//输出
System.out.println(message.trim());
}
//移除已完成请求
selectionKeyIterator.remove();
}
}
} catch (Exception e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
GroupChatClient groupChatClient = new GroupChatClient();
/**
* 开启一个线程，每3秒读取一次服务器发来的信息
*/
new Thread() {
@Override
public void run() {
while (true) {
groupChatClient.readInfo();
try {
Thread.sleep(3000);
} catch (Exception e) {
e.printStackTrace();
}
}
}
}.start();
//信息输入
Scanner scanner = new Scanner(System.in);
System.out.println("请输入信息：");
while (scanner.hasNextLine()) {
String s = scanner.nextLine();
//信息发送
groupChatClient.sendInfo(s);
System.out.println("请输入信息：");
}
}
}
1、服务器
2、客户端01
3、客户端02
