This article has participated in the “Digitalstar Project” and won a creative gift package to challenge the creative incentive money.

Previous: Your first Netty app

In the previous article, I wrote the first Netty entry application. In this article, I will further analyze the Netty workflow and core components from the above code combined with the flow chart of this article.

Finally, I give a further example to let you further understand.

Hope can let you gain something!! 🚀

Netty workflow

Let’s take a look at the Netty working diagram, briefly describe the workflow, and then use this diagram to analyze the core components of Netty.

1.1 workflow flow chart of Server

1.2 Analysis of Server workflow:

  1. When the server starts, it is bound to a local port, and NioServerSocketChannel is initialized.

  2. Register your NioServerSocketChannel with a selector of a BossNioEventLoopGroup.

    • The server contains oneBoss NioEventLoopGroupAnd 1Worker NioEventLoopGroup.
    • Boss NioEventLoopGroupIs responsible for receiving connections from clients,Worker NioEventLoopGroupSpecialized in network read and write
    • A NioEventLoopGroup is equivalent to a group of event loops containing multiple NioEventloops, each of which contains a selector and an event loop thread.

  3. BossNioEventLoopGroup Tasks performed by the loop:

    1. Poll accept event;

    2. Handle the Accept event and register the generated NioSocketChannel with a Selector of a WorkNioEventLoopGroup.

    3. Process tasks in the task queue, runAllTasks. The tasks in the task queue include tasks performed by users calling Eventloop. execute or Schedule, or tasks submitted to the Eventloop by other threads.

  4. WorkNioEventLoopGroup Tasks performed by the loop:

    • Poll for read and Write events

    • Handles IO events and calls back to the ChannelHandler when a NioSocketChannel readable or writable event occurs.

    • Tasks that process the task queue, namely runAllTasks

1.3. Work Flow chart of Client

The process will not be repeated overview 😁

2. Core module components

The core components of Netty are as follows:

  1. The Channel interface
  2. EventLoopGroup interface
  3. ChannelFuture interface
  4. ChannelHandler interface
  5. ChannelPipeline interface
  6. ChannelHandlerContext interface
  7. SimpleChannelInboundHandler abstract class
  8. The Bootstrap, ServerBootstrap class
  9. ChannelFuture interface
  10. ChannelOption class

2.1. Channel Interface

Basic I/O operations (bind(), connect(), read(), and write()) rely on primitives provided by the underlying network transport, which in Java is the Socket class.

The API provided by Netty’s Channel interface greatly reduces the complexity of using the Socket class directly. In addition, channels provide asynchronous network I/O operations (such as establishing connections, reading and writing, and binding ports). Asynchronous calls mean that any I/O calls are returned immediately, and there is no guarantee that the requested I/O operation has been completed at the end of the call.

Returns an instance of ChannelFuture immediately after the call, and registers listeners on ChannelFuture to allow callers to be notified immediately if an I/O operation succeeds, fails, or is cancelled.

In addition, channels are the root of an extensive class hierarchy with many predefined, specialized implementations, such as:

  • LocalServerChannel: ServerChannel for local transport, allowing VM communication.
  • EmbeddedChannel: Base class for Channel implementations used in embedded fashion.
  • NioSocketChannel: Indicates asynchronous client TCP and Socket connections.
  • NioServerSocketChannel: indicates asynchronous TCP and Socket connections on the server.
  • NioDatagramChannel: Asynchronous UDP connection.
  • NioSctpChannel: an asynchronous client Sctp connection that uses non-blocking mode and allows SctpMessage to be read/written to the underlying SctpChannel.
  • NioSctpServerChannel: asynchronous Sctp server connection. These channels cover UDP and TCP network IO and file IO.

2.2 EventLoopGroup Interface

EventLoop defines Netty’s core abstraction for handling events that occur during the lifetime of a connection.

Netty abstracts the Selector from the application by triggering events, eliminating all distribution code that would otherwise need to be written manually. Internally, an EventLoop will be assigned to each Channel to handle all events, including:

  • Registration event;
  • Send the event to ChannelHandler;
  • Schedule further action.

We won’t go into that here, but will give a brief explanation of the relationships among channels, EventLoop, Thread, and EventLoopGroup.

  • aEventLoopGroupContains one or moreEventLoop;
  • eachEventLoopTo maintain aSelectorInstance, so an EventLoop only combines one EventLoop during its lifetimeThread Binding;
  • So all of theEventLoopProcessing of I/O events will be proprietary in itThreadIs processed, virtually eliminating the need for synchronization;
  • aChannelOnly one is registered during its lifetimeEventLoop;
  • aEventLoopMay be assigned to one or moreChannel.
  • Usually one service port is oneServerSocketChannelCorresponds to aSelectorAnd aEventLoopThreads.BossEventLoopIs responsible for receiving the client’s connection and willSocketChanneltoWorkerEventLoopGroupIO processing, as shown in the flow chart above.

2.3 ChannelFuture Interface

All I/O operations in Netty are asynchronous. Because an operation may not return immediately, we need a way to determine its outcome at a later point in time. The implementation is through Future and ChannelFutures, where the addListener() method registers a ChannelFutureListener to automatically trigger a registered listening event when an operation completes (successfully or not).

Common methods are

  • Channel channel(), returns the current ongoingIOChannel of operation
  • ChannelFuture sync()Wait for the asynchronous operation to complete

2.4. ChannelHandler Interface

From the previous introductory program, you can see the importance of ChannelHandler in Netty, which acts as a container for all the application logic that handles inbound and outbound data. Most of our business logic is also written in the implemented word class, and the ChannelHandler methods are automatically triggered by events and do not need to be distributed by us.

ChannelHandler has many implementation classes or implementation subinterfaces. Usually we just inherit or subinterface and override the methods inside.

The most common types of handlers are:

  • ChannelInboundHandler : Access station events and data
  • ChannelOutboundHandler: used to handle outbound events and data.

Common adapters:

  • ChannelInboundHandlerAdapter: used for processing the inbound IO events

    The abstract base class of the ChannelInboundHandler implementation, which provides implementations of all methods. This implementation simply forwards the operation to the next ChannelPipeline handler. Subclasses can override method implementations to change this

  • ChannelOutboundHandlerAdapter: used for processing the outbound IO events

We often need to define a Handler class to inherit ChannelInboundHandlerAdapter, then rewrite the corresponding method to implement the business logic, let’s take a look at what method can be rewritten:

public class ChannelInboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelInboundHandler {
	// Register events
    public void channelRegistered(ChannelHandlerContext ctx) ;
	// 
    public void channelUnregistered(ChannelHandlerContext ctx);
	// The channel is ready
    public void channelActive(ChannelHandlerContext ctx);
	
    public void channelInactive(ChannelHandlerContext ctx) ;
    // Channel read data event
    public void channelRead(ChannelHandlerContext ctx, Object msg) ;
	// The channel read data event is complete
    public void channelReadComplete(ChannelHandlerContext ctx) ;

    public void userEventTriggered(ChannelHandlerContext ctx, Object evt);
	// Channel writability has changed
    public void channelWritabilityChanged(ChannelHandlerContext ctx);
	// Exception handling
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
}
Copy the code

2.5. ChannelPipeline interface

The ChannelPipeline provides a container for the ChannelHandler chain and defines an API for propagating a stream of inbound and outbound events on that chain. When a Channel is created, it is automatically assigned to its own ChannelPipeline. Their composition is as follows:

A Channel contains a ChannelPipeline, which maintains a two-way list of ChannelHandlerContext, And each ChanneHandlerContext is associated with a ChannelHandler.

ChannelHandler Installed into ChannelPipeline:

  1. aChannelInitializerThe implementation of theServerBootstrap;
  2. whenChannelInitializer.initChannel()When the method is called,ChannelInitializerWill be inChannelPipelineTo install a set of customChannelHandler;
  3. ChannelInitializerTake it from itselfChannelPipelineRemoved.

From the perspective of a client application, events are called outbound if they travel from client to server, and inbound if they travel from client to server. On the server, the reverse is true.

If a message or any other inbound event is read, it flows from the header of the ChannelPipeline and is passed to the first ChannelInboundHandler. After this handler completes processing, the data will be passed to the next ChannelInboundHandler in the chain. Eventually, the data will arrive at the end of the ChannelPipeline, where all processing ends.

Outbound events are passed forward from the tail to the last outbound handler. Outbound and inbound handlers do not interfere with each other.

2.6 ChannelHandlerContext interface

The purpose is to enable ChannelHandler to interact with its ChannelPipeline and other handlers. Because the ChannelHandlerContext holds all of the context information associated with a channel and is associated with a ChannelHandler object, ChannelHandlerContext can notify the next ChannelHandler of a ChannelPipeline and dynamically modify the ChannelPipeline to which it belongs.

2.7, SimpleChannelInboundHandler abstract class

It is common to see applications using a ChannelHandler to receive decoded messages and implement business logic in this Handler. To write a ChannelHandler, We only need to extend the abstract class SimpleChannelInboundHandler < T >, which is T type we have to deal with the news of the Java type.

The most important way is void in SimpleChannelInboundHandler channelRead0 (ChannelHandlerContext CTX, T MSG),

After we implement this method ourselves, the received message is already decoded.

Here’s an example:

2.8. Bootstrap and ServerBootstrap classes

Bootstrap means Bootstrap. A Netty application usually starts with a Bootstrap, which configures the entire Netty program and connects various components.

category Bootstrap ServerBootstrap
guide Used to boot the client Used to boot the server
Function in network programming Used to connect to remote hosts and ports Used to bind to a local port
The number of EventLoopGroup 1 2

I think you might be confused about that last point, why is one 1 and one 2?

Because the server requires two different sets of channels.

The first group will contain only a ServerChannel, representing the server’s own listening socket bound to a local port.

The second group will contain all channels that have been created to handle incoming client connections (one for each connection that has been accepted by the server).

This can be seen in the flow chart above.

2.9 ChannelFuture Interface

Result of an asynchronous Channel I/O operation. All I/O operations in Netty are asynchronous. This means that any I/O calls will return immediately, but there is no guarantee that the requested I/O operation has been completed at the end of the call. Instead, you return a ChannelFuture instance that gives you information about the result or status of an I/O operation. ChannelFuture is either unfinished or completed. When the I/O operation begins, a new future object is created. The new future is initially unfinished — it doesn’t succeed, fail, or cancel because the I/O operation hasn’t completed yet. If the I/O operation completes successfully, fails, or is canceled, the future is marked as completed with more specific information, such as the cause of the failure. Note that even failure and cancellation are complete states.

All I/O operations in Netty are asynchronous, so you cannot immediately know whether the message is correctly processed. However, you can either wait for it to complete or register a listener directly. This is done with Future and ChannelFutures. They can register a listener and the listener will automatically trigger the registered listener event when the operation succeeds or fails

Common methods are

  • Channel channel(), returns the current ongoingIOChannel of operation
  • ChannelFuture sync()Wait for the asynchronous operation to complete

2.10, ChannelOption class

  1. NettyWhen creating aChannelAfter the instance, you generally need to set itChannelOptionParameters.
  2. ChannelOptionThe parameters are as follows:
    • ChannelOption.SO_KEEPALIVE: Keeps the connection status
    • ChannelOption.SO_BACKLOG: Corresponds to the backlog parameter in THE TCP/IP LISTEN function, which is used to initialize the size of the server connectable queue. The backlog parameter specifies the size of the queue. Backilog specifies the backlog parameter. The backlog parameter specifies the queue size.

3. Application examples

【 Case 】 :

Write one server, two or more clients that can communicate with each other.

3.1. Server Handler

ChannelHandler has many implementation classes or implementation subinterfaces. Usually we just inherit or subinterface and override the methods inside.

Here we just inherited SimpleChannelInboundHandler < T >, which many methods are mostly as long as we rewrite the business logic, the trigger is invoked automatically in the event, mostly without we call manually.

package com.crush.atguigu.group_chat;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/ * * *@author crush
 */
public class GroupChatServerHandler extends SimpleChannelInboundHandler<String> {

    / * * * to define a channle group, manage all channel * GlobalEventExecutor INSTANCE) is a global event actuator, is a singleton * /
    private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");


    /** * handlerAdded Adds the current channel to the channelGroup@param ctx
     * @throws Exception
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        // Push the chat information of this client to other online clients
        /* This method iterates through all channels in the channelGroup and sends messages. We don't need to iterate */ ourselves
        channelGroup.writeAndFlush("[client]" + channel.remoteAddress() + "Join the chat" + sdf.format(new java.util.Date()) + " \n");
        channelGroup.add(channel);

    }

    /** * Disconnect and push xx customer departure information to the current online customer *@param ctx
     * @throws Exception
     */
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {

        Channel channel = ctx.channel();
        channelGroup.writeAndFlush("[client]" + channel.remoteAddress() + "Left \n");
        System.out.println("channelGroup size" + channelGroup.size());

    }

    /** * indicates that the channel is active@param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {

        System.out.println(ctx.channel().remoteAddress() + "Online ~");
    }

    /** * indicates that the channel is not active. If the channel is dead, xx is offline@param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {

        System.out.println(ctx.channel().remoteAddress() + "Offline ~");
    }

    // Read data
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {

        // Get the current channel
        Channel channel = ctx.channel();
        // We iterate through channelGroup and send back different messages depending on the situation

        channelGroup.forEach(ch -> {
            if(channel ! = ch) {// Not the current channel, forward the message
                ch.writeAndFlush("[customer]" + channel.remoteAddress() + "Message sent" + msg + "\n");
            } else {// Echo the message you sent to yourself
                ch.writeAndFlush("[himself] sent the message." + msg + "\n"); }}); }@Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // Close the channelctx.close(); }}Copy the code

3.2. Start the Server

package com.crush.atguigu.group_chat;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

/ * * *@author crush
 */
public class GroupChatServer {

    /** */ / listen on port */
    private int port;

    public GroupChatServer(int port) {
        this.port = port;
    }

    /** * Write the run method to handle the request *@throws Exception
     */
    public void run(a) throws Exception {

        // Create two thread groups
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        Eight NioEventLoop / /
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();

            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            // Get pipeline
                            ChannelPipeline pipeline = ch.pipeline();
                            // Add the decoder to the pipeline
                            pipeline.addLast("decoder".new StringDecoder());
                            // Add encoder to pipeline
                            pipeline.addLast("encoder".new StringEncoder());
                            // Add your own business handler
                            pipeline.addLast(newGroupChatServerHandler()); }}); System.out.println("Netty Server startup");

            ChannelFuture channelFuture = b.bind(port).sync();

            // The listener is closed
            channelFuture.channel().closeFuture().sync();
        } finally{ bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); }}public static void main(String[] args) throws Exception {
        new GroupChatServer(7000).run(); }}Copy the code

3.3. Client Handler

package com.crush.atguigu.group_chat;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
/ * * *@author crush
 */
public class GroupChatClientHandler extends SimpleChannelInboundHandler<String> {
    
    // Called when the current Channel has read a message from the other Channel.
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { System.out.println(msg.trim()); }}Copy the code

3.4. Client Server

package com.crush.atguigu.group_chat;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.util.Scanner;


/ * * *@author crush
 */
public class GroupChatClient {
    
    private final String host;
    private final int port;

    public GroupChatClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public void run(a) throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();

        try {
            
            Bootstrap bootstrap = new Bootstrap()
                    .group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {

                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {

                            / / get the pipeline
                            ChannelPipeline pipeline = ch.pipeline();
                            // Add the relevant handler
                            pipeline.addLast("decoder".new StringDecoder());
                            pipeline.addLast("encoder".new StringEncoder());
                            // Add a custom handler
                            pipeline.addLast(newGroupChatClientHandler()); }}); ChannelFuture channelFuture = bootstrap.connect(host, port).sync();/ / get the channel
            Channel channel = channelFuture.channel();
            System.out.println("-- -- -- -- -- -- --" + channel.localAddress() + "-- -- -- -- -- -- -- --");
            // The client needs to enter information
            Scanner scanner = new Scanner(System.in);
            while (scanner.hasNextLine()) {
                String msg = scanner.nextLine();
                // Send to the server through channel
                channel.writeAndFlush(msg + "\r\n"); }}finally{ group.shutdownGracefully(); }}public static void main(String[] args) throws Exception {
        new GroupChatClient("127.0.0.1".7000).run(); }}Copy the code

Multiple clients, CV can be.

3.5. Test:

The test process is to start the Server and then the client.

4. Talk to yourself

This article should be regarded as a save draft, before busy other things to 😂.

That’s all for today’s article.

Hello, THIS is blogger Ning Zaichun: homepage

If you encounter doubts in the article, please leave a message or private letter, or add the homepage contact information, will reply as soon as possible.

If you find any problems in the article, please correct them. Thank you very much.

If you think it will help you, please click “like” before leaving!

Welcome everyone to discuss in the discussion forum, increase your luck, rub a rub against the nuggets official sent around oh!!

The more you comment, the more likely you are to win an award!!

Details 👉 Diggin project