1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132
| public class ChannelDemo7 { public static void main(String[] args) throws IOException { new BossEventLoop().register(); }
@Slf4j static class BossEventLoop implements Runnable { private Selector boss; private WorkerEventLoop[] workers; private volatile boolean start = false; AtomicInteger index = new AtomicInteger();
public void register() throws IOException { if (!start) { ServerSocketChannel ssc = ServerSocketChannel.open(); ssc.bind(new InetSocketAddress(8080)); ssc.configureBlocking(false); boss = Selector.open(); SelectionKey ssckey = ssc.register(boss, 0, null); ssckey.interestOps(SelectionKey.OP_ACCEPT); workers = initEventLoops(); new Thread(this, "boss").start(); log.debug("boss start..."); start = true; } }
public WorkerEventLoop[] initEventLoops() {
WorkerEventLoop[] workerEventLoops = new WorkerEventLoop[2]; for (int i = 0; i < workerEventLoops.length; i++) { workerEventLoops[i] = new WorkerEventLoop(i); } return workerEventLoops; }
@Override public void run() { while (true) { try { boss.select(); Iterator<SelectionKey> iter = boss.selectedKeys().iterator(); while (iter.hasNext()) { SelectionKey key = iter.next(); iter.remove(); if (key.isAcceptable()) { ServerSocketChannel c = (ServerSocketChannel) key.channel(); SocketChannel sc = c.accept(); sc.configureBlocking(false); log.debug("{} connected", sc.getRemoteAddress()); workers[index.getAndIncrement() % workers.length].register(sc); } } } catch (IOException e) { e.printStackTrace(); } } } }
@Slf4j static class WorkerEventLoop implements Runnable { private Selector worker; private volatile boolean start = false; private int index;
private final ConcurrentLinkedQueue<Runnable> tasks = new ConcurrentLinkedQueue<>();
public WorkerEventLoop(int index) { this.index = index; }
public void register(SocketChannel sc) throws IOException { if (!start) { worker = Selector.open(); new Thread(this, "worker-" + index).start(); start = true; } tasks.add(() -> { try { SelectionKey sckey = sc.register(worker, 0, null); sckey.interestOps(SelectionKey.OP_READ); worker.selectNow(); } catch (IOException e) { e.printStackTrace(); } }); worker.wakeup(); }
@Override public void run() { while (true) { try { worker.select(); Runnable task = tasks.poll(); if (task != null) { task.run(); } Set<SelectionKey> keys = worker.selectedKeys(); Iterator<SelectionKey> iter = keys.iterator(); while (iter.hasNext()) { SelectionKey key = iter.next(); if (key.isReadable()) { SocketChannel sc = (SocketChannel) key.channel(); ByteBuffer buffer = ByteBuffer.allocate(128); try { int read = sc.read(buffer); if (read == -1) { key.cancel(); sc.close(); } else { buffer.flip(); log.debug("{} message:", sc.getRemoteAddress()); debugAll(buffer); } } catch (IOException e) { e.printStackTrace(); key.cancel(); sc.close(); } } iter.remove(); } } catch (IOException e) { e.printStackTrace(); } } } } }
|