1 最原始架构
一个线程负责处理连接、读写等各种请求。
创建一个线程，注册到 Selector，将 serversocketchannel 注册到Selector selectionKey 里就有具体的事件
对应代码
代码语言：javascript复制package io.netty.example.helloworld;
import io.netty.channel.EventLoopGroup;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;
import java.util.Iterator;
import java.util.Set;
/**
* @author JavaEdge
* @date 2021/5/17
*/
public class NioServer {
public static void main(String[] args) throws Exception {
// 创建一个 ServerSocketChannel
ServerSocketChannel serverChannel = ServerSocketChannel.open();
serverChannel.bind(new InetSocketAddress(8080));
// 设置为非阻塞模式
serverChannel.configureBlocking(false);
// 创建一个事件查询器
Selector selector = SelectorProvider.provider().openSelector();
// 把 ServerSocketChannel 注册到 selector，并且感兴趣 OP_ACCEPT 事件
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
while (true) {
// 阻塞方法，等待系统有I/O事件发生
int eventNum = selector.select();
System.out.println("系统发生IO事件 数量->" eventNum);
Set<SelectionKey> keySet = selector.selectedKeys();
Iterator<SelectionKey> iterable = keySet.iterator();
while (iterable.hasNext()) {
// 拿到该 key
SelectionKey key = iterable.next();
// 拿到后就移除它，否则后面遍历还会重复拿到它
iterable.remove();
// 连接事件
if (key.isAcceptable()) {
ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
// 接受客户端的连接,一个 SocketChannel 代表一个TCP连接
SocketChannel socketChannel = ssc.accept();
// 把SocketChannel设置为非阻塞模式
socketChannel.configureBlocking(false);
System.out.println("服务器接受了一个新的连接 " socketChannel.getRemoteAddress());
}
}
}
}
}package io.netty.example.helloworld;
import io.netty.channel.EventLoopGroup;
import java.io.IOException;
import java.net.InetSocketAddress;
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.nio.channels.spi.SelectorProvider;
import java.util.Arrays;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* @author JavaEdge
* @date 2021/5/17
*/
public class NioServer {
public static void main(String[] args) throws Exception {
//创建一个ServerSocket
ServerSocketChannel serverChannel = ServerSocketChannel.open();
serverChannel.bind(new InetSocketAddress(8089));
//设置为非阻塞模式
serverChannel.configureBlocking(false);
// 创建一个事件查询器
Selector selector = SelectorProvider.provider().openSelector();
// 把 ServerSocketChannel 注册到事件查询器上，并且感兴趣 OP_ACCEPT 事件
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
//
// //创建一组事件查询器
// EventLoopGroup eventLoopGroup = new EventLoopGroup();
while (true) {
// 阻塞方法，等待系统有I/O事件发生
int eventNum = selector.select();
System.out.println("系统发生IO事件 数量->" eventNum);
Set<SelectionKey> keySet = selector.selectedKeys();
Iterator<SelectionKey> iterable = keySet.iterator();
while (iterable.hasNext()) {
// 拿到该 key
SelectionKey key = iterable.next();
// 拿到后就移除它，否则后面遍历还会重复拿到它
iterable.remove();
// 连接事件
if (key.isAcceptable()) {
// 因为只有 ServerSocketChannel 有接收事件，所以可直接强转
ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
// 接受客户端的连接,一个 SocketChannel 代表一个TCP连接
// 事件如果发生了，就肯定有新的连接
SocketChannel socketChannel = ssc.accept();
// 把SocketChannel设置为非阻塞模式
socketChannel.configureBlocking(false);
System.out.println("服务器接受了一个新的连接 " socketChannel.getRemoteAddress());
// 把SocketChannel注册到Selector，并关注OP_READ事件
socketChannel.register(selector, SelectionKey.OP_READ);
// eventLoopGroup.register(socketChannel, SelectionKey.OP_READ);
}
// 可读事件
if (key.isReadable()) {
SocketChannel socketChannel = (SocketChannel) key.channel();
ByteBuffer buffer = ByteBuffer.allocate(1024);
try {
int readNum = socketChannel.read(buffer);
if (readNum == -1) {
System.out.println("读取结束,关闭 socket");
key.channel();
socketChannel.close();
break;
}
// 将Buffer从写模式切到读模式
buffer.flip();
byte[] bytes = new byte[readNum];
buffer.get(bytes, 0, readNum);
System.out.println(new String(bytes));
/* byte[] response = "client hello".getBytes();
// 清理了才可以重新使用
buffer.clear();
buffer.put(response);
buffer.flip();
// 该方法非阻塞的，如果此时无法写入也不会阻塞在此，而是直接返回 0 了
socketChannel.write(buffer);
*/
// 在 key 上附加一个对象
key.attach("hello client".getBytes());
// 把 key 关注的事件切换为写
key.interestOps(SelectionKey.OP_WRITE);
} catch (IOException e) {
System.out.println("读取时发生异常,关闭 socket");
// 取消 key
key.channel();
}
}
if (key.isWritable()) {
SocketChannel socketChannel = (SocketChannel) key.channel();
// 可写时再将那个对象拿出来
byte[] bytes = (byte[]) key.attachment();
key.attach(null);
System.out.println("可写事件发生 写入消息" Arrays.toString(bytes));
if (bytes != null) {
socketChannel.write(ByteBuffer.wrap(bytes));
}
// 写完后，就不需要写了，就切换为读事件 如果不写该行代码就会死循环
// key.interestOps(SelectionKey.OP_READ);
}
}
}
}
}
2 接收请求单独处理
- 架构图
2.1 死锁案例
代码语言：javascript复制package io.netty.example.helloworld;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;
import java.util.Iterator;
import java.util.Set;
/**
* @author JavaEdge
* @date 2021/5/17
*/
public class NioServer {
public static void main(String[] args) throws Exception {
// 创建一个ServerSocket
ServerSocketChannel serverChannel = ServerSocketChannel.open();
serverChannel.bind(new InetSocketAddress(8089));
// 设置为非阻塞模式
serverChannel.configureBlocking(false);
// 创建一个事件查询器
Selector selector = SelectorProvider.provider().openSelector();
// 把 ServerSocketChannel 注册到事件查询器上，并且感兴趣 OP_ACCEPT 事件
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
EventLoop eventLoop = new EventLoop();
while (true) {
// 阻塞方法，等待系统有I/O事件发生
int eventNum = selector.select();
System.out.println("系统发生IO事件 数量->" eventNum);
Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
while (keyIterator.hasNext()) {
// 拿到该 key
SelectionKey key = keyIterator.next();
// 拿到后就移除它，否则后面遍历还会重复拿到它
keyIterator.remove();
// 只需处理【连接事件】 a connection was accepted by a ServerSocketChannel.
if (key.isAcceptable()) {
// 因为只有 ServerSocketChannel 有接收事件，所以可直接强转
ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
// 接受客户端的连接,一个 SocketChannel 代表一个TCP连接
// 事件如果发生了，就肯定有新的连接
SocketChannel socketChannel = ssc.accept();
// 把SocketChannel设置为非阻塞模式
socketChannel.configureBlocking(false);
System.out.println("服务器接受了一个新的连接 " socketChannel.getRemoteAddress());
// 把SocketChannel注册到Selector，并关注OP_READ事件
// socketChannel.register(selector, SelectionKey.OP_READ);
eventLoop.register(socketChannel, SelectionKey.OP_READ);
}
}
}
}
}package io.netty.example.helloworld;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;
import java.util.Arrays;
import java.util.Iterator;
import java.util.Set;
/**
* @author JavaEdge
* @date 2021/5/25
*/
public class EventLoop implements Runnable {
private Selector selector;
private Thread thread;
public EventLoop() throws IOException {
this.selector = SelectorProvider.provider().openSelector();
this.thread = new Thread(this);
this.thread.start();
}
/**
* 把 channel 注册到 事件查询器
*/
public void register(SocketChannel channel, int keyOps) throws ClosedChannelException {
channel.register(selector, keyOps);
}
@Override
public void run() {
while (!Thread.interrupted()) {
try {
// 阻塞方法，等待系统有 I/0 事件产生
int eventNum = selector.select();
System.out.println("系统发生IO事件 数量->" eventNum);
Set<SelectionKey> keySet = selector.selectedKeys();
Iterator<SelectionKey> iterable = keySet.iterator();
while (iterable.hasNext()) {
SelectionKey key = iterable.next();
iterable.remove();
// 可读事件
if (key.isReadable()) {
SocketChannel socketChannel = (SocketChannel) key.channel();
ByteBuffer buffer = ByteBuffer.allocate(1024);
try {
int readNum = socketChannel.read(buffer);
if (readNum == -1) {
System.out.println("读取结束,关闭 socket");
key.channel();
socketChannel.close();
break;
}
// 将Buffer从写模式切到读模式
buffer.flip();
byte[] bytes = new byte[readNum];
buffer.get(bytes, 0, readNum);
System.out.println(new String(bytes));
/* byte[] response = "client hello".getBytes();
// 清理了才可以重新使用
buffer.clear();
buffer.put(response);
buffer.flip();
// 该方法非阻塞的，如果此时无法写入也不会阻塞在此，而是直接返回 0 了
socketChannel.write(buffer);
*/
// 在 key 上附加一个对象
key.attach("EventLoop says hello to client".getBytes());
// 把 key 关注的事件切换为写
key.interestOps(SelectionKey.OP_WRITE);
} catch (IOException e) {
System.out.println("读取时发生异常,关闭 socket");
// 取消 key
key.channel();
}
}
if (key.isWritable()) {
SocketChannel socketChannel = (SocketChannel) key.channel();
// 可写时再将那个对象拿出来
byte[] bytes = (byte[]) key.attachment();
key.attach(null);
System.out.println("可写事件发生 写入消息" Arrays.toString(bytes));
if (bytes != null) {
socketChannel.write(ByteBuffer.wrap(bytes));
}
// 写完后，就不需要写了，就切换为读事件 如果不写该行代码就会死循环
key.interestOps(SelectionKey.OP_READ);
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
}启动之后，开启一个客户端连接请求：
打断点到该行代码：
点击继续执行时，dump此时的线程状态：主线程已经阻塞在此
说明主线程在等待 @574线程的锁，它是谁呢？没错
EventLoop 线程阻塞在select 方法，而且它此时已经获取了Selector 内部的一把锁，所以不是Blocked状态。 但此时主线程执行 register 也需要该Selector内部的这把锁，但又不是同一线程，所以产生死锁。 所以不能由main 线程调用注册方法。
2.2 解决死锁
改造后的 EventLoop 类：
代码语言：javascript复制package io.netty.example.helloworld;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;
import java.util.Arrays;
import java.util.Iterator;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.LinkedBlockingDeque;
/**
* @author JavaEdge
* @date 2021/5/25
*/
public class EventLoop implements Runnable {
private Selector selector;
private Thread thread;
private Queue<Runnable> taskQueue = new LinkedBlockingDeque<>(32);
public EventLoop() throws IOException {
this.selector = SelectorProvider.provider().openSelector();
this.thread = new Thread(this);
this.thread.start();
}
/**
* 把 channel 注册到 事件查询器
*/
public void register(SocketChannel channel, int keyOps) {
// 将注册的逻辑封装成一个任务，因为不能让主线程执行，必须由 eventloop 的线程执行
taskQueue.add(() -> {
try {
channel.register(selector, keyOps);
} catch (ClosedChannelException e) {
e.printStackTrace();
}
});
// 但此时EventLoop的线程阻塞在 selector.select()，通过主线程唤醒它
selector.wakeup();
}
@Override
public void run() {
while (!Thread.interrupted()) {
try {
System.out.println(thread "开始查询 I/O 事件...");
// 阻塞方法，等待系统有 I/0 事件产生
int eventNum = selector.select();
System.out.println("系统发生IO事件 数量->" eventNum);
// 有事件则处理
if (eventNum > 0) {
Set<SelectionKey> keySet = selector.selectedKeys();
Iterator<SelectionKey> iterable = keySet.iterator();
while (iterable.hasNext()) {
SelectionKey key = iterable.next();
iterable.remove();
// 可读事件
if (key.isReadable()) {
SocketChannel socketChannel = (SocketChannel) key.channel();
ByteBuffer buffer = ByteBuffer.allocate(1024);
try {
int readNum = socketChannel.read(buffer);
if (readNum == -1) {
System.out.println("读取结束,关闭 socket");
key.channel();
socketChannel.close();
break;
}
// 将Buffer从写模式切到读模式
buffer.flip();
byte[] bytes = new byte[readNum];
buffer.get(bytes, 0, readNum);
System.out.println(new String(bytes));
/* byte[] response = "client hello".getBytes();
// 清理了才可以重新使用
buffer.clear();
buffer.put(response);
buffer.flip();
// 该方法非阻塞的，如果此时无法写入也不会阻塞在此，而是直接返回 0 了
socketChannel.write(buffer);
*/
// 在 key 上附加一个对象
key.attach("EventLoop says hello to client".getBytes());
// 把 key 关注的事件切换为写
key.interestOps(SelectionKey.OP_WRITE);
} catch (IOException e) {
System.out.println("读取时发生异常,关闭 socket");
// 取消 key
key.channel();
}
}
if (key.isWritable()) {
SocketChannel socketChannel = (SocketChannel) key.channel();
// 可写时再将那个对象拿出来
byte[] bytes = (byte[]) key.attachment();
key.attach(null);
System.out.println("可写事件发生 写入消息" Arrays.toString(bytes));
if (bytes != null) {
socketChannel.write(ByteBuffer.wrap(bytes));
}
// 写完后，就不需要写了，就切换为读事件 如果不写该行代码就会死循环
key.interestOps(SelectionKey.OP_READ);
}
}
}
// 无事件则执行任务
Runnable task;
while ((task = taskQueue.poll()) != null) {
// EventLoop执行队列中的任务，即注册任务
task.run();
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
}3 EventLoopGroup
由于只使用一个 Selector 来处理客户端的读写请求，如果并发太大，太多 socketchannel，这个死循环就可能处理不过来，造成大量请求超时。 所以有了EventLoopGroup。 且 channel 负责读、写事件的处理。
代码语言：javascript复制package io.netty.example.helloworld;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
/**
* 类似 netty 的 channel
*
* @author JavaEdge
*
* @date 2021/5/27
*/
public class MyChannel {
private SocketChannel channel;
private EventLoop eventLoop;
/**
* 写数据的缓冲区
*/
private Queue<ByteBuffer> writeQueue = new ArrayBlockingQueue<>(16);
public MyChannel(SocketChannel channel,EventLoop eventLoop) {
this.channel = channel;
this.eventLoop = eventLoop;
}
public void read(SelectionKey key) throws IOException {
SocketChannel socketChannel = (SocketChannel) key.channel();
ByteBuffer buffer = ByteBuffer.allocate(1024);
try {
int readNum = socketChannel.read(buffer);
if (readNum == -1) {
System.out.println("读取结束,关闭 socket");
key.channel();
socketChannel.close();
return;
}
// 将Buffer从写模式切到读模式
buffer.flip();
byte[] bytes = new byte[readNum];
// 客户端发来的数据
buffer.get(bytes, 0, readNum);
String clientData = new String(bytes);
System.out.println(clientData);
// 加入写缓冲区
writeQueue.add(ByteBuffer.wrap("hello JavaEdge".getBytes()));
if ("flush".equals(clientData)) {
// 把 key 关注的事件切换为写
key.interestOps(SelectionKey.OP_WRITE);
}
} catch (IOException e) {
System.out.println("读取时发生异常,关闭 socket");
// 取消 key
key.channel();
socketChannel.close();
}
}
public void write(SelectionKey key) throws IOException {
ByteBuffer byteBuffer;
while ((byteBuffer = writeQueue.poll()) != null) {
channel.write(byteBuffer);
}
// 写完后，就不需要写了，就切换为读事件 如果不写该行代码就会死循环
key.interestOps(SelectionKey.OP_READ);
}
}package io.netty.example.helloworld;
import java.io.IOException;
import java.nio.channels.SocketChannel;
import java.util.concurrent.atomic.AtomicInteger;
/**
* @author JavaEdge
* @date 2021/5/25
*/
public class EventLoopGroup {
private EventLoop[] eventLoops = new EventLoop[2];
private final AtomicInteger idx = new AtomicInteger(0);
public EventLoop next() {
// 轮询算法
return eventLoops[idx.getAndIncrement() & eventLoops.length - 1];
}
public EventLoopGroup() throws IOException {
for (int i = 0; i < eventLoops.length; i ) {
eventLoops[i] = new EventLoop();
}
}
/**
* 其实啥也不干，直接找到一个EventLoop，丢给他干
*/
public void register(SocketChannel channel, int keyOps) {
next().register(channel, keyOps);
}
}canceled key
启动程序，客户端发起连接请求，然后点击断开连接
报错如下：
因为点击断开连接时，是会产生一个读事件请求，而这时会将该 channel 关闭
- 这里加一行即可
pipeline
我们需要将业务代码抽出来，给业务开发人员使用。基本上开发人员只需要写各个编解码器即可。
代码语言：javascript复制package io.netty.example.helloworld;
import lombok.extern.slf4j.Slf4j;
/**
* @author JavaEdge
* @date 2021/5/28
*/
@Slf4j
public class MyHandler2 implements Handler {
@Override
public void channelRead(HandlerContext ctx, Object msg) {
// 上一个处理器解码成 String 了，所以这里直接转型处理 String
String string = (String) msg;
// 处理业务
log.debug(string);
// 传给 handler2
ctx.getMyChannel().doWrite("hello client");
if ("flush".equals(string)) {
/**
* 这样调用，会跳过而不调用 handler2 的 flush 方法
* 若还需要调用 handler2 的 flush 方法，应该通过 channel 调用:
* ctx.getMyChannel().flush();
*/
ctx.flush();
}
}
@Override
public void write(HandlerContext ctx, Object msg) {
log.debug("msg=" msg);
msg = "!!!";
// 传递给 handler1
ctx.write(msg);
}
@Override
public void flush(HandlerContext ctx) {
log.debug("flush");
// 调用 handler1
ctx.flush();
}
}package io.netty.example.helloworld;
import lombok.extern.slf4j.Slf4j;
import java.nio.ByteBuffer;
/**
* @author JavaEdge
* @date 2021/5/28
*/
@Slf4j
public class PipeLine {
private MyChannel myChannel;
private EventLoop eventLoop;
HandlerContext headCtx;
HandlerContext tailCtx;
public PipeLine(MyChannel myChannel, EventLoop eventLoop) {
this.myChannel = myChannel;
this.eventLoop = eventLoop;
PileLineHandler pileLineHandler = new PileLineHandler();
this.headCtx = new HandlerContext(pileLineHandler, myChannel);
this.tailCtx = new HandlerContext(pileLineHandler, myChannel);
// 构建初始化的链表
this.headCtx.next = this.tailCtx;
this.tailCtx.prev = this.headCtx;
}
class PileLineHandler implements Handler {
@Override
public void channelRead(HandlerContext ctx, Object msg) {
log.debug(msg.toString());
log.info("tail handler" msg);
}
/**
* 因为写数据是从后往前处理，所以最终到该处理器，必须要调用 channel 执行底层的写数据到 socket
*/
@Override
public void write(HandlerContext ctx, Object msg) {
log.debug(msg.toString());
// 既然是写底层，那就必须是 ByteBuffer 类型
if (!(msg instanceof ByteBuffer)) {
throw new RuntimeException("error class type" msg.getClass());
}
// 类型符合，则加入到 channel 的缓冲区队列
PipeLine.this.myChannel.addWriteQueue((ByteBuffer) msg);
}
/**
* 上边的 write 方法也只是将数据写到 channel 的临时缓冲区队列，并没有真正写进socket 输出
* 当客户端调用了 flush 才真正的写数据出去。
*/
@Override
public void flush(HandlerContext ctx) {
log.debug("flush");
// 最后是由 pipeline 和 channel 交互写的数据
PipeLine.this.myChannel.doFlush();
}
}
/**
* 仅演示添加到链尾
*
* @param handler
*/
public void addLast(Handler handler) {
HandlerContext ctx = new HandlerContext(handler, myChannel);
HandlerContext prev = this.tailCtx.prev;
prev.next = ctx;
ctx.prev = prev;
ctx.next = this.tailCtx;
tailCtx.prev = ctx;
}
}
