并发设计模式实战系列(18)：反应器（Reactor）

import java.nio.*;
import java.nio.channels.*;
import java.util.*;
public class Reactor implements Runnable {
    final Selector selector;
    final ServerSocketChannel serverSocket;
    // 启动方法
    public Reactor(int port) throws Exception {
        selector = Selector.open();
        serverSocket = ServerSocketChannel.open();
        serverSocket.socket().bind(new InetSocketAddress(port));
        serverSocket.configureBlocking(false);
        SelectionKey sk = serverSocket.register(selector, SelectionKey.OP_ACCEPT);
        sk.attach(new Acceptor());
    }
    // 事件循环核心
    public void run() {
        try {
            while (!Thread.interrupted()) {
                selector.select();
                Set<SelectionKey> selected = selector.selectedKeys();
                Iterator<SelectionKey> it = selected.iterator();
                while (it.hasNext()) {
                    dispatch(it.next());
                }
                selected.clear();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    void dispatch(SelectionKey key) {
        Runnable r = (Runnable) key.attachment();
        if (r != null) r.run();
    }
    // 连接处理器
    class Acceptor implements Runnable {
        public void run() {
            try {
                SocketChannel c = serverSocket.accept();
                if (c != null) new Handler(selector, c);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
// 业务处理器
class Handler implements Runnable {
    final SocketChannel socket;
    final SelectionKey sk;
    ByteBuffer input = ByteBuffer.allocate(1024);
    Handler(Selector sel, SocketChannel c) throws Exception {
        socket = c;
        c.configureBlocking(false);
        sk = socket.register(sel, 0);
        sk.attach(this);
        sk.interestOps(SelectionKey.OP_READ);
        sel.wakeup(); // 唤醒selector
    }
    public void run() {
        try {
            if (sk.isReadable()) read();
            else if (sk.isWritable()) write();
        } catch (Exception e) {
            close();
        }
    }
    void read() throws Exception {
        socket.read(input);
        if (inputIsComplete()) {
            process();
            sk.interestOps(SelectionKey.OP_WRITE);
        }
    }
    void write() throws Exception {
        if (outputIsComplete()) sk.cancel();
    }
    boolean inputIsComplete() { /* 解析协议头判断 */ return true; }
    boolean outputIsComplete() { /* 判断写入完成 */ return true; }
    void process() { /* 业务处理逻辑 */ }
    void close() { /* 资源释放 */ }
}
// 启动类
class Main {
    public static void main(String[] args) throws Exception {
        new Thread(new Reactor(8080)).start();
    }
}

// 重要参数调优
serverSocket.setOption(StandardSocketOptions.SO_REUSEADDR, true);
socket.setOption(StandardSocketOptions.TCP_NODELAY, true);
// Buffer分配策略
ByteBuffer.allocateDirect(1024); // 直接内存，减少拷贝

// 主从Reactor配置
Selector[] selectors = new Selector[Runtime.getRuntime().availableProcessors()];
for (int i = 0; i < selectors.length; i++) {
    selectors[i] = Selector.open();
    new Thread(new SubReactor(selectors[i])).start();
}

/ FileChannel.transferTo实现零拷贝
fileChannel.transferTo(position, count, socketChannel);

// 使用Netty的ByteBuf内存池
ByteBufAllocator alloc = PooledByteBufAllocator.DEFAULT;
ByteBuf buffer = alloc.directBuffer(1024);

// 关键指标采集
selector.keys().size();      // 注册通道数
selector.selectNow();        // 就绪事件数
bufferPool.usedMemory();     // 内存使用量

// 主从Reactor线程组实现
class MasterSlaveReactor {
    private final Selector masterSelector;
    private final Selector[] slaveSelectors;
    private final ExecutorService masterExecutor;
    private final ExecutorService[] slaveExecutors;
    public MasterSlaveReactor(int slaveCount) throws Exception {
        // 主Reactor负责连接接入
        masterSelector = Selector.open();
        masterExecutor = Executors.newSingleThreadExecutor(r -> new Thread(r, "Master-Reactor"));
        
        // 从Reactor负责IO读写
        slaveSelectors = new Selector[slaveCount];
        slaveExecutors = new ExecutorService[slaveCount];
        for (int i = 0; i < slaveCount; i++) {
            slaveSelectors[i] = Selector.open();
            slaveExecutors[i] = Executors.newSingleThreadExecutor(r -> new Thread(r, "Slave-Reactor-" + i));
        }
    }
    public void start() {
        // 主Reactor启动
        masterExecutor.execute(() -> {
            while (!Thread.interrupted()) {
                try {
                    masterSelector.select();
                    Set<SelectionKey> keys = masterSelector.selectedKeys();
                    Iterator<SelectionKey> it = keys.iterator();
                    while (it.hasNext()) {
                        SelectionKey key = it.next();
                        if (key.isAcceptable()) {
                            // 分配连接到从Reactor
                            int slaveIndex = key.attachment().hashCode() % slaveSelectors.length;
                            dispatchToSlave(slaveIndex, key);
                        }
                        it.remove();
                    }
                } catch (Exception e) { /* 处理异常 */ }
            }
        });
        // 从Reactor启动
        for (int i = 0; i < slaveSelectors.length; i++) {
            final int index = i;
            slaveExecutors[i].execute(() -> {
                while (!Thread.interrupted()) {
                    try {
                        slaveSelectors[index].select();
                        Set<SelectionKey> keys = slaveSelectors[index].selectedKeys();
                        // 处理IO读写（同基础版Handler逻辑）
                    } catch (Exception e) { /* 处理异常 */ }
                }
            });
        }
    }
}

// 基于FileChannel的零拷贝传输
void sendFile(SocketChannel channel, File file) throws Exception {
    try (FileInputStream fis = new FileInputStream(file)) {
        FileChannel fileChannel = fis.getChannel();
        long position = 0;
        long remaining = fileChannel.size();
        while (remaining > 0) {
            long transferred = fileChannel.transferTo(position, remaining, channel);
            position += transferred;
            remaining -= transferred;
        }
    }
}
// 内存映射解析大文件
ByteBuffer mapFile(String path) throws Exception {
    try (RandomAccessFile raf = new RandomAccessFile(path, "r")) {
        return raf.getChannel().map(FileChannel.MapMode.READ_ONLY, 0, raf.length());
    }
}

// 协议帧结构示例
class ProtocolFrame {
    byte magic;    // 魔数标识
    int length;    // 数据长度
    byte type;     // 协议类型
    byte[] data;   // 有效载荷
    // 使用ByteBuffer解析
    public static ProtocolFrame decode(ByteBuffer buffer) {
        ProtocolFrame frame = new ProtocolFrame();
        frame.magic = buffer.get();
        frame.length = buffer.getInt();
        frame.type = buffer.get();
        frame.data = new byte[frame.length];
        buffer.get(frame.data);
        return frame;
    }
}

// 关键指标采集示例
class ReactorMetrics {
    void collectMetrics(Selector selector) {
        // 连接数监控
        int connectionCount = selector.keys().size() - 1; // 排除ServerSocketChannel
        
        // 事件处理耗时
        long start = System.nanoTime();
        selector.select(100);
        long latency = System.nanoTime() - start;
        
        // 内存使用监控
        long directMemory = ((sun.misc.VM) Class.forName("sun.misc.VM")
            .getMethod("maxDirectMemory").invoke(null));
    }
}

// Netty线程模型配置示例
EventLoopGroup bossGroup = new NioEventLoopGroup(1);  // 主Reactor
EventLoopGroup workerGroup = new NioEventLoopGroup(); // 从Reactor
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
 .channel(NioServerSocketChannel.class)
 .childHandler(new ChannelInitializer<SocketChannel>() {
     @Override
     public void initChannel(SocketChannel ch) {
         ch.pipeline().addLast(new HttpServerCodec());
         ch.pipeline().addLast(new CustomHandler());
     }
 });