+-----------------+
              |  事件到达        |
              +--------+--------+
                       |
           +-----------v-----------+ 
           | Leader线程监听事件     |←----+
           +-----------+-----------+     |
                       | 处理事件        |
           +-----------v-----------+     |
           | 指定新Leader          |     |
           +-----------+-----------+     |
                       |                 |
           +-----------v-----------+     |
           | Follower晋升为Leader  |-----+
           +-----------------------+

import java.nio.channels.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
public class LeaderFollowersServer {
    private static final int MAX_THREADS = 8;
    private final AtomicInteger leaderStatus = new AtomicInteger(0); // 0=可用, 1=忙碌
    // 线程工作单元
    class Worker implements Runnable {
        private final Selector selector;
        private volatile boolean isLeader = false;
        public Worker(Selector selector) {
            this.selector = selector;
        }
        @Override
        public void run() {
            while (!Thread.interrupted()) {
                try {
                    // 尝试成为Leader
                    if (leaderStatus.compareAndSet(0, 1)) {
                        isLeader = true;
                        System.out.println(Thread.currentThread().getName() + " 成为Leader");
                        // 监听事件（非阻塞模式）
                        selector.selectNow();
                        for (SelectionKey key : selector.selectedKeys()) {
                            if (key.isAcceptable()) {
                                handleAccept((ServerSocketChannel) key.channel());
                            }
                        }
                        // 移交Leader身份
                        leaderStatus.set(0);
                        isLeader = false;
                    } else {
                        // 作为Follower处理事件
                        TimeUnit.MILLISECONDS.sleep(100);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
        private void handleAccept(ServerSocketChannel server) {
            try {
                SocketChannel client = server.accept();
                System.out.println(Thread.currentThread().getName() + " 处理连接: " + client);
                // 模拟业务处理
                TimeUnit.MILLISECONDS.sleep(500);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    public void start() throws Exception {
        Selector selector = Selector.open();
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.bind(new java.net.InetSocketAddress(8080));
        ssc.configureBlocking(false);
        ssc.register(selector, SelectionKey.OP_ACCEPT);
        ExecutorService pool = Executors.newFixedThreadPool(MAX_THREADS);
        for (int i = 0; i < MAX_THREADS; i++) {
            pool.execute(new Worker(selector));
        }
    }
    public static void main(String[] args) throws Exception {
        new LeaderFollowersServer().start();
    }
}

// CAS实现无锁选举
if (leaderStatus.compareAndSet(0, 1)) { 
    // 成功获取Leader身份
}
// 优雅退出处理
selector.wakeup(); // 唤醒阻塞的select()
pool.shutdownNow(); // 关闭线程池

// 使用Phaser实现协调
Phaser phaser = new Phaser(1);
while (true) {
    phaser.arriveAndAwaitAdvance();
    // 选举新Leader...
}

// 基于处理能力的Leader选择
if (worker.getLoad() < threshold) {
    promoteToLeader(worker);
}

// Leader切换频率监控
AtomicLong leaderChangeCount = new AtomicLong();
// 线程负载统计
ConcurrentHashMap<Worker, Integer> workloadMap = new ConcurrentHashMap<>();