Super detailed introduction to Netty, just read this article!

Introduction: This article mainly describes some features and important components of the Netty framework. I hope that after reading it, I can have a more intuitive feeling about the Netty framework. I hope it can help readers get started with Netty quickly and reduce some detours.

 

mind Mapping


foreword


This article mainly describes some features and important components of the Netty framework. I hope that after reading it, I can have a more intuitive feeling about the Netty framework. I hope it can help readers get started with Netty quickly and reduce some detours.


1. Overview of Netty


Official introduction:
Netty is an asynchronous event-driven network application framework
for rapid development of maintainable high performance protocol servers & clients.
Netty is an asynchronous event-driven network application framework for rapid development of maintainable high-performance protocol servers and clients.


2. Why use Netty


From the official website, Netty is a network application framework for developing servers and clients. That is, a framework for network programming. Since it is network programming, Socket will not be discussed, why not use NIO?


2.1 Disadvantages of NIO
For this problem, I wrote an article before "Introduction to NIO" There is a more detailed introduction to NIO. The main problems of NIO are:

  • The class library and API of NIO are complicated, and the learning cost is high. You need to be proficient in Selector, ServerSocketChannel, SocketChannel, ByteBuffer, etc.
  • Familiarity with Java multithreaded programming is required. This is because NIO programming involves Reactor mode, you must be very familiar with multithreading and network programming in order to write high-quality NIO programs.
  • The infamous epoll bug. It causes the Selector to poll empty, which eventually leads to 100% CPU. Until the JDK1.7 version has not been fundamentally resolved.

2.2 Advantages of Netty
In contrast, Netty has many advantages:

  • The API is easy to use and the learning cost is low.
  • Powerful, built-in a variety of decoding encoders, support for a variety of protocols.
  • High performance. Compared with other mainstream NIO frameworks, Netty has the best performance.
  • The community is active, BUG will be repaired in time, the iterative version cycle is short, and new functions are constantly added.
  • Both Dubbo and Elasticsearch use Netty, and the quality has been verified.

 

3. Structure diagram


The picture above is the architecture picture on the homepage of the official website. Let's analyze it from top to bottom.
The green Core core module includes zero copy, API library, and extensible event model.


The orange part is Protocol Support protocol support, including Http protocol, webSocket, SSL (Secure Socket Protocol), Google Protobuf protocol, zlib/gzip compression and decompression, Large File Transfer large file transfer, etc.
The red part is Transport Services transmission services, including Socket, Datagram, Http Tunnel and so on.
It can be seen from the above that Netty's functions, protocols, and transmission methods are relatively complete and powerful.


4. Forever Hello Word


First build a HelloWord project, familiarize yourself with the API, and pave the way for later learning. Based on the picture below:


4.1 Introducing Maven dependencies


The version used is 4.1.20, a relatively stable version.
<dependency> <groupId>io.netty</groupId> <artifactId>netty-all</artifactId> <version>4.1.20.Final</version> </dependency>

4.2 Create a server startup class

public class MyServer {
    public static void main(String[] args) throws Exception {
        //Create two thread groups boosGroup, workerGroup
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            //Create the startup object of the server and set the parameters
            ServerBootstrap bootstrap = new ServerBootstrap();
            //Set up two thread groups boosGroup and workerGroup
            bootstrap.group(bossGroup, workerGroup)
                //Set the server channel implementation type    
                .channel(NioServerSocketChannel.class)
                //Set the thread queue to get the number of connections    
                .option(ChannelOption.SO_BACKLOG, 128)
                //Set up keep-alive connections    
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                //Initialize the channel object in the form of an anonymous inner class    
                .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            //Set the processor for the pipeline pipeline
                            socketChannel.pipeline().addLast(new MyServerHandler());
                        }
                    });//Set the processor for the pipeline corresponding to the EventLoop of the workerGroup
            System.out.println("java A server for tech lovers is ready...");
            //Bind the port number and start the server
            ChannelFuture channelFuture = bootstrap.bind(6666).sync();
            //Listen for closed channels
            channelFuture.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

4.3 Create a server-side processor

/**
 * The custom Handler needs to inherit the HandlerAdapter specified by Netty
 * It can be associated with the Netty framework, which is somewhat similar to the adapter mode of SpringMVC
 **/
public class MyServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //Get the message sent by the client
        ByteBuf byteBuf = (ByteBuf) msg;
        System.out.println("received client" + ctx.channel().remoteAddress() + "Message sent:" + byteBuf.toString(CharsetUtil.UTF_8));
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        //send message to client
        ctx.writeAndFlush(Unpooled.copiedBuffer("The server has received the message and sent you a question mark?", CharsetUtil.UTF_8));
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        //An exception occurred, closing the channel
        ctx.close();
    }
}

4.4 Create a client startup class

public class MyClient {

    public static void main(String[] args) throws Exception {
        NioEventLoopGroup eventExecutors = new NioEventLoopGroup();
        try {
            //Create a bootstrap object and configure parameters
            Bootstrap bootstrap = new Bootstrap();
            //set thread group
            bootstrap.group(eventExecutors)
                //Set the channel implementation type of the client    
                .channel(NioSocketChannel.class)
                //Initialize channel with anonymous inner class
                .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            //Add a handler for the client channel
                            ch.pipeline().addLast(new MyClientHandler());
                        }
                    });
            System.out.println("Client ready and ready to fly~");
            //Connect to the server
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6666).sync();
            //Listen for channel closure
            channelFuture.channel().closeFuture().sync();
        } finally {
            //close thread group
            eventExecutors.shutdownGracefully();
        }
    }
}

4.5 Create a client handler

public class MyClientHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        //send message to server
        ctx.writeAndFlush(Unpooled.copiedBuffer("Wabibab~jasmine~Are you good~Malaysia~", CharsetUtil.UTF_8));
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //Receive messages from the server
        ByteBuf byteBuf = (ByteBuf) msg;
        System.out.println("Received from the server" + ctx.channel().remoteAddress() + "message:" + byteBuf.toString(CharsetUtil.UTF_8));
    }
}

4.6 Testing

Start the server first, then start the client, and you can see the result:

MyServer print result:

 

 

MyClient prints the result:

 

5. Features and important components of Netty

5.1 taskQueue task queue

If the Handler processor has some long-term business processing, it can be handed over to taskQueue for asynchronous processing. How to use it, please see the code demo:

public class MyServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //Get the thread pool eventLoop, add threads, execute
        ctx.channel().eventLoop().execute(new Runnable() {
            @Override
            public void run() {
                try {
                    //Long-term operation, so that the long-term business operation will not cause the Handler to block
                    Thread.sleep(1000);
                    System.out.println("long business processing");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }
}

When we make a debug debug, we can see that the added taskQueue has a task.

 

 

5.2 scheduleTaskQueue delayed task queue

The delayed task queue is very similar to the task queue described above, except that there is an additional setting that can delay execution for a certain period of time. Please see the code demonstration:

ctx.channel().eventLoop().schedule(new Runnable() {
    @Override
    public void run() {
        try {
            //Long-term operation, so that the long-term business operation will not cause the Handler to block
            Thread.sleep(1000);
            System.out.println("long business processing");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
},5, TimeUnit.SECONDS);//Execute after 5 seconds

Still open debug to debug and view, we can have a scheduleTaskQueue task to be executed

 

 

5.3 Future asynchronous mechanism

When building the HelloWord project, we saw a line of code like this:

ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6666);

Many operations return this ChannelFuture object. What is this ChannelFuture object used for?

ChannelFuture provides a way to be notified asynchronously when an operation is complete. Generally, in Socket programming, waiting for the response result is blocked synchronously, but Netty will not cause blocking, because ChannelFuture obtains the result in a form similar to the observer mode. Please see a code demo:

//add listener
channelFuture.addListener(new ChannelFutureListener() {
    //Use anonymous inner class, ChannelFutureListener interface
    //Override the operationComplete method
    @Override
    public void operationComplete(ChannelFuture future) throws Exception {
        //Determine whether the operation is successful    
        if (future.isSuccess()) {
            System.out.println("connection succeeded");
        } else {
            System.out.println("Connection failed");
        }
    }
});

5.4 Bootstrap and Server Bootstrap

Bootstrap and ServerBootStrap are a factory class provided by Netty to create client and server starters. Using this factory class is very convenient to create startup classes. According to some examples above, it can be seen that it can greatly reduce the difficulty of development. . First look at a class diagram:

 

 

It can be seen that they are all inherited from the AbstractBootStrap abstract class, so the configuration methods are roughly the same.

In general, the steps to create a starter with Bootstrap can be divided into the following steps:

 

 

5.4.1 group()

in the previous article "Reactor mode" , we said that the server uses two thread groups:

  • bossGroup is used to monitor client connections, and is responsible for creating connections with clients and registering connections to the Selector of workerGroup.
  • workerGroup is used to handle read and write events for each connection.

Generally, create a thread group and use the following new directly to finish the job:

EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();

I am a little curious, since it is a thread group, what is the default number of threads? In-depth source code:

//save with a constant
    private static final int DEFAULT_EVENT_LOOP_THREADS;

    static {
        //NettyRuntime.availableProcessors() * 2, twice the number of cpu cores is assigned to a constant
        DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(
                "io.netty.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 in, the value of the constant is used, which is twice the number of cpu cores
        super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
    }

As you can see from the source code, the default number of threads is twice the number of cpu cores. Suppose you want to customize the number of threads, you can use a parameterized constructor:

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

5.4.2 channel()

This method is used to set the channel type. When the connection is established, the corresponding Channel instance will be created according to this setting.

You can see it in debug mode

 

 

The channel types are as follows:

NioSocketChannel: Asynchronous non-blocking client TCP Socket connection.

NioServerSocketChannel: Asynchronous non-blocking server-side TCP Socket connection.

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

OioSocketChannel: Synchronous blocking client TCP Socket connections.

OioServerSocketChannel: Synchronously blocking server-side TCP Socket connections.

I have debugged it locally, but it is different from Nio in use, because it is blocked, so the API calls are also different. Because it is blocking IO, few people will choose to use Oio, so it is difficult to find examples. I thought about it for a while, and after several error reports, I finally got it through. code show as below: 
//The server-side code is almost the same as above, only need to change three places
//This place uses OioEventLoopGroup
EventLoopGroup bossGroup = new OioEventLoopGroup();
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup)//Only need to set up a thread group boosGroup
        .channel(OioServerSocketChannel.class)//Set the server channel implementation type

//client-side code, only need to change two places
//Using OioEventLoopGroup
EventLoopGroup eventExecutors = new OioEventLoopGroup();
//The channel type is set to OioSocketChannel
bootstrap.group(eventExecutors)//set thread group
        .channel(OioSocketChannel.class)//Set the channel implementation type of the client

NioSctpChannel: Asynchronous client Sctp (Stream Control Transmission Protocol, Stream Control Transmission Protocol) connection.

NioSctpServerChannel: Asynchronous Sctp server connection.

It failed to start locally. I read some comments from netizens on the Internet, saying that it can only be started under the linux environment. From the error message: SCTP not supported on this platform, does not support this platform. Because my computer is a window system, what netizens said makes sense.

5.4.3 option() and childOption()

Let me first talk about the difference between the two.

option() sets the server to receive incoming connections, that is, the boosGroup thread.

childOption() is provided to the connection received by the parent pipeline, that is, the workerGroup thread.

After figuring it out, let's take a look at some commonly used settings:

SocketChannel parameters, which are commonly used parameters of childOption():

SO_RCVBUF Socket parameter, TCP data receiving buffer size.
TCP_NODELAY TCP parameter, send data immediately, the default value is True.
SO_KEEPALIVE Socket parameter, connection keep alive, the default value is False. When this function is enabled, TCP will actively detect the validity of idle connections.

ServerSocketChannel parameters, which are common parameters of option():

SO_BACKLOG Socket parameter, the queue length 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.

Due to space limitations, I won’t list others. You can go to the Internet to find information and have a look.

5.4.4 Setting up the pipeline (emphasis)

ChannelPipeline is the chain of responsibility for Netty to process requests, and ChannelHandler is the processor that specifically processes requests. In fact, each channel has a pipeline of processors.

In Bootstrap, the childHandler() method needs to initialize the channel and instantiate a ChannelInitializer. At this time, it is necessary to rewrite the method of initChannel() to initialize the channel. The assembly line is carried out in this place. The code demonstration is as follows:

//Initialize the channel object in the form of an anonymous inner class
bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        //Set a custom processor for the pipeline pipeline
        socketChannel.pipeline().addLast(new MyServerHandler());
    }
});

Processor Handler is mainly divided into two types:

ChannelInboundHandlerAdapter (inbound handler), ChannelOutboundHandler (outbound handler)

Inbound refers to the data from the underlying java NIO Channel to the Netty Channel.

Outbound refers to operating the underlying java NIO Channel through Netty's Channel.

Commonly used events for ChannelInboundHandlerAdapter processors are:

  1. Register event fireChannelRegistered.
  2. Connection establishment event fireChannelActive.
  3. Read event and read complete event fireChannelRead, fireChannelReadComplete.
  4. Exception notification event fireExceptionCaught.
  5. User-defined event fireUserEventTriggered.
  6. Channel writable state change event fireChannelWritabilityChanged.
  7. Connection closed event fireChannelInactive.

Commonly used events for ChannelOutboundHandler processors are:

  1. Port binding bind.
  2. Connect to the server connect.
  3. Write event write.
  4. Refresh time flush.
  5. Read event read.
  6. Actively disconnect disconnect.
  7. Close the channel event close.
There is also a similar handler(), which is mainly used to assemble the parent channel, which is the bossGroup thread. Under normal circumstances, this method is not used.

5.4.5 bind()

Provide the address and port number for the server or client to bind the server. The default is asynchronous startup. If the sync() method is added, it is synchronous.

There are five overloaded methods with the same name, all of which are used to bind the address port number. Not introduced one by one.

5.4.6 Close EventLoopGroup gracefully

//Release all resources, including created threads
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();

All child Channel s will be closed. After closing, the underlying resources are released.

5.5 Channel

What is a Channel? Take a look at the description of the official documentation:

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

Translation idea: A component that connects to a network socket or can perform I/O operations such as reading, writing, connecting, and binding.

If the above paragraph is more abstract, here is another paragraph:

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.

Translation to the effect:

The channel provides users with:

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

5.5.1 Get channel status

boolean isOpen(); //Returns true if the channel is open
boolean isRegistered();//Returns true if the channel is registered to the EventLoop
boolean isActive();//Returns true if the channel is active and connected
boolean isWritable();//Returns true if and only if the I/O thread will perform the requested write operation immediately.

The above are the methods to obtain the four states of the channel.

5.5.2 Get channel configuration parameters

To obtain a single piece of configuration information, use getOption(), code demonstration:

ChannelConfig config = channel.config();//Get configuration parameters
//Get the ChannelOption.SO_BACKLOG parameter,
Integer soBackLogConfig = config.getOption(ChannelOption.SO_BACKLOG);
//Because my launcher configuration is 128, so the soBackLogConfig=128 I got here

To get multiple pieces of configuration information, use getOptions(), code demonstration:

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
 Omitted below...
*/

5.5.3 IO operations supported by channel

Write operation, here demonstrates writing a message from the server to the client:

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    ctx.channel().writeAndFlush(Unpooled.copiedBuffer("This wave, this wave is meat, eggs, onion chicken~", CharsetUtil.UTF_8));
}

Client console:

//Received a message from the server/127.0.0.1:6666: This wave, this wave is meat, eggs, onion chicken~

Connection operation, code demonstration:

ChannelFuture connect = channelFuture.channel().connect(new InetSocketAddress("127.0.0.1", 6666));//Generally use the launcher, this method is not commonly used

Get ChannelPipeline through channel and do related processing:

//Get the ChannelPipeline object
ChannelPipeline pipeline = ctx.channel().pipeline();
//Add ChannelHandler processor to pipeline, assembly line
pipeline.addLast(new MyServerHandler());

5.6 Selector

In NioEventLoop, there is a member variable selector, which is the Selector of the nio package, before "Introduction to NIO" In, I have already talked about Selector.

The Selector in Netty is also the same as the Selector in NIO, which is used to listen to events, manage channel s registered in the Selector, and implement multiplexers.

 

 

5.7 PiPeline and ChannelPipeline

When introducing Channel earlier, we know that ChannelHandler pipeline processors can be assembled in the channel. It is impossible for a channel to have only one ChannelHandler processor. There must be many. Since many ChannelHandlers work in a pipeline, there must be an order.

So the pipeline appeared, and the pipeline is equivalent to the container of the processor. When initializing the channel, install the channelHandler in the pipeline in order, and then the channelHandler can be executed in order.

 

 

In a Channel, there is only one ChannelPipeline. The pipeline is created when the Channel is created. ChannelPipeline contains a list of ChannelHander s, and all ChannelHandler s will be registered to ChannelPipeline.

5.8 ChannelHandlerContext

In Netty, the Handler processor is defined by us, as mentioned above, it is realized by integrating the inbound processor or the outbound processor. At this time, if we want to get the pipeline object or channel object in the Handler, how to get it.

So Netty designed this ChannelHandlerContext context object, you can get channel s, pipeline s and other objects, and you can read and write operations.

 

 

Through the class diagram, ChannelHandlerContext is an interface, and there are three implementation classes below.

In fact, ChannelHandlerContext is in the form of a linked list in the pipeline. Look at a piece of source code to understand:

//ChannelPipeline implements the constructor method of class DefaultChannelPipeline
protected DefaultChannelPipeline(Channel channel) {
    this.channel = ObjectUtil.checkNotNull(channel, "channel");
    succeededFuture = new SucceededChannelFuture(channel, null);
    voidPromise =  new VoidChannelPromise(channel, true);
    //Set the head node head and the tail node tail
    tail = new TailContext(this);
    head = new HeadContext(this);
    
    head.next = tail;
    tail.prev = head;
}

Let me use a picture to show it, it will be more clear:

 

5.9 EventLoopGroup

Let's first look at the class diagram of EventLoopGroup:

 

 

It includes the commonly used implementation class NioEventLoopGroup. OioEventLoopGroup is also used in previous examples.

From the architecture diagram of Netty, we can know that the server needs two thread groups to work together, and the interface of this thread group is EventLoopGroup.

Each EventLoopGroup includes one or more EventLoops, and each EventLoop maintains a Selector instance.

5.9.1 Implementation Principle of Polling Mechanism

Let's take a look at the source code of DefaultEventExecutorChooserFactory:

private final AtomicInteger idx = new AtomicInteger();
private final EventExecutor[] executors;

@Override
public EventExecutor next() {
    //idx.getAndIncrement() is equivalent to idx++, and then modulo the task length
    return executors[idx.getAndIncrement() & executors.length - 1];
}

This code can determine that the execution method is the polling mechanism, and then debug it:

 

 

It also has a judgment here, if the number of threads is not 2 to the N th power, it will be implemented using a modulo algorithm.

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

 

Author: Omi No Ga

Tags: Netty

Posted by dearsina on Thu, 01 Dec 2022 20:21:03 +1030

Popular Tags

©2023 Programs Wiki