netty reactor.png
BossGroup 线程组只负责Accept, WorkerGroup线程组只负责R/W。
workerGroup线程组中的工作线程是线性处理的
SelectorThread
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;
// 每线程对应一个selector，
// 多线程情况下，该主机，该程序的并发客户端被分配到多个selector上
// 注意，每个客户端，只绑定到其中一个selector
// 其实不会有交互问题
public class SelectorThread extends ThreadLocal<LinkedBlockingQueue<Channel>> implements Runnable {
Selector selector = null;
// LinkedBlockingQueue<Channel> lbq = new LinkedBlockingQueue<>();
LinkedBlockingQueue<Channel> lbq = get(); //lbq 在接口或者类中是固定使用方式逻辑写死了。你需要是lbq每个线程持有自己的独立对象
SelectorThreadGroup stg;
@Override
protected LinkedBlockingQueue<Channel> initialValue() {
return new LinkedBlockingQueue<>();//你要丰富的是这里！ pool。。。
}
SelectorThread() {
try {
selector = Selector.open();
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void run() {
// Loop
while (true) {
try {
// 1,select()
int nums = selector.select(); //阻塞 需要 wakeup() 去唤醒
// 2,处理selectkeys
if (nums > 0) {
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> iter = keys.iterator();
while (iter.hasNext()) { // 线程处理的过程
SelectionKey key = iter.next();
iter.remove();
if (key.isAcceptable()) { // 复杂,接受客户端的过程（接收之后，要注册，多线程下，新的客户端，注册到那里呢？）
acceptHandler(key);
} else if (key.isReadable()) {
readHander(key);
} else if (key.isWritable()) {
// TODO: 2021-02-26
}
}
}
//3,处理一些task : listen client
if (!lbq.isEmpty()) { // 队列是个啥东西啊？ 堆里的对象，线程的栈是独立，堆是共享的
// 只有方法的逻辑，本地变量是线程隔离的
Channel c = lbq.take();
if (c instanceof ServerSocketChannel) {
ServerSocketChannel server = (ServerSocketChannel) c;
server.register(selector, SelectionKey.OP_ACCEPT);
System.out.println(Thread.currentThread().getName() + " register listen");
} else if (c instanceof SocketChannel) {
SocketChannel client = (SocketChannel) c;
ByteBuffer buffer = ByteBuffer.allocateDirect(4096);
client.register(selector, SelectionKey.OP_READ, buffer);
System.out.println(Thread.currentThread().getName() + " register client: " + client.getRemoteAddress());
}
}
} catch (IOException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
private void readHander(SelectionKey key) {
System.out.println(Thread.currentThread().getName() + " read......");
ByteBuffer buffer = (ByteBuffer) key.attachment();
SocketChannel client = (SocketChannel) key.channel();
buffer.clear();
while (true) {
try {
int num = client.read(buffer);
if (num > 0) {
buffer.flip(); //将读到的内容翻转，然后直接写出
while (buffer.hasRemaining()) {
client.write(buffer);
}
buffer.clear();
} else if (num == 0) {
break;
} else if (num < 0) {
//客户端断开了
System.out.println("client: " + client.getRemoteAddress() + " closed......");
key.cancel();
break;
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
// 处理client的链接
private void acceptHandler(SelectionKey key) {
System.out.println(Thread.currentThread().getName() + " acceptHandler......");
ServerSocketChannel server = (ServerSocketChannel) key.channel();
try {
SocketChannel client = server.accept();
client.configureBlocking(false);
// choose a selector and register!!
stg.nextSelectorV3(client);
} catch (IOException e) {
e.printStackTrace();
}
}
public void setWorker(SelectorThreadGroup stgWorker) {
this.stg = stgWorker;
}
}
SelectorThreadGroup
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.Channel;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.ServerSocketChannel;
import java.util.concurrent.atomic.AtomicInteger;
public class SelectorThreadGroup { // 天生都是boss
SelectorThread[] sts;
ServerSocketChannel server = null;
AtomicInteger xid = new AtomicInteger(0);
SelectorThreadGroup stg = this;
public void setWorker(SelectorThreadGroup stg) {
this.stg = stg;
}
SelectorThreadGroup(int num) {
//num 线程数
sts = new SelectorThread[num];
for (int i = 0; i < num; i++) {
sts[i] = new SelectorThread();
new Thread(sts[i]).start();
}
}
public void bind(int port) {
try {
server = ServerSocketChannel.open();
server.configureBlocking(false);
server.bind(new InetSocketAddress(port));
// 注册到那个selector上呢？
nextSelectorV3(server);
} catch (IOException e) {
e.printStackTrace();
}
}
public void nextSelectorV3(Channel c) {
try {
if (c instanceof ServerSocketChannel) {
SelectorThread st = next(); //listen 选择了 boss组中的一个线程后，要更新这个线程的work组
st.setWorker(stg);
//1,通过队列传递数据 消息
st.lbq.put(c);
//2,通过打断阻塞，让对应的线程去自己在打断后完成注册selector
st.selector.wakeup();
} else {
SelectorThread st = nextV3(); //在 main线程种，取到堆里的selectorThread对象
st.lbq.add(c);
st.selector.wakeup();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 无论 serversocket socket 都复用这个方法
private SelectorThread next() {
int index = xid.incrementAndGet() % sts.length; // 轮询就会很尴尬，倾斜
return sts[index];
}
private SelectorThread nextV3() {
int index = xid.incrementAndGet() % stg.sts.length; // 动用worker的线程分配
return stg.sts[index];
}
}
MainThread
// 这里不做关于IO 和 业务的事情
public class MainThread {
public static void main(String[] args) {
// 1,创建 IO Thread （一个或者多个）
SelectorThreadGroup boss = new SelectorThreadGroup(3); // boss有自己的线程组
SelectorThreadGroup worker = new SelectorThreadGroup(4); // worker有自己的线程组
// 但是，boss得持有worker线程组的引用
boss.setWorker(worker);
/*
* boss里选一个线程注册listen ， 触发bind，从而，这个选中的线程得持有workerGroup的引用
* 因为未来 listen 一旦accept得到client后得去worker中 next出一个线程分配(R/W)
*/
// 2，把 监听（9999）的 server 注册到某一个 selector 上
boss.bind(9999);
// boss.bind(8888);
// boss.bind(6666);
// boss.bind(7777);
}
}
总结
1，每个客户端分配一个selector，boss组中的线程只负责处理accept,worker组中的线程只负责R/W（并且处理过程是线性的）
2，执行完new SelectorThreadGroup(n)这行代码后，自动创建的线程全部阻塞在 int nums = selector.select(); 需要 wakeup() 去唤醒，通过队列解决
3，boss组里选一个线程注册listen ， 触发bind，从而，这个选中的线程得持有workerGroup的引用，因为未来 listen 一旦accept得到client后，得去worker中 next出一个线程分配(R/W)
————————————————————
坐标帝都，白天上班族，晚上是知识的分享者
如果读完觉得有收获的话，欢迎点赞加关注
网友评论