细谈 Linux 中的多路复用epoll - ToB企服应用市场:ToB评测及商务社交产业平台

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.*;
public class WGNioChatServer {
private final int port;
private Selector selector;
private ServerSocketChannel serverSocketChannel;
private final Map<SocketChannel, String> clientNames = new HashMap<>(); // 保存客户端名称
public WGNioChatServer(int port) {
this.port = port;
}
public void start() throws IOException {
// 初始化服务器通道和选择器
selector = Selector.open();
serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
serverSocketChannel.bind(new InetSocketAddress(port));
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
System.out.println("Chat server started on port " + port);
while (true) {
// 轮询准备就绪的事件
selector.select();
Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();
while (keyIterator.hasNext()) {
SelectionKey key = keyIterator.next();
keyIterator.remove();
if (key.isAcceptable()) {
handleAccept();
} else if (key.isReadable()) {
handleRead(key);
}
}
}
}
// 处理新客户端连接
private void handleAccept() throws IOException {
SocketChannel clientChannel = serverSocketChannel.accept();
clientChannel.configureBlocking(false);
clientChannel.register(selector, SelectionKey.OP_READ);
String clientAddress = clientChannel.getRemoteAddress().toString();
clientNames.put(clientChannel, clientAddress);
System.out.println("Connected: " + clientAddress);
broadcast("User " + clientAddress + " joined the chat", clientChannel);
}
// 读取客户端消息并广播给其他客户端
private void handleRead(SelectionKey key) throws IOException {
SocketChannel clientChannel = (SocketChannel) key.channel();
ByteBuffer buffer = ByteBuffer.allocate(256);
int bytesRead = clientChannel.read(buffer);
if (bytesRead == -1) {
// 客户端断开连接
String clientName = clientNames.get(clientChannel);
System.out.println("Disconnected: " + clientName);
clientNames.remove(clientChannel);
key.cancel();
clientChannel.close();
broadcast("User " + clientName + " left the chat", clientChannel);
return;
}
buffer.flip();
String message = new String(buffer.array(), 0, bytesRead);
System.out.println(clientNames.get(clientChannel) + ": " + message.trim());
broadcast(clientNames.get(clientChannel) + ": " + message, clientChannel);
}
// 向所有客户端广播消息
private void broadcast(String message, SocketChannel sender) throws IOException {
ByteBuffer buffer = ByteBuffer.wrap(message.getBytes());
for (SelectionKey key : selector.keys()) {
Channel targetChannel = key.channel();
if (targetChannel instanceof SocketChannel && targetChannel != sender) {
SocketChannel clientChannel = (SocketChannel) targetChannel;
clientChannel.write(buffer.duplicate());
}
}
}
public static void main(String[] args) throws IOException {
int port = 123456;
new WGNioChatServer(port).start();
}
}

复制代码

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.*;
import java.util.concurrent.*;
import com.fasterxml.jackson.databind.ObjectMapper;
public class EnhancedNioChatServer {
private final int port;
private Selector selector; // 多路复用器，负责管理多个通道
private ServerSocketChannel serverSocketChannel; // 服务器通道，用于监听客户端连接
private final Map<SocketChannel, String> clientNames = new HashMap<>(); // 存储客户端名称
private final Map<SocketChannel, Long> lastActiveTime = new ConcurrentHashMap<>(); // 存储客户端最后活动时间
private final ScheduledExecutorService heartbeatScheduler = Executors.newScheduledThreadPool(1); // 心跳检测定时任务
private final ExecutorService workerPool = Executors.newFixedThreadPool(10); // 处理客户端请求的线程池
private final ObjectMapper objectMapper = new ObjectMapper(); // 用于 JSON 序列化的对象
public EnhancedNioChatServer(int port) {
this.port = port;
}
public void start() throws IOException {
// 初始化服务器通道和选择器
selector = Selector.open();
serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false); // 配置非阻塞模式
serverSocketChannel.bind(new InetSocketAddress(port)); // 绑定端口
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT); // 注册连接接收事件
System.out.println("Chat server started on port " + port);
// 启动心跳检测任务
startHeartbeatCheck();
while (true) {
selector.select(); // 阻塞直到至少有一个事件发生
Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();
while (keyIterator.hasNext()) {
SelectionKey key = keyIterator.next();
keyIterator.remove(); // 防止重复处理
if (key.isAcceptable()) {
handleAccept(); // 处理客户端连接
} else if (key.isReadable()) {
handleRead(key); // 处理客户端的消息读取
}
}
}
}
// 处理新的客户端连接
private void handleAccept() throws IOException {
SocketChannel clientChannel = serverSocketChannel.accept(); // 接受新的客户端连接
clientChannel.configureBlocking(false); // 设置非阻塞模式
clientChannel.register(selector, SelectionKey.OP_READ); // 注册读事件
String clientAddress = clientChannel.getRemoteAddress().toString();
clientNames.put(clientChannel, clientAddress); // 保存客户端地址
lastActiveTime.put(clientChannel, System.currentTimeMillis()); // 记录最后活动时间
System.out.println("Connected: " + clientAddress);
broadcast(new Message("System", "User " + clientAddress + " joined the chat"), clientChannel);
}
// 处理读取客户端消息
private void handleRead(SelectionKey key) {
SocketChannel clientChannel = (SocketChannel) key.channel();
ByteBuffer buffer = ByteBuffer.allocate(256); // 缓冲区用于读取客户端数据
// 使用线程池处理，以免阻塞主线程
workerPool.submit(() -> {
try {
int bytesRead = clientChannel.read(buffer); // 读取客户端数据
if (bytesRead == -1) {
disconnect(clientChannel); // 客户端关闭连接
return;
}
lastActiveTime.put(clientChannel, System.currentTimeMillis()); // 更新最后活动时间
buffer.flip(); // 准备读取缓冲区内容
String messageContent = new String(buffer.array(), 0, bytesRead).trim();
Message message = new Message(clientNames.get(clientChannel), messageContent);
System.out.println(message.getSender() + ": " + message.getContent());
broadcast(message, clientChannel); // 广播消息给其他客户端
} catch (IOException e) {
disconnect(clientChannel); // 处理异常情况下的客户端断开
}
});
}
// 处理客户端断开连接
private void disconnect(SocketChannel clientChannel) {
try {
String clientName = clientNames.get(clientChannel);
System.out.println("Disconnected: " + clientName);
clientNames.remove(clientChannel); // 移除客户端信息
lastActiveTime.remove(clientChannel); // 移除最后活动时间
clientChannel.close(); // 关闭连接
broadcast(new Message("System", "User " + clientName + " left the chat"), clientChannel);
} catch (IOException e) {
e.printStackTrace();
}
}
// 广播消息给所有连接的客户端（除了消息发送者）
private void broadcast(Message message, SocketChannel sender) {
ByteBuffer buffer;
try {
buffer = ByteBuffer.wrap(objectMapper.writeValueAsBytes(message)); // 将消息序列化为 JSON
} catch (IOException e) {
e.printStackTrace();
return;
}
for (SelectionKey key : selector.keys()) {
Channel targetChannel = key.channel();
if (targetChannel instanceof SocketChannel && targetChannel != sender) { // 排除发送者
SocketChannel clientChannel = (SocketChannel) targetChannel;
try {
clientChannel.write(buffer.duplicate()); // 写入消息
} catch (IOException e) {
disconnect(clientChannel); // 处理写入失败的情况
}
}
}
}
// 定期检查客户端是否超时未响应，超时则断开连接
private void startHeartbeatCheck() {
heartbeatScheduler.scheduleAtFixedRate(() -> {
long currentTime = System.currentTimeMillis();
for (SocketChannel clientChannel : lastActiveTime.keySet()) {
long lastActive = lastActiveTime.get(clientChannel);
if (currentTime - lastActive > 60000) { // 如果超时 1 分钟
System.out.println("Client timeout: " + clientNames.get(clientChannel));
disconnect(clientChannel); // 断开超时客户端
}
}
}, 10, 30, TimeUnit.SECONDS); // 每隔 30 秒执行一次
}
public static void main(String[] args) throws IOException {
int port = 123456; // 定义端口号
new EnhancedNioChatServer(port).start(); // 启动服务器
}
// 用于封装消息的内部类
private static class Message {
private String sender;
private String content;
public Message(String sender, String content) {
this.sender = sender;
this.content = content;
}
public String getSender() {
return sender;
}
public String getContent() {
return content;
}
}
}

复制代码