Super detailed Netty entry, look at this is enough!

This article mainly describes the Netty framework features and important components, hope to have a more intuitive feeling to the Netty framework after reading, hope to help readers quickly get started Netty, reduce some detour.

Mind mapping

preface

Overview of Netty

Why use Netty

• NIO class library and API complex, high learning cost, you need to master Selector, ServerSocketChannel, SocketChannel, ByteBuffer and so on.
• Familiarity with Java multithreaded programming is required. This is because NIO programming involves the Reactor model, and you need to be very familiar with multithreading and network programming to write high quality NIO programs.
• The infamous epoll bug. It causes the Selector to poll for nothing, eventually causing the CPU to be 100%. Until JDK1.7, there was no fundamental solution.

• The API is simple to use and cheap to learn.
• Powerful, built in a variety of decoding encoder, support a variety of protocols.
• High performance. Netty has the best performance compared to other mainstream NIO frameworks.
• Active community, found bugs will be timely repair, iteration cycle is short, constantly add new features.
• Dubbo and Elasticsearch both use Netty, and the quality is proven.

3. Architecture diagram

Hello Word forever

4.1 Introducing Maven dependencies

4.2 Creating a Server Startup Class

Public class MyServer {public static void main(String[] args) throws Exception {// Create two thread groups boosGroup and workerGroup EventLoopGroup bossGroup = new NioEventLoopGroup(); EventLoopGroup workerGroup = new NioEventLoopGroup(); ServerBootstrap = new ServerBootstrap(); ServerBootstrap = new ServerBootstrap(); // Set two thread groups bootstrap.group(bossGroup, WorkerGroup) / / set the service side channel implementation types. The channel (NioServerSocketChannel. Class) / / set the thread queue to get connection number. The option (ChannelOption SO_BACKLOG, ChildOption (channeloption.so_keepalive, channeloption.so_keepalive, ChildHandler (new ChannelInitializer<SocketChannel>() {@override protected void InitChannel (SocketChannel SocketChannel) throws Exception {// Sets the handler socketchannel.pipeline ().addlast (new) to the pipeline pipeline MyServerHandler()); }}); // Set the system.out.println (" Java enthusiast server ready...") handler for workerGroup EventLoop ); ChannelFuture = bootstrap.bind(6666).sync(); ChannelFuture = bootstrap.bind(6666).sync(); Channelfuture.channel ().closeFuture().sync(); } finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); }}}Copy the code

4.3 Creating a Server Processor

/** * A custom Handler must inherit the Netty specified HandlerAdapter * to be associated with the Netty framework. Is similar, the SpringMVC adapter pattern * * / public class MyServerHandler extends ChannelInboundHandlerAdapter {@ Override public void ChannelRead (ChannelHandlerContext CTX, Object MSG) throws Exception {// Obtain the message sent by the client. System.out.println(" received from client "+ ctx.channel().remoteAddress() +" sent message: "+ bytebuf.toString (charsetutil.utf_8)); } @override public void channelReadComplete(ChannelHandlerContext CTX) throws Exception {// Send a message to the client WriteAndFlush (Unpooled. CopiedBuffer (" Has the server received the message and sent you a question mark?" , CharsetUtil.UTF_8)); } @override public void exceptionCaught(ChannelHandlerContext CTX, Throwable Cause) throws Exception {// An Exception occurs, Close channel ctx.close(); }}Copy the code

4.4 Creating a Client Startup Class

public class MyClient { public static void main(String[] args) throws Exception { NioEventLoopGroup eventExecutors = new  NioEventLoopGroup(); Try {// Create the bootstrap object and configure the parameter bootstrap bootstrap = new bootstrap (); // Set the thread group bootstrap.group(eventExecutors) // Set the channel implementation type of the client. Channel (nioSocketChannel.class) // Initialize the channel.handler (new) using the anonymous inner class ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { Ch.pipeline ().addLast(new MyClientHandler()); }}); System.out.println(" Client is ready to fly ~"); ChannelFuture = bootstrap.connect("127.0.0.1", 6666).sync(); ChannelFuture = bootstrap.connect("127.0.0.1", 6666). Channelfuture.channel ().closeFuture().sync(); } the finally {/ / close the thread group eventExecutors. ShutdownGracefully (); }}}Copy the code

4.5 Creating a Client Processor

public class MyClientHandler extends ChannelInboundHandlerAdapter { @Override public void ChannelActive (ChannelHandlerContext CTX) throws Exception {// Sends a message to the server Ctx. writeAndFlush(Unpooled. CopiedBuffer (" Kibiba ~ jasmine ~Are you good~ Malaysia ~", charsetutil.utf_8)); } @Override public void channelRead(ChannelHandlerContext ctx, Object MSG) throws Exception {// Receive the message from the server ByteBuf ByteBuf = (ByteBuf) MSG; System.out.println(" received from server "+ ctx.channel().remoteAddress() +" : + bytebuf.toString (charsetutil.utf_8)); }}Copy the code

4.6 test

5. Netty features and key components

5.1 taskQueue taskQueue

public class MyServerHandler extends ChannelInboundHandlerAdapter { @Override public void ChannelRead (ChannelHandlerContext CTX, Object MSG) throws Exception { Ctx.channel ().eventloop ().execute(new Runnable() {@override public void run() {try { Thread.sleep(1000); thread.sleep (1000); System.out.println(" long service processing "); } catch (Exception e) { e.printStackTrace(); }}}); }}Copy the code

5.2 scheduleTaskQueue Indicates the queue of delayed tasks

Ctx.channel ().eventloop ().schedule(new Runnable() {@override public void run() {try { Thread.sleep(1000); thread.sleep (1000); System.out.println(" long service processing "); } catch (Exception e) { e.printStackTrace(); } } },5, TimeUnit.SECONDS); // Execute after 5 secondsCopy the code

5.3 Future Asynchrony

ChannelFuture = bootstrap.connect("127.0.0.1", 6666);Copy the code

/ / add listeners channelFuture. AddListener (new ChannelFutureListener () {/ / use an anonymous inner class, ChannelFutureListener interface // Override operationComplete method @Override public void operationComplete(ChannelFuture Future) throws Exception {// Check whether the operation succeeded if (future.issuccess ()) {system.out.println (" connection succeeded "); } else {system.out.println (" connection failed "); }}});Copy the code

5.4 the Bootstrap and ServerBootStrap

5.4.1 group ()

• BossGroup is used to listen for client connections and is responsible for creating connections with clients and registering them with the Selector of the workerGroup.
• WorkerGroup is used to handle read and write events for each connection.

EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
Copy the code

Private static final int DEFAULT_EVENT_LOOP_THREADS; The static {/ / NettyRuntime availableProcessors () * 2, Twice the number of CPU cores to constant DEFAULT_EVENT_LOOP_THREADS = math.h Max (1, SystemPropertyUtil. Get int (" io.net ty. EventLoopThreads ", NettyRuntime.availableProcessors() * 2)); if (logger.isDebugEnabled()) { logger.debug("-Dio.netty.eventLoopThreads: {}", DEFAULT_EVENT_LOOP_THREADS); } } protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... Args) {// If not passed, use the constant value, i.e., twice the number of CPU cores super(nThreads == 0? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args); }Copy the code

// Set the number of bossGroup threads to 1 EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(16);Copy the code

5.4.2 channel ()

These two channel types are commonly used because they are asynchronous and non-blocking. So it’s the first choice.

Slightly debugged locally, it works a little differently from Nio, blocking, so the API calls are different. Because it is blocking IO, few people will choose to use Oio, so it is hard to find examples. I thought about it a little, after several times of error, finally tune through. The code is as follows:

OioEventLoopGroup bossGroup = new OioEventLoopGroup(); ServerBootstrap bootstrap = new ServerBootstrap(); The bootstrap. Group (bossGroup) / / boosGroup. Only need to set up a thread group channel (OioServerSocketChannel. Class) / / set the server channel implementation type/code/client end, // If OioEventLoopGroup EventLoopGroup eventExecutors = new OioEventLoopGroup(); // If OioEventLoopGroup eventExecutors = new OioEventLoopGroup(); // Set the channel type to OioSocketChannel bootstrap.group(eventExecutors)// Set thread group. channel(oiosocketchannel. class)// Set the channel implementation type for the clientCopy the code

Local startup is not successful, online read some netizens comments, it is only in the Linux environment can be started. SCTP is not supported on this platform. Because my computer is Windows system, so the netizen said some truth.

5.4.3 option () and childOption ()

SO_RCVBUF Socket specifies the size of the TCP data receiving buffer. TCP_NODELAY TCP parameter: sends data immediately. The default value is true. The SO_KEEPALIVE Socket parameter is used to keep the connection alive. The default value is False. When this function is enabled, TCP proactively detects the validity of idle connections.

The SO_BACKLOG Socket parameter indicates the length of the queue for the server to accept connections. If the queue is full, the client connection will be rejected. The default value is 200 for Windows and 128 for others.

5.4.4 Setting the Pipeline (key)

Bootstrap.childhandler (new ChannelInitializer<SocketChannel>() {@override protected void InitChannel (SocketChannel SocketChannel) throws Exception {// Sets a custom handler for the pipeline pipeline socketChannel.pipeline().addlast (new) MyServerHandler()); }});Copy the code

ChannelInboundHandlerAdapter (inbound processor), ChannelOutboundHandler (outbound processor)

1. Register event fireChannelRegistered.
2. Connection establishment event fireChannelActive.
3. Read events and read completion events fireChannelRead and fireChannelReadComplete.
4. Abnormal notification event fireExceptionCaught.
5. User-defined event fireUserEventTriggered.
6. The Channel can be written fireChannelWritabilityChanged state change events.
7. Connection closing event fireChannelInactive.

1. Bind the port.
2. Connect to server Connect.
3. Write event Write.
4. Flush time Flush.
5. Read event read.
6. Disconnect actively.
7. Close the channel event close.

A similar handler() is used to assemble parent channels, also known as bossGroup threads. In most cases, you don’t use this method.

5.4.5 bind ()

5.4.6 Gracefully closing the EventLoopGroup

/ / release all resources, including the creation of the thread bossGroup. ShutdownGracefully (); workerGroup.shutdownGracefully();Copy the code

5.5 the Channel

A nexus to a network socket or a component which is capable of I/O operations such as read, write, connect, and bind

A channel provides a user: the current state of the channel (e.g. is it open? is it connected?) , the configuration parameters of the channel (e.g. receive buffer size), the I/O operations that the channel supports (e.g. read, write, connect, and bind), and the ChannelPipeline which handles all I/O events and requests associated with the channel.

1. The current state of the channel (e.g., is it open? Or connected?
2. Configuration parameters for a channel (such as the size of the receive buffer)
3. The IO operations supported by a channel (such as read, write, connect, and bind) and the Channel pipeline that handles all IO events and requests associated with a channel.

5.5.1 Obtaining the Channel status

boolean isOpen(); // If the channel is open, return true Boolean isRegistered(); // Return true Boolean isActive() if the channel is registered with EventLoop; // Return true Boolean isWritable() if the channel is active and connected; // Returns true if and only if the I/O thread will execute the requested write immediately.Copy the code

5.5.2 Obtaining Channel Configuration Parameters

ChannelConfig config = channel.config(); SO_BACKLOG parameter, Integer soBackLogConfig = config.getoption (channeloption.so_backlog); // soBackLogConfig=128 because my launcher is configured with 128Copy the code

ChannelConfig config = channel.config(); Map<ChannelOption<? >, Object> options = config.getOptions(); for (Map.Entry<ChannelOption<? >, Object> entry : options.entrySet()) { System.out.println(entry.getKey() + " : " + entry.getValue()); } /** SO_REUSEADDR : false WRITE_BUFFER_LOW_WATER_MARK : 32768 WRITE_BUFFER_WATER_MARK : WriteBufferWaterMark(low: 32768, high: 65536) SO_BACKLOG: 128 * /Copy the code

5.5.3 I/O operations supported by channel

@Override public void channelRead(ChannelHandlerContext ctx, Object MSG) throws Exception {ctx.channel().writeAndFlush(Unpooled. CopiedBuffer (" This wave, this wave is egg ~", charsetutil.utf_8)); }Copy the code

// Received server /127.0.0.1:6666 message: this wave ah, this wave is meat egg chicken ~Copy the code

ChannelFuture connect = channelFuture.channel().connect(new InetSocketAddress("127.0.0.1", 6666)); // Initiators are usually usedCopy the code

ChannelPipeline = ctx.channel().pipeline(); // Add the ChannelHandler handler to pipeline. AddLast (new MyServerHandler());Copy the code

5.6 the Selector

5.7 PiPeline and ChannelPipeline

5.8 ChannelHandlerContext

Protected DefaultChannelPipeline(Channel Channel) {this.channel = ObjectUtil.checkNotNull(channel, "channel"); succeededFuture = new SucceededChannelFuture(channel, null); voidPromise = new VoidChannelPromise(channel, true); Tail tail = new TailContext(this); head = new HeadContext(this); head.next = tail; tail.prev = head; }Copy the code

5.9 EventLoopGroup

5.9.1 Implementation principles of the polling mechanism

private final AtomicInteger idx = new AtomicInteger(); private final EventExecutor[] executors; @override public EventExecutor Next () {//idx.getAndIncrement() = idX ++, Return executors[IDx.getandIncrement () & Executors. Leng-1]; }Copy the code

@Override
public EventExecutor next() {
    return executors[Math.abs(idx.getAndIncrement() % executors.length)];
}
Copy the code