选择器Selector案例

本文目录导读：

1. 完整的Selector案例：简易聊天服务器
2. 使用说明

我来为你展示一个完整的Java NIO Selector（选择器）案例，这是一个使用非阻塞I/O实现的多路复用服务器。

完整的Selector案例：简易聊天服务器

服务器端代码

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
/**
 * 基于Selector的多路复用聊天服务器
 * 可以同时处理多个客户端的连接和消息
 */
public class ChatServer {
    // 选择器
    private Selector selector;
    // 服务器通道
    private ServerSocketChannel serverChannel;
    // 存储所有连接的客户端
    private static ConcurrentHashMap<SocketChannel, String> clients = new ConcurrentHashMap<>();
    // 服务器端口
    private static final int PORT = 8888;
    public ChatServer() {
        try {
            // 1. 打开选择器
            selector = Selector.open();
            // 2. 打开服务器通道
            serverChannel = ServerSocketChannel.open();
            // 3. 设置为非阻塞模式
            serverChannel.configureBlocking(false);
            // 4. 绑定端口
            serverChannel.socket().bind(new InetSocketAddress(PORT));
            // 5. 注册到选择器，关注OP_ACCEPT事件
            serverChannel.register(selector, SelectionKey.OP_ACCEPT);
            System.out.println("服务器启动，监听端口：" + PORT);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    /**
     * 启动服务器 - 核心事件循环
     */
    public void start() {
        try {
            while (true) {
                // 核心方法：阻塞等待就绪的事件
                // select() 会阻塞，直到至少有一个通道的事件就绪
                int readyCount = selector.select();
                if (readyCount == 0) {
                    continue;
                }
                // 获取所有就绪的SelectionKey
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectedKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    // 处理完当前key后必须移除，防止重复处理
                    iterator.remove();
                    // 处理事件
                    try {
                        if (key.isAcceptable()) {
                            // 新连接事件
                            handleAccept(key);
                        } else if (key.isReadable()) {
                            // 可读事件
                            handleRead(key);
                        } else if (key.isWritable()) {
                            // 可写事件
                            handleWrite(key);
                        }
                    } catch (IOException e) {
                        // 客户端断开连接
                        closeConnection(key);
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    /**
     * 处理新的客户端连接
     */
    private void handleAccept(SelectionKey key) throws IOException {
        ServerSocketChannel server = (ServerSocketChannel) key.channel();
        // 接受客户端连接
        SocketChannel clientChannel = server.accept();
        // 设置为非阻塞模式
        clientChannel.configureBlocking(false);
        // 为客户端生成名称
        String clientName = "Client-" + clientChannel.hashCode();
        // 注册该客户端到选择器，关注读事件
        clientChannel.register(selector, SelectionKey.OP_READ);
        // 保存客户端信息
        clients.put(clientChannel, clientName);
        System.out.println(clientName + " 已连接");
        // 通知其他客户端
        broadcast("【系统】" + clientName + " 加入了聊天室", null);
    }
    /**
     * 处理客户端发来的消息
     */
    private void handleRead(SelectionKey key) throws IOException {
        SocketChannel clientChannel = (SocketChannel) key.channel();
        // 创建缓冲区
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        // 读取数据
        int bytesRead = clientChannel.read(buffer);
        if (bytesRead == -1) {
            // 客户端关闭了连接
            closeConnection(key);
            return;
        }
        // 切换为读模式
        buffer.flip();
        // 读取内容
        byte[] data = new byte[buffer.remaining()];
        buffer.get(data);
        String message = new String(data, StandardCharsets.UTF_8).trim();
        String clientName = clients.get(clientChannel);
        System.out.println(clientName + " 发送：" + message);
        // 广播给所有客户端
        broadcast(clientName + "：" + message, clientChannel);
    }
    /**
     * 处理写事件（本例主要用于演示）
     */
    private void handleWrite(SelectionKey key) throws IOException {
        SocketChannel channel = (SocketChannel) key.channel();
        // 这里简化为：取消写关注
        key.interestOps(key.interestOps() & ~SelectionKey.OP_WRITE);
        // 实际项目中，这里会处理待发送的数据队列
    }
    /**
     * 广播消息给所有客户端
     */
    private void broadcast(String message, SocketChannel exceptChannel) {
        ByteBuffer buffer = ByteBuffer.wrap(message.getBytes(StandardCharsets.UTF_8));
        for (SocketChannel client : clients.keySet()) {
            if (client != exceptChannel) {
                try {
                    // 将缓冲区重置到读位置
                    buffer.rewind();
                    client.write(buffer);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    /**
     * 关闭连接
     */
    private void closeConnection(SelectionKey key) {
        try {
            SocketChannel channel = (SocketChannel) key.channel();
            String clientName = clients.remove(channel);
            if (clientName != null) {
                System.out.println(clientName + " 断开连接");
                broadcast("【系统】" + clientName + " 离开了聊天室", null);
            }
            // 取消注册
            key.cancel();
            // 关闭通道
            channel.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    public static void main(String[] args) {
        ChatServer server = new ChatServer();
        server.start();
    }
}

客户端代码

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;
/**
 * 聊天客户端
 */
public class ChatClient {
    private SocketChannel channel;
    public ChatClient(String host, int port) {
        try {
            // 打开客户端通道
            channel = SocketChannel.open(new InetSocketAddress(host, port));
            // 设置阻塞模式
            channel.configureBlocking(false);
            System.out.println("已连接到服务器");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    /**
     * 启动客户端
     */
    public void start() {
        // 启动线程接收服务器消息
        new Thread(this::receiveMessage).start();
        // 主线程用于发送消息
        sendMessage();
    }
    /**
     * 接收消息线程
     */
    private void receiveMessage() {
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        try {
            while (true) {
                // 清除缓冲区
                buffer.clear();
                // 读取数据
                int bytesRead = channel.read(buffer);
                if (bytesRead > 0) {
                    buffer.flip();
                    byte[] data = new byte[buffer.remaining()];
                    buffer.get(data);
                    String message = new String(data, StandardCharsets.UTF_8);
                    System.out.println(message);
                }
                Thread.sleep(100);
            }
        } catch (IOException | InterruptedException e) {
            System.out.println("连接已断开");
            e.printStackTrace();
        }
    }
    /**
     * 发送消息
     */
    private void sendMessage() {
        Scanner scanner = new Scanner(System.in);
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        try {
            while (true) {
                System.out.print("请输入消息（输入exit退出）：");
                String message = scanner.nextLine();
                if ("exit".equalsIgnoreCase(message)) {
                    channel.close();
                    break;
                }
                // 发送消息
                buffer.clear();
                buffer.put(message.getBytes(StandardCharsets.UTF_8));
                buffer.flip();
                while (buffer.hasRemaining()) {
                    channel.write(buffer);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            scanner.close();
        }
    }
    public static void main(String[] args) {
        ChatClient client = new ChatClient("localhost", 8888);
        client.start();
    }
}

测试示例代码

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.net.InetSocketAddress;
import java.util.Iterator;
/**
 * 简单Selector测试示例
 * 演示Selector的基本使用
 */
public class SelectorDemo {
    public static void main(String[] args) throws IOException {
        // 1. 创建Selector
        Selector selector = Selector.open();
        // 2. 创建ServerSocketChannel
        ServerSocketChannel serverChannel = ServerSocketChannel.open();
        serverChannel.configureBlocking(false);
        serverChannel.bind(new InetSocketAddress(8080));
        // 3. 注册到Selector，关注ACCEPT事件
        SelectionKey serverKey = serverChannel.register(selector, SelectionKey.OP_ACCEPT);
        System.out.println("服务器启动，等待连接...");
        // 4. 事件循环
        while (true) {
            // 阻塞等待，直到有事件发生
            int readyChannels = selector.select();
            if (readyChannels == 0) {
                continue;
            }
            // 获取就绪的事件集合
            Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();
            while (keyIterator.hasNext()) {
                SelectionKey key = keyIterator.next();
                // 处理完必须移除
                keyIterator.remove();
                if (key.isAcceptable()) {
                    // 有新的连接
                    ServerSocketChannel server = (ServerSocketChannel) key.channel();
                    SocketChannel clientChannel = server.accept();
                    clientChannel.configureBlocking(false);
                    // 注册读事件
                    clientChannel.register(selector, SelectionKey.OP_READ);
                    System.out.println("新的客户端连接: " + clientChannel.getRemoteAddress());
                    // 发送欢迎消息
                    ByteBuffer buffer = ByteBuffer.wrap("欢迎连接服务器!\n".getBytes());
                    clientChannel.write(buffer);
                } else if (key.isReadable()) {
                    // 客户端发送了数据
                    SocketChannel clientChannel = (SocketChannel) key.channel();
                    ByteBuffer buffer = ByteBuffer.allocate(256);
                    int bytesRead = clientChannel.read(buffer);
                    if (bytesRead > 0) {
                        buffer.flip();
                        byte[] data = new byte[buffer.remaining()];
                        buffer.get(data);
                        String message = new String(data);
                        System.out.println("收到消息: " + message.trim());
                        // 回显消息
                        ByteBuffer response = ByteBuffer.wrap(("服务器收到: " + message).getBytes());
                        clientChannel.write(response);
                    } else if (bytesRead == -1) {
                        // 客户端断开
                        clientChannel.close();
                        System.out.println("客户端断开连接");
                    }
                }
            }
        }
    }
}

使用说明

运行步骤：

1. 先运行服务器：

   javac ChatServer.java
   java ChatServer

2. 运行多个客户端：

   javac ChatClient.java
   java ChatClient

3. 开始聊天：多个客户端可以互相发送消息

关键概念解析：

优点：

• ✅ 单线程可处理多个连接
• ✅ 高并发性能
• ✅ 节省系统资源
• ✅ 非阻塞I/O模型

这个案例展示了Selector在实际应用中的完整用法,你可以根据需求自由修改和扩展。