# Netty
## NIO
###
- ChannelstreamInputStreamOutputStreamChannelchannelbufferbufferchannelstreamchannelstream
Channel
- FileChannel
- DatagramChannelUDP
- SocketChannelTCP
- ServerSocketChannelTCP
- Bufferbuffer
buffer
- ByteBuffer
- MappedByteBuffer
- DirectByteBuffer
- HeapByteBuffer
- SelectorSelectorchannelchannel
### Buffer
#####
```java
try(RandomAccessFile file = new RandomAccessFile("file", "rw")) {
FileChannel fileChannel = file.getChannel();
//
ByteBuffer buffer = ByteBuffer.allocate(10);
int len;
// 1fileChannelbuffer
while ((len = fileChannel.read(buffer)) != -1) {
System.out.println(": " + len);
// 2 buffer
buffer.flip();
//
while (buffer.hasRemaining()) {
// 3buffer
System.out.println((char) buffer.get());
}
// 4 buffer
buffer.clear();
}
} catch (Exception e) {
e.printStackTrace();
}
```
##### ByteBuffer
ByteBuffer
- capacity
- position/
- limit/
positionlimitcapacity
flip()positionlimit
4
clear
compact()
#####
###
FileChannelFileChannelFileInputStreamFileOutputStreamFileChannel
- FileInputStreamchannel
- FileOutputStreamchannel
- RandomAccessFilechannelRandomAccessFile
Path & Files
###
####
-
-
- TLVTypeLengthValuebufferbuffer
- HTTP 1.0TLV
- HTTP 2.0LTV
> Redis
###
#### IO
IOsocket
```java
RandomAccessFile file = new RandomAccessFile("file", "r");
byte[] buf = new byte[1024];
file.read(buf);
Socket socket = ...;
socket.getOutputStream().write(buf);
```
1. JavaIOreadJavaKernelDMADirect Memory AccessCPU
> DMACPUIO
2. `byte[] buf`CPUDMA
3. write`byte[] buf`socketCPU
4. JavaDMAsocketCPU
JavaIO
-
-
#### NIO
- ByteBuffer.allocate(10)HeadByteBufferJava
- ByteBuffer.allocateDirect(10)DirectByteBuffer
JavaDirectByteBufferJVM
-
#### NIO
Linux 2.1 sendFileJavaChanneltransferTo/transferFrom
1. JavatransferTo()JavaDMACPU
2. socketCPU
3. DMAsocketCPU
Linux 2.4
1. JavatransferTo()JavaDMACPU
- offsetlengthsocket
2. DMACPU
``**JVM**
- CPUPCU
-
-
##
###
#### server
```java
public class Server {
public static void main(String[] args) throws InterruptedException {
new ServerBootstrap()
.group(new NioEventLoopGroup())
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer() {
@Override
//
protected void initChannel(NioSocketChannel ch) throws Exception {
ch.pipeline().addLast(new StringDecoder());
ch.pipeline().addLast(new SimpleChannelInboundHandler() {
@Override
public void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println(msg);
ctx.channel().writeAndFlush(Unpooled.copiedBuffer("", CharsetUtil.UTF_8));
}
});
}
}).bind(8080);
}
}
```
#### client
```java
public class Client {
public static void main(String[] args) throws InterruptedException {
new Bootstrap()
.group(new NioEventLoopGroup())
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer() {
@Override
//
protected void initChannel(NioSocketChannel ch) throws Exception {
ch.pipeline().addLast(new StringEncoder());
ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
@Override
// i
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf byteBuf=(ByteBuf)msg;
System.out.println(""+byteBuf.toString(CharsetUtil.UTF_8));
}
});
}
})
//
.connect(new InetSocketAddress("localhost", 8080))
.sync()
.channel()
.writeAndFlush("hello world");
}
}
```
###
- Channel
- msgByteBufferpipelineByteBuffer
- handler
- handlerpipelinepipeline...handlerhandlerhandlerhandler
- handlerInboundOutbound
- EventLoop
- EventLoopchannelioEventLoopchannel
- EventLoopioEventLoopchannel
- EventLooppipelinehandlerhandlerEventLoop
###
#### EventLoop
EventLoopSelectorrun()ChannelIO
```java
public interface EventLoop extends OrderedEventExecutor, EventLoopGroup {}
public interface OrderedEventExecutor extends EventExecutor {}
public interface EventLoopGroup extends EventExecutorGroup {}
public interface EventExecutorGroup extends ScheduledExecutorService, Iterable {}
```
- j.u.c.ScheduledExecutorService
- NettyOrderedEventExecutor
- `boolean inEventLoop(Thread thread)`EventLoop
- `EventExecutorGroup parent()`EventLoopGroup
EventLoopGroupEventLoopChannelEventLoopGroupregisterEventLoopChannelIO EventLoop
- NettyEventExecutorGroup
- IterableEventLoop
- next()EventLoopGroupEventLoop
##### &
```java
// NioEventLoopGroupio
EventLoopGroup group = new NioEventLoopGroup(2);
// DefaultEventLoopGroup
EventLoopGroup group = new DefaultEventLoopGroup(2);
// EventLoop
group.next();
//
group.next().submit(() -> {
......
});
//
group.next().scheduleAtFixedRate(() -> {
......
}, 0, 1, TimeUnit.SECONDS);
```
#####
```java
public class Server {
public static void main(String[] args) throws InterruptedException {
EventLoopGroup group = new DefaultEventLoopGroup(2);
new ServerBootstrap()
.group(new NioEventLoopGroup(1), new NioEventLoopGroup())
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer() {
@Override
//
protected void initChannel(NioSocketChannel ch) throws Exception {
ch.pipeline().addLast(new StringDecoder());
// group
ch.pipeline().addLast("handler-1", new SimpleChannelInboundHandler() {
@Override
public void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println(msg);
// msghandler
ctx.fireChannelRead(msg);
}
}).addLast(group, "handler-2", new SimpleChannelInboundHandler() {
@Override
public void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println(msg);
ctx.channel().writeAndFlush(Unpooled.copiedBuffer("", CharsetUtil.UTF_8));
}
});
}
}).bind(8080);
}
}
```
EventLoopGroupbossGroupworkerGroupbossGroupNioServerSocketChannelacceptworkerGroupNioSocketChannel
handlerhandlerEventLoopIO
#####
```java
// io.netty.channel.AbstractChannelHandlerContext#fireChannelRead
public ChannelHandlerContext fireChannelRead(final Object msg) {
invokeChannelRead(findContextInbound(MASK_CHANNEL_READ), msg);
return this;
}
// io.netty.channel.AbstractChannelHandlerContext#invokeChannelRead
static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
// handlerEventLoop
EventExecutor executor = next.executor();
/*
handlerhandlerEventLoop
handlerEventLoopifelse
*/
if (executor.inEventLoop()) {
next.invokeChannelRead(m);
} else {
// handlerexecutor
executor.execute(new Runnable() {
@Override
public void run() {
next.invokeChannelRead(m);
}
});
}
}
```
#### Channel
NettyChannelJDKChannelI/Oread()write()connect()bind()NettyNioServerSocketChannelNioSocketChannel
ChannelPipelineChannelHandler
channelRegisteredchannelReadchannelActive...
Channel
- close()channel
- closeFuture()channel
- sync()channel
- addListener()channel
- pipeline()handler
- write()
- writeAndFlush()
#####
```java
public class Client {
public static void main(String[] args) throws InterruptedException {
new Bootstrap()
.group(new NioEventLoopGroup())
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer() {
@Override
//
protected void initChannel(NioSocketChannel ch) throws Exception {
ch.pipeline().addLast(new StringEncoder());
ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
@Override
// i
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf byteBuf=(ByteBuf)msg;
System.out.println(""+byteBuf.toString(CharsetUtil.UTF_8));
}
});
}
})
//
.connect(new InetSocketAddress("localhost", 8080))
.sync()
.channel()
.writeAndFlush("hello world");
}
}
```
connect()ChannelFuturemainconnectniosync()addListener()nionio
closeFuture()sync()
NettyFutureFuture
-
-
-
#### Future&Promise
FuturePromise
NettyFutureJDKNettyFutureJDKFuturePromiseNetty Future
Promise
NettyPromise
#### Handler&Pipeline
handlersocket io
ChannelHandlerChannelIOChannelHandlerPipeline
- ChannelInboundHandlerAdapter
- ChannelOutboundHandlerAdapter
> ChannelPipelineChannelHandlerByteBuf
`head -> h1 -> h2 -> h3 -> h4 -> h5 -> h6 -> tail`
- `h1 -> h2 -> h3`
- `h6 -> h5 -> h4`
- `ctx.channel().writeAndFlush()`tail -> head
- `ctx.writeAndFlush()` -> head
channel
pipeline
NettyPipeline
`EmbeddedChannel`Nettyhandlerhandler
#### ByteBuf
NettyByteBufnioByteBuffer
#####
```java
/**
256
* ByteBuf buffer();
* ByteBuf buffer(int initialCapacity);
* ByteBuf buffer(int initialCapacity, int maxCapacity);
*/
ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
buffer.writeBytes("".toString().getBytes());
```
##### &
-
- `ByteBuf buffer = ByteBufAllocator.DEFAULT.heapBuffer(10);`
-
- `ByteBuf buffer = ByteBufAllocator.DEFAULT.directBuffer(10);`
- ByteBuf
-
- GCJVM
#####
- ByteBuf
-
- ByteBuf ByteBuf ;
- ByteBuf
- ByteBuf
- readerIndexwriterIndex capacity ByteBuf
- ByteBuf readerIndex 1ByteBuf writerIndex-readerIndex , readerIndex writerIndex ByteBuf
- writerIndex writerIndex 1 capacity ByteBuf
- ByteBuf maxCapacity ByteBuf capacity maxCapacity maxCapacity
##### &
-
| | |
|:-------------------------------------------------------------:| ------------------------- |
| writeBoolean(boolean value) | |
| writeByte(int value) | |
| writeShort(int value) | |
| writeInt(int value) | int |
| writeIntLE(int value) | int |
| writeLong(long value) | |
| writeChar(int value) | |
| writeFloat(float value) | |
| writeDouble(double value) | |
| writeBytes(ByteBuf src) | netty ByteBuf |
| writeBytes(byte[] src) | byte[] |
| writeBytes(ByteBuffer src) | nio ByteBuffer |
| int writeCharSequence(CharSequence sequence, Charset charset) | |
ByteBuf
set
-
- ByteBuf1012
- 5121612capacity16
- 5122^n513capacity2^10=1024
- `maxcapacity`
-
| | |
| ------------------------- | --------------------------------- |
| buffer.readByte() | |
| buffer.readInt() | |
| buffer.markReaderIndex() | |
| buffer.resetReaderIndex() | |
readgetget
buffer.markReaderIndex()buffer.resetReaderIndex()mark
#####
NettyByteBufReferenceCounted
- ByteBuf1
- release10ByteBuf
- retain1handlerrelease
- 0ByteBuf
ByteBufrelease
- ByteBuftailreleaseByteBufheadrelease
#####
- slice
- ByteBufByteBufByteBufByteBufByteBufreadwrite
```java
ByteBuf buffer = ByteBufAllocator.DEFAULT.directBuffer(10);
buffer.writeBytes("abcdefg".toString().getBytes());
/*
buffer.slice()
buffer.slice(0, 3)
*/
buffer.slice();
buffer.slice(0, 3);
```
ByteBuf
ByteBufByteBufByteBufretain()1retainrelease
- composite
- CompositeByteBufByteBufByteBufByteBuf
```java
ByteBuf buffer1 = ByteBufAllocator.DEFAULT.buffer(10);
buffer1.writeBytes(new byte[]{1, 2, 3, 4, 5});
ByteBuf buffer2 = ByteBufAllocator.DEFAULT.buffer(10);
buffer2.writeBytes(new byte[]{6, 7, 8, 9, 10});
CompositeByteBuf buffer = ByteBufAllocator.DEFAULT.compositeBuffer(10);
buffer.addComponents(true, buffer1, buffer2);
```
#### Unpooled
UnpooledByteBuf
##
###
-
-
TCP
####
- `adb``def``abcdef`
-
- ByteBufNetty1024
- 256bytes256bytes
- Nagle
- `abcdef``adb``def`
-
- ByteBuf
- 128bytes256bytes128bytesack
- MSSMSS
> MSS (Maximum Segment Size)TCP payloadTCPMSSTCP
>
> Maximum Transmission UnitMTU**payload,**,IP,ICMP
TCP
####
- `-`
- bit
-
- LTC`LengthFieldBasedFrameDecoder`
###
####
-
-
- ProtoBufhessian
- ...
-
-
-
##
###
**DefaultChannelConfig**
####
- SocketChannel`CONNECT_TIMEOUT_MILLIS`
- timeout
- `SO_TIMEOUT`IOIOacceptread`SO_TIMEOUT`
```java
// SocketChannel .option(xxx, xxx)
//
// ServerSocketChannel .option(xxx, xxx)
// SockerChannel childOption(xxx, xxx)
```
#### SO_BACKLOG
WindowsSO_BACKLOG200128
NettyLinux`proc/ysy/net/core/somaxconn`
#### ulimit -n
#### TCP_NODELAY
SocketChannelNaglefalseNagletrueNagle
#### SO_SENDBUF&SO_RCVBUF
- SO_SENDBUFSocketChannel
- SO_RCVBUFSocketChannelServerSocketChannelServerSocketChannel
- SO_SNDBUFTCP****
- SO_RCVBUFTCP****
`SO_SNDBUF``SO_RCVBUF``SO_SNDBUF``SO_RCVBUF`
TCP
#### ALLOCATOR
SocketChannelallocator
ByteBufctx.alloc()
#### RCVBUF_ALLOCATOR
SocketChannel
Netty
directallocator
##
###
JavaNIONetty
```java
// 1Selector
Selector selector = Selector.open();
// NioServerSocketChannel attachment = new NioServerSocketChannel();
// 2ServerSocketChannel
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
//
serverSocketChannel.configureBlocking(false);
// SelectionKey selectionKey = serverSocketChannel.register(selector, 0, attachment);
// 3 serverSocketChannel selector
SelectionKey selectionKey = serverSocketChannel.register(selector, 0);
// 4SelectionKey
selectionKey.interestOps(SelectionKey.OP_ACCEPT);
// 5
serverSocketChannel.bind(new InetSocketAddress(8080));
while(true) {
selector.select();
Iterator keys = selector.selectedKeys().iterator();
while(keys.hasNext()) {
SelectionKey key = keys.next();
keys.remove();
if (key.isAcceptable()) {
ServerSocketChannel channel = (ServerSocketChannel)key.channel();
SocketChannel sc = channel.accept();
sc.configureBlocking(false);
SelectionKey scKey = sc.register(selector, 0);
scKey.interestOps(SelectionKey.OP_READ);
}
if (key.isReadable()) {
try {
SocketChannel channel = (SocketChannel)key.channel();
ByteBuffer buffer = ByteBuffer.allocate(16);
// -1
int read = channel.read(buffer);
if (read == -1) {
key.cancel();
continue;
}
buffer.flip();
......
} catch(Exception e) {
// Selector
key.cancel();
}
}
}
}
```
Netty
```java
new ServerBootstrap()
// NioEventLoopGroup nio
.group(new NioEventLoopGroup())
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer() {
@Override
protected void initChannel(NioSocketChannelch) throws Exception {
ch.pipeline().addLast(new SimpleChannelInboundHandler() {
@Override
protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("Netty!");
}
});
}
}).bind(8080);
```
NettyJavaNIONioEventLoopGroupbind
###
#### init
```java
public ChannelFuture bind(int inetPort) {
return bind(new InetSocketAddress(inetPort));
}
public ChannelFuture bind(SocketAddress localAddress) {
validate();
return doBind(ObjectUtil.checkNotNull(localAddress, "localAddress"));
}
private ChannelFuture doBind(final SocketAddress localAddress) {
// 1initAndRegister
final ChannelFuture regFuture = initAndRegister();
final Channel channel = regFuture.channel();
if (regFuture.cause() != null) {
return regFuture;
}
if (regFuture.isDone()) {
// At this point we know that the registration was complete and successful.
ChannelPromise promise = channel.newPromise();
// 2 ServerSocketChannel bind
doBind0(regFuture, channel, localAddress, promise);
return promise;
} else {
// Registration future is almost always fulfilled already, but just in case it's not.
final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
/*
initAndRegister()Nio Thread
*/
regFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
Throwable cause = future.cause();
if (cause != null) {
// Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an
// IllegalStateException once we try to access the EventLoop of the Channel.
promise.setFailure(cause);
} else {
// Registration was successful, so set the correct executor to use.
// See https://github.com/netty/netty/issues/2586
promise.registered();
// 2 ServerSocketChannel bind
doBind0(regFuture, channel, localAddress, promise);
}
}
});
return promise;
}
}
```
initAndRegister()
1. `ServerSocketChannel.open();`
2. `serverSocketChannel.register(selector, 0);`
```java
final ChannelFuture initAndRegister() {
Channel channel = null;
try {
/*
1ServerSocketChannel serverSocketChannel = ServerSocketChannel.open()
NioServerSocketChannel
ServerSocketChannel.open()
*/
channel = channelFactory.newChannel();
// NioServerSocketChannel
init(channel);
} catch (Throwable t) {
if (channel != null) {
// channel can be null if newChannel crashed (eg SocketException("too many open files"))
channel.unsafe().closeForcibly();
// as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
}
// as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t);
}
/*
2serverSocketChannel.register(selector, 0, att)
ServerSocketChannel Selector
*/
ChannelFuture regFuture = config().group().register(channel);
if (regFuture.cause() != null) {
if (channel.isRegistered()) {
channel.close();
} else {
channel.unsafe().closeForcibly();
}
}
return regFuture;
}
```
doBind0()
- `serverSocketChannel.bind(new InetSocketAddress(8080));`
#### register
```java
// ChannelFuture regFuture = config().group().register(channel);
// io.netty.channel.MultithreadEventLoopGroup#register
public ChannelFuture register(Channel channel) {
return next().register(channel);
}
// io.netty.channel.SingleThreadEventLoop#register
public ChannelFuture register(Channel channel) {
return register(new DefaultChannelPromise(channel, this));
}
public ChannelFuture register(final ChannelPromise promise) {
ObjectUtil.checkNotNull(promise, "promise");
promise.channel().unsafe().register(this, promise);
return promise;
}
// io.netty.channel.AbstractChannel.AbstractUnsafe#register
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
ObjectUtil.checkNotNull(eventLoop, "eventLoop");
if (isRegistered()) {
promise.setFailure(new IllegalStateException("registered to an event loop already"));
return;
}
if (!isCompatible(eventLoop)) {
promise.setFailure(
new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
return;
}
AbstractChannel.this.eventLoop = eventLoop;
// nio thread else
if (eventLoop.inEventLoop()) {
register0(promise);
} else {
try {
/*
eventLoop nio thread
execute() eventLoop nio thread
nio boss
*/
eventLoop.execute(new Runnable() {
@Override
public void run() {
// ServerSocketChannel Selector
register0(promise);
}
});
} catch (Throwable t) {
logger.warn(
"Force-closing a channel whose registration task was not accepted by an event loop: {}",
AbstractChannel.this, t);
closeForcibly();
closeFuture.setClosed();
safeSetFailure(promise, t);
}
}
}
// io.netty.channel.AbstractChannel.AbstractUnsafe#register0
private void register0(ChannelPromise promise) {
try {
// check if the channel is still open as it could be closed in the mean time when the register
// call was outside of the eventLoop
if (!promise.setUncancellable() || !ensureOpen(promise)) {
return;
}
boolean firstRegistration = neverRegistered;
/*
(AbstractNioChannel)
do
ServerSocketChannel Selector
serverSocketChannel.register(selector, 0)
*/
doRegister();
neverRegistered = false;
registered = true;
// Ensure we call handlerAdded(...) before we actually notify the promise. This is needed as the
// user may already fire events through the pipeline in the ChannelFutureListener.
/*
init(channel)
NioServerSockerChannel handler ChannelInitializer
*/
pipeline.invokeHandlerAddedIfNeeded();
safeSetSuccess(promise);
// channelRegistered()
pipeline.fireChannelRegistered();
// Only fire a channelActive if the channel has never been registered. This prevents firing
// multiple channel actives if the channel is deregistered and re-registered.
if (isActive()) {
if (firstRegistration) {
pipeline.fireChannelActive();
} else if (config().isAutoRead()) {
// This channel was registered before and autoRead() is set. This means we need to begin read
// again so that we process inbound data.
//
// See https://github.com/netty/netty/issues/4805
beginRead();
}
}
} catch (Throwable t) {
// Close the channel directly to avoid FD leak.
closeForcibly();
closeFuture.setClosed();
safeSetFailure(promise, t);
}
}
```
AbstractChannel`io.netty.channel.nio.AbstractNioChannel#doRegister`Javanio
```java
protected void doRegister() throws Exception {
boolean selected = false;
for (;;) {
try {
/*
Java nio serverSocketChannel.register(selector, 0, att);
1SelectoreventLoopeventLoop().unwrappedSelector()Selector
2
3Netty this NioServerSocketChannelselector NioServerSocketChannel
*/
selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
return;
} catch (CancelledKeyException e) {
......
}
}
}
protected SelectableChannel javaChannel() {
return ch;
}
```
Java nio
```java
/*
public abstract class ServerSocketChannel extends AbstractSelectableChannel implements NetworkChannel
public abstract class AbstractSelectableChannel extends SelectableChannel
*/
// SelectionKey selectionKey = serverSocketChannel.register(selector, 0);
// java.nio.channels.SelectableChannel#register
public final SelectionKey register(Selector sel, int ops)
throws ClosedChannelException
{
return register(sel, ops, null);
}
// java.nio.channels.spi.AbstractSelectableChannel#register
public final SelectionKey register(Selector sel, int ops,
Object att)
throws ClosedChannelException
{
synchronized (regLock) {
if (!isOpen())
throw new ClosedChannelException();
if ((ops & ~validOps()) != 0)
throw new IllegalArgumentException();
if (blocking)
throw new IllegalBlockingModeException();
SelectionKey k = findKey(sel);
if (k != null) {
k.interestOps(ops);
k.attach(att);
}
if (k == null) {
// New registration
synchronized (keyLock) {
if (!isOpen())
throw new ClosedChannelException();
k = ((AbstractSelector)sel).register(this, ops, att);
addKey(k);
}
}
return k;
}
}
```
Java Nio Netty ServerSocketChannelSelector
#### doBind0
initAndRegister()regFuturedoBind0()`regFuture.addListener(new ChannelFutureListener() {...doBind0()...})` ServerSocketChannel
```java
// io.netty.bootstrap.AbstractBootstrap#doBind0
private static void doBind0(
final ChannelFuture regFuture, final Channel channel,
final SocketAddress localAddress, final ChannelPromise promise) {
// This method is invoked before channelRegistered() is triggered. Give user handlers a chance to set up
// the pipeline in its channelRegistered() implementation.
// eventLoop nio
channel.eventLoop().execute(new Runnable() {
@Override
public void run() {
if (regFuture.isSuccess()) {
//
channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
} else {
promise.setFailure(regFuture.cause());
}
}
});
}
// io.netty.channel.AbstractChannel#bind
public ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
return pipeline.bind(localAddress, promise);
}
// io.netty.channel.DefaultChannelPipeline#bind
public final ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
return tail.bind(localAddress, promise);
}
// io.netty.channel.AbstractChannelHandlerContext#bind
public ChannelFuture bind(final SocketAddress localAddress, final ChannelPromise promise) {
ObjectUtil.checkNotNull(localAddress, "localAddress");
if (isNotValidPromise(promise, false)) {
// cancelled
return promise;
}
final AbstractChannelHandlerContext next = findContextOutbound(MASK_BIND);
EventExecutor executor = next.executor();
if (executor.inEventLoop()) {
//
next.invokeBind(localAddress, promise);
} else {
safeExecute(executor, new Runnable() {
@Override
public void run() {
next.invokeBind(localAddress, promise);
}
}, promise, null, false);
}
return promise;
}
private void invokeBind(SocketAddress localAddress, ChannelPromise promise) {
if (invokeHandler()) {
try {
// DefaultChannelPipeline.HeadContext#bind
((ChannelOutboundHandler) handler()).bind(this, localAddress, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
}
} else {
bind(localAddress, promise);
}
}
// io.netty.channel.DefaultChannelPipeline.HeadContext#bind
public void bind(
ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) {
unsafe.bind(localAddress, promise);
}
// io.netty.channel.AbstractChannel.AbstractUnsafe#bind
public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
assertEventLoop();
if (!promise.setUncancellable() || !ensureOpen(promise)) {
return;
}
// See: https://github.com/netty/netty/issues/576
if (Boolean.TRUE.equals(config().getOption(ChannelOption.SO_BROADCAST)) &&
localAddress instanceof InetSocketAddress &&
!((InetSocketAddress) localAddress).getAddress().isAnyLocalAddress() &&
!PlatformDependent.isWindows() && !PlatformDependent.maybeSuperUser()) {
// Warn a user about the fact that a non-root user can't receive a
// broadcast packet on *nix if the socket is bound on non-wildcard address.
logger.warn(
"A non-root user can't receive a broadcast packet if the socket " +
"is not bound to a wildcard address; binding to a non-wildcard " +
"address (" + localAddress + ") anyway as requested.");
}
boolean wasActive = isActive();
try {
// ServerSocketChannel
doBind(localAddress);
} catch (Throwable t) {
safeSetFailure(promise, t);
closeIfClosed();
return;
}
// ServerSocketChannelactive
if (!wasActive && isActive()) {
invokeLater(new Runnable() {
@Override
public void run() {
pipeline.fireChannelActive();
}
});
}
safeSetSuccess(promise);
}
```
NioServerSocketChannelbind()Java nio bind
```java
// io.netty.channel.socket.nio.NioServerSocketChannel#doBind
protected void doBind(SocketAddress localAddress) throws Exception {
// JDK1.8
if (PlatformDependent.javaVersion() >= 7) {
// serverSocketChannel.bind(new InetSocketAddress(8080));
javaChannel().bind(localAddress, config.getBacklog());
} else {
javaChannel().socket().bind(localAddress, config.getBacklog());
}
}
// javaChannel()sun.nio.ch.ServerSocketChannelImpl
```
Java Nio Netty bind()
#### accept
```java
/*
ServerSocketChannelactive
channelpipeline handlerchannelhandler
pipeline3handlerhead -> ServerBootstrapAcceptor -> tail
headtailhandlerChannel pipeline NioServerSocketChannel
ServerBootstrapAcceptorNeety
*/
if (!wasActive && isActive()) {
invokeLater(new Runnable() {
@Override
public void run() {
// pipelinehandler channelActive()
// headHeadContext
// NioServerSocketChannel channelActive
pipeline.fireChannelActive();
}
});
}
```
ServerSocketChannel`selectionKey.interestOps(SelectionKey.OP_ACCEPT)`pipeline handler
```java
// io.netty.channel.DefaultChannelPipeline#fireChannelActive
public final ChannelPipeline fireChannelActive() {
AbstractChannelHandlerContext.invokeChannelActive(head);
return this;
}
// io.netty.channel.AbstractChannelHandlerContext#invokeChannelActive
static void invokeChannelActive(final AbstractChannelHandlerContext next) {
EventExecutor executor = next.executor();
// nio if
if (executor.inEventLoop()) {
next.invokeChannelActive();
} else {
executor.execute(new Runnable() {
@Override
public void run() {
next.invokeChannelActive();
}
});
}
}
private void invokeChannelActive() {
if (invokeHandler()) {
try {
// channelActive()
((ChannelInboundHandler) handler()).channelActive(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
} else {
fireChannelActive();
}
}
```
pipelinehead`DefaultChannelPipeline.HeadContext`
```java
public class DefaultChannelPipeline implements ChannelPipeline {
final class HeadContext extends AbstractChannelHandlerContext
implements ChannelOutboundHandler, ChannelInboundHandler {
public void channelActive(ChannelHandlerContext ctx) {
ctx.fireChannelActive();
// selectionKey.interestOps(SelectionKey.OP_ACCEPT);
readIfIsAutoRead();
}
private void readIfIsAutoRead() {
if (channel.config().isAutoRead()) {
channel.read();
}
}
}
}
// io.netty.channel.AbstractChannel#read
public Channel read() {
pipeline.read();
return this;
}
// io.netty.channel.DefaultChannelPipeline#read
public final ChannelPipeline read() {
tail.read();
return this;
}
// io.netty.channel.AbstractChannelHandlerContext#read
public ChannelHandlerContext read() {
final AbstractChannelHandlerContext next = findContextOutbound(MASK_READ);
EventExecutor executor = next.executor();
if (executor.inEventLoop()) {
next.invokeRead();
} else {
Tasks tasks = next.invokeTasks;
if (tasks == null) {
next.invokeTasks = tasks = new Tasks(next);
}
executor.execute(tasks.invokeReadTask);
}
return this;
}
private void invokeRead() {
if (invokeHandler()) {
try {
((ChannelOutboundHandler) handler()).read(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
} else {
read();
}
}
// io.netty.channel.DefaultChannelPipeline.HeadContext#read
public void read(ChannelHandlerContext ctx) {
unsafe.beginRead();
}
// io.netty.channel.AbstractChannel.AbstractUnsafe#beginRead
public final void beginRead() {
assertEventLoop();
try {
doBeginRead();
} catch (final Exception e) {
invokeLater(new Runnable() {
@Override
public void run() {
pipeline.fireExceptionCaught(e);
}
});
close(voidPromise());
}
}
// io.netty.channel.nio.AbstractNioMessageChannel#doBeginRead
protected void doBeginRead() throws Exception {
if (inputShutdown) {
return;
}
super.doBeginRead();
}
```
AbstractNioChanneldoBeginRead()`SelectionKey.OP_ACCEPT`selectionKey
```java
// io.netty.channel.nio.AbstractNioChannel#doBeginRead
protected void doBeginRead() throws Exception {
// Channel.read() or ChannelHandlerContext.read() was called
final SelectionKey selectionKey = this.selectionKey;
if (!selectionKey.isValid()) {
return;
}
readPending = true;
/*
SelectionKey.OP_ACCEPT SelectionKey.OP_ACCEPT
readInterestOp NioServerSocketChannel
*/
final int interestOps = selectionKey.interestOps();
if ((interestOps & readInterestOp) == 0) {
// selectionKey.interestOps(SelectionKey.OP_ACCEPT);
selectionKey.interestOps(interestOps | readInterestOp);
}
}
```
Java Nio Netty accept
Java nio Netty
### EventLoop
- NioEventLoopSelector
- NioEventLoop io
```java
//
public final class NioEventLoop extends SingleThreadEventLoop {
private Selector selector;
private Selector unwrappedSelector;
/*
thread
IO
*/
private final Queue taskQueue;
/*
SingleThreadEventLoopSingleThreadEventExecutor
IO
*/
private volatile Thread thread;
/*
SingleThreadEventLoopSingleThreadEventExecutor
thread
*/
private final Executor executor;
/*
SingleThreadEventLoopSingleThreadEventExecutorAbstractScheduledEventExecutor
*/
PriorityQueue selectorImplClass = (Class) maybeSelectorImplClass;
// set
final SelectedSelectionKeySet selectedKeySet = new SelectedSelectionKeySet();
Object maybeException = AccessController.doPrivileged(new PrivilegedAction() {
@Override
public Object run() {
try {
Field selectedKeysField = selectorImplClass.getDeclaredField("selectedKeys");
Field publicSelectedKeysField = selectorImplClass.getDeclaredField("publicSelectedKeys");
if (PlatformDependent.javaVersion() >= 9 && PlatformDependent.hasUnsafe()) {
// Let us try to use sun.misc.Unsafe to replace the SelectionKeySet.
// This allows us to also do this in Java9+ without any extra flags.
long selectedKeysFieldOffset = PlatformDependent.objectFieldOffset(selectedKeysField);
long publicSelectedKeysFieldOffset =
PlatformDependent.objectFieldOffset(publicSelectedKeysField);
if (selectedKeysFieldOffset != -1 && publicSelectedKeysFieldOffset != -1) {
PlatformDependent.putObject(
unwrappedSelector, selectedKeysFieldOffset, selectedKeySet);
PlatformDependent.putObject(
unwrappedSelector, publicSelectedKeysFieldOffset, selectedKeySet);
return null;
}
// We could not retrieve the offset, lets try reflection as last-resort.
}
Throwable cause = ReflectionUtil.trySetAccessible(selectedKeysField, true);
if (cause != null) {
return cause;
}
cause = ReflectionUtil.trySetAccessible(publicSelectedKeysField, true);
if (cause != null) {
return cause;
}
/*
selectedKeysSelectedSelectionKeySet)HashSetHashMap
publicSelectedKeysSelectedSelectionKeySet)HashSetHashMap
*/
selectedKeysField.set(unwrappedSelector, selectedKeySet);
publicSelectedKeysField.set(unwrappedSelector, selectedKeySet);
return null;
} catch (NoSuchFieldException e) {
return e;
} catch (IllegalAccessException e) {
return e;
}
}
});
if (maybeException instanceof Exception) {
selectedKeys = null;
Exception e = (Exception) maybeException;
logger.trace("failed to instrument a special java.util.Set into: {}", unwrappedSelector, e);
return new SelectorTuple(unwrappedSelector);
}
selectedKeys = selectedKeySet;
logger.trace("instrumented a special java.util.Set into: {}", unwrappedSelector);
return new SelectorTuple(unwrappedSelector,
new SelectedSelectionKeySetSelector(unwrappedSelector, selectedKeySet));
}
```
openSelector() Selector SelectedSelectionKeySet Selector
- selectedKey
- Selector
#### nio thread
```java
EventLoop eventLoop = new NioEventLoopGroup().next();
eventLoop.execute(() -> {
System.out.println("hello");
});
```
nio thread
```java
// io.netty.util.concurrent.SingleThreadEventExecutor#execute
public void execute(Runnable task) {
execute0(task);
}
private void execute0(@Schedule Runnable task) {
//
ObjectUtil.checkNotNull(task, "task");
execute(task, !(task instanceof LazyRunnable) && wakesUpForTask(task));
}
private void execute(Runnable task, boolean immediate) {
// EventLoop
boolean inEventLoop = inEventLoop();
//
addTask(task);
// inEventLoopfalsetrueif
if (!inEventLoop) {
//
startThread();
if (isShutdown()) {
boolean reject = false;
try {
if (removeTask(task)) {
reject = true;
}
} catch (UnsupportedOperationException e) {
// The task queue does not support removal so the best thing we can do is to just move on and
// hope we will be able to pick-up the task before its completely terminated.
// In worst case we will log on termination.
}
if (reject) {
reject();
}
}
}
if (!addTaskWakesUp && immediate) {
wakeup(inEventLoop);
}
}
```
`startThread()` nio thread
```java
// NioEventLoop
public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {
private static final int ST_NOT_STARTED = 1;
private static final int ST_STARTED = 2;
private void startThread() {
//
if (state == ST_NOT_STARTED) {
// cas casnio threadexecute()cas
if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
boolean success = false;
try {
//
doStartThread();
success = true;
} finally {
if (!success) {
STATE_UPDATER.compareAndSet(this, ST_STARTED, ST_NOT_STARTED);
}
}
}
}
}
private void doStartThread() {
assert thread == null;
executor.execute(new Runnable() {
@Override
public void run() {
/*
executornio thread
executor EventLoop thread
*/
thread = Thread.currentThread();
if (interrupted) {
thread.interrupt();
}
boolean success = false;
updateLastExecutionTime();
try {
/*
NioEventLoop run Run the tasks in the taskQueue
IO
*/
SingleThreadEventExecutor.this.run();
success = true;
} catch (Throwable t) {
logger.warn("Unexpected exception from an event executor: ", t);
} finally {
... ...
}
}
});
}
}
```
- EventLoop#execute() nio thread
- state
#### select
```java
// io.netty.channel.nio.NioEventLoop#run
protected void run() {
int selectCnt = 0;
for (;;) {
try {
int strategy;
try {
strategy = selectStrategy.calculateStrategy(selectNowSupplier, hasTasks());
switch (strategy) {
case SelectStrategy.CONTINUE:
continue;
case SelectStrategy.BUSY_WAIT:
// fall-through to SELECT since the busy-wait is not supported with NIO
case SelectStrategy.SELECT:
long curDeadlineNanos = nextScheduledTaskDeadlineNanos();
if (curDeadlineNanos == -1L) {
curDeadlineNanos = NONE; // nothing on the calendar
}
nextWakeupNanos.set(curDeadlineNanos);
try {
if (!hasTasks()) {
/*
select()
wakeup()
*/
strategy = select(curDeadlineNanos);
}
} finally {
// This update is just to help block unnecessary selector wakeups
// so use of lazySet is ok (no race condition)
nextWakeupNanos.lazySet(AWAKE);
}
// fall through
default:
}
} catch (IOException e) {
......
}
selectCnt++;
cancelledKeys = 0;
needsToSelectAgain = false;
final int ioRatio = this.ioRatio;
boolean ranTasks;
......
} catch (CancelledKeyException e) {
......
} catch (Error e) {
throw e;
} catch (Throwable t) {
handleLoopException(t);
} finally {
......
}
}
}
private int select(long deadlineNanos) throws IOException {
if (deadlineNanos == NONE) {
return selector.select();
}
// Timeout will only be 0 if deadline is within 5 microsecs
long timeoutMillis = deadlineToDelayNanos(deadlineNanos + 995000L) / 1000000L;
return timeoutMillis MIN_PREMATURE_SELECTOR_RETURNS && logger.isDebugEnabled()) {
logger.debug("Selector.select() returned prematurely {} times in a row for Selector {}.",
selectCnt - 1, selector);
}
selectCnt = 0;
// selectbug512 rebuild Selector
} else if (unexpectedSelectorWakeup(selectCnt)) { // Unexpected wakeup (unusual case)
selectCnt = 0;
}
}
......
}
}
private boolean unexpectedSelectorWakeup(int selectCnt) {
if (Thread.interrupted()) {
// Thread was interrupted so reset selected keys and break so we not run into a busy loop.
// As this is most likely a bug in the handler of the user or it's client library we will
// also log it.
//
// See https://github.com/netty/netty/issues/2426
if (logger.isDebugEnabled()) {
logger.debug("Selector.select() returned prematurely because " +
"Thread.currentThread().interrupt() was called. Use " +
"NioEventLoop.shutdownGracefully() to shutdown the NioEventLoop.");
}
return true;
}
if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&
selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
// The selector returned prematurely many times in a row.
// Rebuild the selector to work around the problem.
logger.warn("Selector.select() returned prematurely {} times in a row; rebuilding Selector {}.",
selectCnt, selector);
// SelectorbugSelectorSelector selectedKeys Selector
rebuildSelector();
return true;
}
return false;
}
```
- 1NettySelectorbugSelector
- 2Selector
- bugJDKLinuxselectorbug
- bugselect()
#### ioRatio100
- ioRatio io 50%50%IO50%
```java
protected void run() {
int selectCnt = 0;
for (;;) {
try {
int strategy;
// ...... strategy = select(curDeadlineNanos);
selectCnt++;
cancelledKeys = 0;
needsToSelectAgain = false;
final int ioRatio = this.ioRatio;
boolean ranTasks;
// 100
if (ioRatio == 100) {
try {
if (strategy > 0) {
processSelectedKeys();
}
} finally {
// Ensure we always run tasks.
ranTasks = runAllTasks();
}
} else if (strategy > 0) {
final long ioStartTime = System.nanoTime();
try {
//
processSelectedKeys();
} finally {
// finally
// Ensure we always run tasks.
/*
ioTime io
IO8s ioTime=8ioRatio=80
ioTime * (100 - ioRatio) / ioRatio = 8*20/80 = 2s
22
*/
final long ioTime = System.nanoTime() - ioStartTime;
ranTasks = runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
}
} else {
ranTasks = runAllTasks(0); // This will run the minimum number of tasks
}
if (ranTasks || strategy > 0) {
if (selectCnt > MIN_PREMATURE_SELECTOR_RETURNS && logger.isDebugEnabled()) {
logger.debug("Selector.select() returned prematurely {} times in a row for Selector {}.",
selectCnt - 1, selector);
}
selectCnt = 0;
} else if (unexpectedSelectorWakeup(selectCnt)) { // Unexpected wakeup (unusual case)
selectCnt = 0;
}
}
......
}
}
```
- ioRatio io
- ioRatio=100IO
- ioRatio=100
#### selectedKeys
`EventLoopselector?`selectedKeys
Netty
```java
// SelectedKeys
// io.netty.channel.nio.NioEventLoop#processSelectedKeys
private void processSelectedKeys() {
// selectedKeys != null NettyselectedKeysHashSet
if (selectedKeys != null) {
//
processSelectedKeysOptimized();
} else {
// HashSet
processSelectedKeysPlain(selector.selectedKeys());
}
}
private void processSelectedKeysOptimized() {
for (int i = 0; i < selectedKeys.size; ++i) {
final SelectionKey k = selectedKeys.keys[i];
// null out entry in the array to allow to have it GC'ed once the Channel close
// See https://github.com/netty/netty/issues/2363
selectedKeys.keys[i] = null;
/*
NioServerSocketChannelNioServerSocketChannelselectedKeys
NioServerSocketChannelNioServerSocketChannelpipeline
pipelinehandlerhandlerselectedKeys
*/
final Object a = k.attachment();
// AbstractNioChannelNioChannelif
if (a instanceof AbstractNioChannel) {
//
processSelectedKey(k, (AbstractNioChannel) a);
} else {
@SuppressWarnings("unchecked")
NioTask task = (NioTask) a;
processSelectedKey(k, task);
}
if (needsToSelectAgain) {
// null out entries in the array to allow to have it GC'ed once the Channel close
// See https://github.com/netty/netty/issues/2363
selectedKeys.reset(i + 1);
selectAgain();
i = -1;
}
}
}
```
####
```java
// io.netty.channel.nio.NioEventLoop#processSelectedKey
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
if (!k.isValid()) {
final EventLoop eventLoop;
try {
eventLoop = ch.eventLoop();
} catch (Throwable ignored) {
// If the channel implementation throws an exception because there is no event loop, we ignore this
// because we are only trying to determine if ch is registered to this event loop and thus has authority
// to close ch.
return;
}
// Only close ch if ch is still registered to this EventLoop. ch could have deregistered from the event loop
// and thus the SelectionKey could be cancelled as part of the deregistration process, but the channel is
// still healthy and should not be closed.
// See https://github.com/netty/netty/issues/5125
if (eventLoop == this) {
// close the channel if the key is not valid anymore
unsafe.close(unsafe.voidPromise());
}
return;
}
try {
int readyOps = k.readyOps();
// We first need to call finishConnect() before try to trigger a read(...) or write(...) as otherwise
// the NIO JDK channel implementation may throw a NotYetConnectedException.
//
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
// remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
// See https://github.com/netty/netty/issues/924
int ops = k.interestOps();
ops &= ~SelectionKey.OP_CONNECT;
k.interestOps(ops);
unsafe.finishConnect();
}
// Process OP_WRITE first as we may be able to write some queued buffers and so free memory.
//
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
// Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
ch.unsafe().forceFlush();
}
// Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
// to a spin loop
// readyOps1OP_READ16OP_ACCEPT
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
// accept
unsafe.read();
}
} catch (CancelledKeyException ignored) {
unsafe.close(unsafe.voidPromise());
}
}
```
### accept
nioaccept
1. selector.select()
2. selectedKeys
3. accept
4. SocketChannel
5. SocketChannelselector
6. selectionKeyread
Netty1~34~6
```java
private final class NioMessageUnsafe extends AbstractNioUnsafe {
private final List readBuf = new ArrayList();
/*
acceptread
accept
*/
@Override
public void read() {
assert eventLoop().inEventLoop();
final ChannelConfig config = config();
final ChannelPipeline pipeline = pipeline();
final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
allocHandle.reset(config);
boolean closed = false;
Throwable exception = null;
try {
try {
do {
/*
4SocketChannel
readBufNioServerSocketChannelpipeline
*/
int localRead = doReadMessages(readBuf);
if (localRead == 0) {
break;
}
if (localRead < 0) {
closed = true;
break;
}
allocHandle.incMessagesRead(localRead);
} while (continueReading(allocHandle));
} catch (Throwable t) {
exception = t;
}
int size = readBuf.size();
for (int i = 0; i < size; i ++) {
readPending = false;
/*
NioServerSocketChannelpipelinehandlerhandler
head -> acceptor(ServerBootstrapAcceptor) -> tail
acceptServerBootstrapAcceptor handler
handler channelRead()
*/
pipeline.fireChannelRead(readBuf.get(i));
}
readBuf.clear();
allocHandle.readComplete();
pipeline.fireChannelReadComplete();
if (exception != null) {
closed = closeOnReadError(exception);
pipeline.fireExceptionCaught(exception);
}
if (closed) {
inputShutdown = true;
if (isOpen()) {
close(voidPromise());
}
}
} finally {
// Check if there is a readPending which was not processed yet.
// This could be for two reasons:
// * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
// * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
//
// See https://github.com/netty/netty/issues/2254
if (!readPending && !config.isAutoRead()) {
removeReadOp();
}
}
}
}
```
#### SocketChannel
```java
// io.netty.channel.socket.nio.NioServerSocketChannel#doReadMessages
protected int doReadMessages(List buf) throws Exception {
// serverSocketChannel.accept()nio
SocketChannel ch = SocketUtils.accept(javaChannel());
try {
if (ch != null) {
/*
new NioSocketChannel(this, ch)
NettyNioSocketChannelJavaSocketChannelSocketChannel
NioSocketChannelpipelinehandler
*/
buf.add(new NioSocketChannel(this, ch));
return 1;
}
} catch (Throwable t) {
logger.warn("Failed to create a new channel from an accepted socket.", t);
try {
ch.close();
} catch (Throwable t2) {
logger.warn("Failed to close a socket.", t2);
}
}
return 0;
}
```
#### SocketChannelselector
```java
public class ServerBootstrap extends AbstractBootstrap {
private static class ServerBootstrapAcceptor extends ChannelInboundHandlerAdapter {
......
/*
msg NioSocketChannel
*/
public void channelRead(ChannelHandlerContext ctx, Object msg) {
final Channel child = (Channel) msg;
child.pipeline().addLast(childHandler);
setChannelOptions(child, childOptions, logger);
setAttributes(child, childAttrs);
try {
/*
register()NioServerSocketChannel
`.childHandler(new ChannelInitializer() {}`
SocketChannel
register()
5. SocketChannelselector
sc.register(eventLoop, 0, NioSocketChannel)
6. selectionKeyread
pipelinehead -> handler -> tail
*/
childGroup.register(child).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (!future.isSuccess()) {
forceClose(child, future.cause());
}
}
});
} catch (Throwable t) {
forceClose(child, t);
}
}
......
}
}
```
### read
nioread
1. selector.select()
2. selectedKeys
3. read
4.
Netty1~34
```java
public abstract class AbstractNioByteChannel extends AbstractNioChannel {
protected class NioByteUnsafe extends AbstractNioUnsafe {
......
@Override
public final void read() {
final ChannelConfig config = config();
if (shouldBreakReadReady(config)) {
clearReadPending();
return;
}
final ChannelPipeline pipeline = pipeline();
// ByteBufPooledByteBufAllocator
final ByteBufAllocator allocator = config.getAllocator();
final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
allocHandle.reset(config);
ByteBuf byteBuf = null;
boolean close = false;
try {
do {
// ByteBufiobytebuf
byteBuf = allocHandle.allocate(allocator);
// socketChannel
allocHandle.lastBytesRead(doReadBytes(byteBuf));
if (allocHandle.lastBytesRead() handler -> tail
*/
pipeline.fireChannelRead(byteBuf);
byteBuf = null;
} while (allocHandle.continueReading());
allocHandle.readComplete();
pipeline.fireChannelReadComplete();
if (close) {
closeOnRead(pipeline);
}
} catch (Throwable t) {
handleReadException(pipeline, byteBuf, t, close, allocHandle);
} finally {
// Check if there is a readPending which was not processed yet.
// This could be for two reasons:
// * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
// * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
//
// See https://github.com/netty/netty/issues/2254
if (!readPending && !config.isAutoRead()) {
removeReadOp();
}
}
}
}
}
```
##
### ChannelHandler
ChannelHandlerChannelInboundHandlerChannelOutboundHandler
ChannelHandlerChannelHandlerContext
ChannelHandlerContextfireXXX()write(msg)
ChannelPipeline
NettyChannelPipelineChannelPipelineChannelHandlerChannelHandlerChannelHandler
Netty`@Sharable`ChannelHandler`@Sharable`
### ChannelPipeline
Netty`Channel``ChannelHandler`Channel
`ChannelPipeline`ChannelHandlerChannelChannelPipelineChannelHandlerChannelPipeline
Netty`ChannelEvent``MessageEvent``ChannelStateEvent`ChannelChannelEvent`ChannelPipeline`ChannelPipelineChannelHandlerChannelHandlerChannelEventChannelHandler
### @Shareable
`@Shareable`handler`@Shareable`
###
Netty`IdleStateHandler`
```java
public class IdleStateHandler extends ChannelDuplexHandler {
public IdleStateHandler(
int readerIdleTimeSeconds,
int writerIdleTimeSeconds,
int allIdleTimeSeconds) {
this(readerIdleTimeSeconds, writerIdleTimeSeconds, allIdleTimeSeconds,
TimeUnit.SECONDS);
}
public IdleStateHandler(
long readerIdleTime, long writerIdleTime, long allIdleTime,
TimeUnit unit) {
this(false, readerIdleTime, writerIdleTime, allIdleTime, unit);
}
......
}
```
- `readerIdleTimeSeconds`. Channel , READER_IDLE IdleStateEvent .
- `writerIdleTimeSeconds`:. Channel , WRITER_IDLE IdleStateEvent .
- `allIdleTimeSeconds`/. , ALL_IDLE IdleStateEvent
1. pipeline().addLast()ChannelHandlerhandlerAdded()IdleStateHandlerhandlerAdded()IdleStateHandlerhandlerAdded()
```java
//
ch.pipeline().addLast(new IdleStateHandler(1, 1, 1));
// io.netty.channel.DefaultChannelPipeline#addLast
public final ChannelPipeline addLast(ChannelHandler... handlers) {
return addLast(null, handlers);
}
public final ChannelPipeline addLast(EventExecutorGroup executor, ChannelHandler... handlers) {
ObjectUtil.checkNotNull(handlers, "handlers");
for (ChannelHandler h: handlers) {
if (h == null) {
break;
}
addLast(executor, null, h);
}
return this;
}
public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
final AbstractChannelHandlerContext newCtx;
synchronized (this) {
checkMultiplicity(handler);
newCtx = newContext(group, filterName(name, handler), handler);
addLast0(newCtx);
// If the registered is false it means that the channel was not registered on an eventLoop yet.
// In this case we add the context to the pipeline and add a task that will call
// ChannelHandler.handlerAdded(...) once the channel is registered.
if (!registered) {
newCtx.setAddPending();
callHandlerCallbackLater(newCtx, true);
return this;
}
EventExecutor executor = newCtx.executor();
if (!executor.inEventLoop()) {
callHandlerAddedInEventLoop(newCtx, executor);
return this;
}
}
callHandlerAdded0(newCtx);
return this;
}
private void callHandlerAdded0(final AbstractChannelHandlerContext ctx) {
try {
//
ctx.callHandlerAdded();
} catch (Throwable t) {
......
}
}
// io.netty.channel.AbstractChannelHandlerContext#callHandlerAdded
final void callHandlerAdded() throws Exception {
// We must call setAddComplete before calling handlerAdded. Otherwise if the handlerAdded method generates
// any pipeline events ctx.handler() will miss them because the state will not allow it.
if (setAddComplete()) {
// handlerAdded()
handler().handlerAdded(this);
}
}
```
2. `IdleStateHandler``handlerAdded()`
```java
// io.netty.handler.timeout.IdleStateHandler#handlerAdded
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
if (ctx.channel().isActive() && ctx.channel().isRegistered()) {
// channelActive() event has been fired already, which means this.channelActive() will
// not be invoked. We have to initialize here instead.
initialize(ctx);
} else {
// channelActive() event has not been fired yet. this.channelActive() will be invoked
// and initialization will occur there.
}
}
private void initialize(ChannelHandlerContext ctx) {
// Avoid the case where destroy() is called before scheduling timeouts.
// See: https://github.com/netty/netty/issues/143
switch (state) {
case 1:
case 2:
return;
default:
break;
}
state = 1;
initOutputChanged(ctx);
lastReadTime = lastWriteTime = ticksInNanos();
if (readerIdleTimeNanos > 0) {
readerIdleTimeout = schedule(ctx, new ReaderIdleTimeoutTask(ctx),
readerIdleTimeNanos, TimeUnit.NANOSECONDS);
}
if (writerIdleTimeNanos > 0) {
writerIdleTimeout = schedule(ctx, new WriterIdleTimeoutTask(ctx),
writerIdleTimeNanos, TimeUnit.NANOSECONDS);
}
if (allIdleTimeNanos > 0) {
allIdleTimeout = schedule(ctx, new AllIdleTimeoutTask(ctx),
allIdleTimeNanos, TimeUnit.NANOSECONDS);
}
}
Future schedule(ChannelHandlerContext ctx, Runnable task, long delay, TimeUnit unit) {
return ctx.executor().schedule(task, delay, unit);
}
```
> 0
Netty
- IdleStateHandlerhandleruserEventTriggered
- IdleStateHandlerEventLoop/
- 3 //
- IdleStateHandler Netty observeOutput
- observeOutput = trueNetty/
- OOMOOMobserveOutput = true
/IdleStateHandlerChannelInboundHandlerChannelOutboundHandlerchannelReadComplete()write()/
#### IdleStateHandler
```java
/*
public class ChannelDuplexHandler extends ChannelInboundHandlerAdapter implements ChannelOutboundHandler
*/
public class IdleStateHandler extends ChannelDuplexHandler {
// /
private final class AllIdleTimeoutTask extends AbstractIdleTask {
AllIdleTimeoutTask(ChannelHandlerContext ctx) {
super(ctx);
}
@Override
protected void run(ChannelHandlerContext ctx) {
long nextDelay = allIdleTimeNanos;
if (!reading) {
nextDelay -= ticksInNanos() - Math.max(lastReadTime, lastWriteTime);
}
if (nextDelay 0 || allIdleTimeNanos > 0) && reading) {
lastReadTime = ticksInNanos();
reading = false;
}
ctx.fireChannelReadComplete();
}
private final ChannelFutureListener writeListener = new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
lastWriteTime = ticksInNanos();
firstWriterIdleEvent = firstAllIdleEvent = true;
}
};
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
// Allow writing with void promise if handler is only configured for read timeout events.
if (writerIdleTimeNanos > 0 || allIdleTimeNanos > 0) {
ctx.write(msg, promise.unvoid()).addListener(writeListener);
} else {
ctx.write(msg, promise);
}
}
}
```
### Java nio
Java NIO
Java NIO `ServerSocketChannel` `Selector` `OP_ACCEPT``OP_READ``OP_WRITE`
- `OP_CONNECT` `SelectionKey.isConnectable()`
- `OP_READ` `SelectionKey.isReadable()` `SocketChannel`
- `OP_WRITE` `SelectionKey.isWritable()` `SocketChannel`
### ByteBuf
```java
public class DefaultChannelPipeline implements ChannelPipeline {
final class HeadContext extends AbstractChannelHandlerContext
implements ChannelOutboundHandler, ChannelInboundHandler {
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
unsafe.write(msg, promise);
}
}
final class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler {
TailContext(DefaultChannelPipeline pipeline) {
super(pipeline, null, TAIL_NAME, TailContext.class);
setAddComplete();
}
......
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
onUnhandledInboundMessage(ctx, msg);
}
......
}
protected void onUnhandledInboundMessage(ChannelHandlerContext ctx, Object msg) {
onUnhandledInboundMessage(msg);
if (logger.isDebugEnabled()) {
logger.debug("Discarded message pipeline : {}. Channel : {}.",
ctx.pipeline().names(), ctx.channel());
}
}
protected void onUnhandledInboundMessage(Object msg) {
try {
logger.debug(
"Discarded inbound message {} that reached at the tail of the pipeline. " +
"Please check your pipeline configuration.", msg);
} finally {
//
ReferenceCountUtil.release(msg);
}
}
}
```
```java
public final void write(Object msg, ChannelPromise promise) {
assertEventLoop();
//
ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
if (outboundBuffer == null) {
try {
// release message now to prevent resource-leak
//
ReferenceCountUtil.release(msg);
} finally {
// If the outboundBuffer is null we know the channel was closed and so
// need to fail the future right away. If it is not null the handling of the rest
// will be done in flush0()
// See https://github.com/netty/netty/issues/2362
safeSetFailure(promise,
newClosedChannelException(initialCloseCause, "write(Object, ChannelPromise)"));
}
return;
}
int size;
try {
msg = filterOutboundMessage(msg);
size = pipeline.estimatorHandle().size(msg);
if (size < 0) {
size = 0;
}
} catch (Throwable t) {
try {
ReferenceCountUtil.release(msg);
} finally {
safeSetFailure(promise, t);
}
return;
}
outboundBuffer.addMessage(msg, size, promise);
}
```
### channelFactory.newChannel()
`constructor``bootstrap.channel(NioServerSocketChannel.class)` NioServerSocketChannel
```java
public class ReflectiveChannelFactory implements ChannelFactory {
public ReflectiveChannelFactory(Class