1 Data Channels: Channel

Overview: In Netty, all data flows through channels, which can be considered as "data transmission pathways"

Common Channel Methods

• close() can be used to close the channel
• closeFuture() handles channel closure
• sync method synchronously waits for channel closure
• addListener method asynchronously waits for channel closure
• pipeline() adds handlers
• write() writes data
• writeAndFlush() writes and flushes data

1-1 Channel and ChannelFuture Relationship

package netty_components.channel;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;

import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;

/*Netty Channel and ChannelFuture Relationship Demonstration*/
public class ConnectionClient {
    public static void main(String[] args) throws InterruptedException {
        ChannelFuture connectionFuture = new Bootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel socketChannel) throws Exception {
                        // Add client channel handler for string encoding
                        socketChannel.pipeline().addLast(new StringEncoder(StandardCharsets.UTF_8));
                    }
                })
                .connect(new InetSocketAddress("localhost", 8080));
        
        // connect is asynchronous and non-blocking
        connectionFuture.sync();

        /*
             Without sync, the channel converted from channelFuture would be invalid
             because the connection to the remote host hasn't been established yet
        */
        Channel activeChannel = connectionFuture.channel();    // Convert ChannelFuture to Channel
        activeChannel.writeAndFlush("hello world");
    }
}

• The code shows that the client's ChannelFuture is an intermediate state class, and the valid Channel object must have established a connection with the remote host.

Note: Classes with Future and Promise typically work with asynchronous method calls

Example: channelFuture uses the asynchronous non-blocking connect method to establish a connection with the remote server, but must ensure the connection is established before becoming a valid channel. This is essentially synchronization of the network connection state.

1-2 ChannelFuture Result Handling

package netty_components.channel;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;

import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;

/*Netty Channel and ChannelFuture Relationship Demonstration*/
public class ConnectionClient {
    public static void main(String[] args) throws InterruptedException {
        ChannelFuture connectionFuture = new Bootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(new StringEncoder(StandardCharsets.UTF_8));
                    }
                })
                .connect(new InetSocketAddress("localhost", 8080));
        
        // connect is asynchronous and non-blocking
        /*
           1) Synchronous result handling with sync (blocks calling thread until result returns)
           sync is a blocking method that only proceeds after connection establishment
        */
//        connectionFuture.sync();
//        Channel channel = connectionFuture.channel();
//        channel.writeAndFlush("Synchronous call, waiting for channel establishment");

        /*
           2) Asynchronous result handling with addListener
           The calling thread doesn't need to wait for connection establishment
           After calling, it provides a callback object as a parameter
           When the connection is established, the NIO thread calls the callback
         */
        connectionFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                Channel channel = future.channel();
                channel.writeAndFlush("Asynchronous call, callback for channel establishment");
            }
        });
    }
}

Connection Establishment Summary:

The client thread initiates a request to the remote server and needs to establish a connection. In real scenarios, the completion time of connection establishment is uncertain.

Therefore, the main thread uses the asynchronous connect method to request connection establishment. After calling connect, another NIO thread performs the connection establishment work.

The main thread can handle the NIO thread's results in two ways:

• Synchronous call: The main thread calls the sync method to block and wait for the NIO thread's result. When results are returned, it continues execution. The channelFuture to channel conversion is done by the main thread
• Asynchronous call: The main thread provides a callback function to the NIO thread. After the NIO thread establishes the connection, it calls the callback to handle the result, completing the channelFuture to channel conversion

1-3 ChannelFuture Connection Closure

Requirement: When a certain condition is met, the client needs to close the remote server's channel and perform some operations after closure. Since the close operation is asynchronous, we need to ensure some code definitely runs after the connection is closed

-netty satisfies this requirement through the channelFuture object

Usage example

package netty_components.channel;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;
@Slf4j
public class CloseConnectionClient {
    public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup group = new NioEventLoopGroup();
        ChannelFuture connectionFuture = new Bootstrap()
                .group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                        socketChannel.pipeline().addLast(new StringEncoder(StandardCharsets.UTF_8));
                    }
                })
                .connect(new InetSocketAddress("localhost", 8080));
        connectionFuture.sync();

        // Thread closes channel
        Channel activeChannel = connectionFuture.channel();
        Thread t = new Thread(new ChannelCloser(activeChannel, group), "input");
        t.start();

    }
}

@Slf4j
class ChannelCloser implements Runnable{
    private Channel activeChannel;
    private NioEventLoopGroup eventGroup;
    
    ChannelCloser(Channel channel, NioEventLoopGroup group){
        this.activeChannel = channel;
        this.eventGroup = group;
    }

    @SneakyThrows
    @Override
    public void run() {
        Scanner sc = new Scanner(System.in);
        while(true) {
            String line = sc.nextLine();
            if (line.equals("q")) {
                activeChannel.close();
                break;
            }
            activeChannel.writeAndFlush(line);
        }

         /*
           closeFuture handles post-connection closure operations, also with two methods:
           1) Synchronous wait for connection closure, then execute handling code
           2) Asynchronous call, where another thread completes channel closure and executes handling code
        */
        ChannelFuture closeFuture = activeChannel.closeFuture();
        // Method 1: Synchronous handling
        //        closeFuture.sync();
        //        log.debug("Code after channel closure");


        // Method 2: Asynchronous handling
        closeFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                log.debug("Code after channel closure");
                // Close channel and event loop group to terminate the application
                eventGroup.shutdownGracefully();
            }
        });
    }
}

• The above code starts a thread to monitor console input and closes the channel connection when "q" is entered

To ensure the channel is closed, similar to channel establishment, we can use the channelFuture object to detect the closure result, which also provides synchronous and asynchronous methods:

Synchronous wait for closure: closeFuture.sync();
Asynchronous closure call: 
closeFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                log.debug("Code after channel closure");
                // Close channel and event loop group to terminate the application
                eventGroup.shutdownGracefully();
            }
});

Significance of Asynchronous Channel Establishment and Closure==========

Netty uses asynchronous operations to increase the number of connections processed per unit time. The entire connection lifecycle between client and server includes connection establishment, data transmission, and connection closure. Asynchronous operations allow each thread to handle only one stage. With multi-core CPUs supporting concurrent thread execution, this forms a pipeline that improves connection throughput.

2 Inter-thread Result Passing: Future and Promise

Interface Name	Purpose
JUC Future	Synchronously **wait for task completion** (success or failure) to get execution results
Netty Future	Extends JUC Future, can use asynchronous methods to **wait for task completion** and get results
Netty Promise	Further enhances Netty Future, can be defined separately as a container for passing results between threads

Note: Future interfaces typically work with thread pools, and Netty's thread pools usually refer to eventLoop, which is a single-thread thread pool.

2-1 JUC Future Interface Implementation

public interface Future<V> {
    boolean cancel(boolean mayInterruptIfRunning);
    boolean isCancelled();
    boolean isDone();
    V get() throws InterruptedException, ExecutionException;
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

package netty_components.future;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.*;
@Slf4j
public class JucFutureDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService service = Executors.newFixedThreadPool(2);
        Future<Integer> future  = service.submit(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                log.debug("Performing calculation");
                Thread.sleep(1000);
                return 50;
            }
        });

        log.debug("Waiting for thread pool execution result");
        log.debug("Result {}", future.get());
    }
}

Execution Result

09:12:39 [DEBUG] [main] n.c.JucFutureDemo - Waiting for thread pool execution result
09:12:39 [DEBUG] [pool-1-thread-1] n.c.JucFutureDemo - Performing calculation
09:12:40 [DEBUG] [main] n.c.JucFutureDemo - Result 50

2-2 Netty Future Interface Implementation

public interface Future<V> extends java.util.concurrent.Future<V> {
    boolean isSuccess();
    boolean isCancellable();
    Throwable cause();
    Future<V> addListener(GenericFutureListener<? extends Future<? super V>> listener);
    Future<V> addListeners(GenericFutureListener<? extends Future<? super V>>... listeners);
    Future<V> removeListener(GenericFutureListener<? extends Future<? super V>> listener);
    Future<V> removeListeners(GenericFutureListener<? extends Future<? super V>>... listeners);
    Future<V> sync() throws InterruptedException;
    Future<V> syncUninterruptibly();
    Future<V> await() throws InterruptedException;
    Future<V> awaitUninterruptibly();
    boolean await(long timeout, TimeUnit unit) throws InterruptedException;
    boolean await(long timeoutMillis) throws InterruptedException;
    boolean awaitUninterruptibly(long timeout, TimeUnit unit);
    boolean awaitUninterruptibly(long timeoutMillis);
    V getNow();
    @Override
    boolean cancel(boolean mayInterruptIfRunning);
}

package netty_components.future;
import io.netty.channel.EventLoop;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;

@Slf4j
public class NettyFutureDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        NioEventLoopGroup group = new NioEventLoopGroup();
        EventLoop eventLoop = group.next();
        Future<Integer> future = eventLoop.submit(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                log.debug("Thread calculation result");
                return 5;
            }
        });
        
        // Synchronous get blocks main thread
        /*
         log.debug("Waiting for thread pool execution result");
         log.debug("Result {}", future.get());
        */
        // Asynchronous execution through callback
        future.addListener(new GenericFutureListener<Future<? super Integer>>() {
            @Override
            public void operationComplete(Future<? super Integer> future) throws Exception {
                log.debug("Waiting for thread pool execution result");
                log.debug("Result {}", future.getNow());
            }
        });
    }
}

Execution Result

09:14:52 [DEBUG] [nioEventLoopGroup-2-1] n.c.NettyFutureDemo - Thread calculation result
09:14:52 [DEBUG] [nioEventLoopGroup-2-1] n.c.NettyFutureDemo - Waiting for thread pool execution result
09:14:52 [DEBUG] [nioEventLoopGroup-2-1] n.c.NettyFutureDemo - Result 5

2-3 Promise Interface Implementation

public interface Promise<V> extends Future<V> {
    Promise<V> setSuccess(V result);        // Set success and notify all listeners
    boolean trySuccess(V result);           // Mark success and notify all listeners
    Promise<V> setFailure(Throwable cause); // Set failure and notify all listeners
    boolean tryFailure(Throwable cause);    // Mark failure and notify all listeners
    boolean setUncancellable();
    @Override
    Promise<V> addListener(GenericFutureListener<? extends Future<? super V>> listener);
    @Override
    Promise<V> addListeners(GenericFutureListener<? extends Future<? super V>>... listeners);
    @Override
    Promise<V> removeListener(GenericFutureListener<? extends Future<? super V>> listener);
    @Override
    Promise<V> removeListeners(GenericFutureListener<? extends Future<? super V>>... listeners);
    @Override
    Promise<V> await() throws InterruptedException;
    @Override
    Promise<V> awaitUninterruptibly();
    @Override
    Promise<V> sync() throws InterruptedException;
    @Override
    Promise<V> syncUninterruptibly();
}

package netty_components.future;

import io.netty.channel.DefaultEventLoop;
import io.netty.channel.EventLoop;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.Promise;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ExecutionException;

@Slf4j
public class NettyPromiseDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        EventLoop eventLoop = new DefaultEventLoop();
        Promise<Integer> promise = new DefaultPromise<>(eventLoop);
        new Thread(()->{
            log.debug("Starting calculation");
            try {
                Thread.sleep(1000);
                promise.setSuccess(100);
            } catch (InterruptedException e) {
                promise.setFailure(e);
                e.printStackTrace();
            }
        }).start();
        log.debug("Waiting for execution result");
        log.debug("Execution result {}", promise.get());
    }
}

Execution Result

09:25:59 [DEBUG] [main] n.c.NettyPromiseDemo - Waiting for execution result
09:25:59 [DEBUG] [Thread-0] n.c.NettyPromiseDemo - Starting calculation
09:26:00 [DEBUG] [main] n.c.NettyPromiseDemo - Execution result 100

Future and Promise Interface Summary============

Both interfaces are used for inter-process result passing. Netty's Future interface adds asynchronous callback methods compared to JUC's Future. However, Future essentially still depends on the thread's return value from submit. Promise can be defined separately, implementing the Future interface while being more flexible.

3 Network Data Processing: Handler & Pipeline

3-1 Inbound vs Outbound ChannelHandler

Handler Purpose: In actual network programming, business logic is typically implemented in handlers. Both inbound and outbound information processing occurs in handlers, and multiple handlers can form a pipeline.

Netty Provided Handlers

Name	Purpose
ChannelHandler	Handles events on the channel, divided into inbound and outbound handlers

• Inbound handlers are typically subclasses of ChannelInboundHandlerAdapter, mainly used to read client data and write back results
• Outbound handlers are typically subclasses of ChannelOutboundHandlerAdapter, mainly used to process write-back results

Handler Usage Example:

package netty_components.handler;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;

/*
    Difference between ChannelInboundHandlerAdapter and ChannelOutboundHandlerAdapter
    When external data arrives, inbound handlers are triggered, processing from the head of the doubly-linked list (same order as handler addition)
    When external data is sent out, outbound handlers are triggered, processing from the tail of the doubly-linked list (reverse order of handler addition)

    Difference: Different triggering times, inbound from pipeline head, outbound from pipeline tail
 */
@Slf4j
public class PipelineDemo {
    public static void main(String[] args) {
        new ServerBootstrap()
                .group(new NioEventLoopGroup(), new NioEventLoopGroup(2))
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast("h1", new ChannelInboundHandlerAdapter(){
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                log.debug("1");
                                // Pass msg to next handler, equivalent to ctx.fireChannelRead
                                ByteBuf buf = (ByteBuf) msg;
                                String s = "h1" + buf.toString(StandardCharsets.UTF_8);
                                super.channelRead(ctx, s);
                            }
                        });
                        pipeline.addLast("h2", new ChannelInboundHandlerAdapter(){
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                log.debug("2");
                                String current = "h2" + msg;
                                log.debug(current);
                                ch.writeAndFlush(ctx.alloc().buffer().writeBytes("servers..".getBytes()));
                            }
                        });

                        pipeline.addLast("h3", new ChannelOutboundHandlerAdapter(){

                            @Override
                            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                                log.debug("3 " + msg);
                                super.write(ctx, msg, promise);
                            }
                        });

                        pipeline.addLast("h4", new ChannelOutboundHandlerAdapter(){
                            @Override
                            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                                log.debug("4 " + msg);
                                super.write(ctx, msg, promise);
                            }
                        });
                    }
                })
                .bind(8080);
    }
}

The above code adds 2 inbound and 2 outbound handlers in sequence. The second inbound handler writes information to the channel, triggering outbound handlers. After sending two messages from the client (client1:hello1 and client:hello), the program output is:

• Inbound messages are processed from head handlers (1,2), outbound messages from tail handlers (4,3)

16:52:21 [DEBUG] [nioEventLoopGroup-3-1] n.h.PipelineDemo - 1
16:52:21 [DEBUG] [nioEventLoopGroup-3-1] n.h.PipelineDemo - 2
16:52:21 [DEBUG] [nioEventLoopGroup-3-1] n.h.PipelineDemo - h2h1client1:hello1
16:52:21 [DEBUG] [nioEventLoopGroup-3-1] n.h.PipelineDemo - 4 PooledUnsafeDirectByteBuf(ridx: 0, widx: 9, cap: 256)
16:52:21 [DEBUG] [nioEventLoopGroup-3-1] n.h.PipelineDemo - 3 PooledUnsafeDirectByteBuf(ridx: 0, widx: 9, cap: 256)
16:52:25 [DEBUG] [nioEventLoopGroup-3-1] n.h.PipelineDemo - 1
16:52:25 [DEBUG] [nioEventLoopGroup-3-1] n.h.PipelineDemo - 2
16:52:25 [DEBUG] [nioEventLoopGroup-3-1] n.h.PipelineDemo - h2h1client2:hello
16:52:25 [DEBUG] [nioEventLoopGroup-3-1] n.h.PipelineDemo - 4 PooledUnsafeDirectByteBuf(ridx: 0, widx: 9, cap: 256)
16:52:25 [DEBUG] [nioEventLoopGroup-3-1] n.h.PipelineDemo - 3 PooledUnsafeDirectByteBuf(ridx: 0, widx: 9, cap: 256)

ChannelHandler and Pipeline Summary=

Each server channel can be seen as a doubly-linked list. Inbound information processing calls inbound handlers forward, outbound information processing calls outbound handlers backward. Channel design concept: Implements the Chain of Responsibility Pattern, creating a processing object chain (pipeline) for requests. Each node in the chain is an object (handler) that processes different requests and internally maintains the next node object. When a request is sent from the head of the chain, it passes through each node in sequence until a node processes it.

Note: Different calling objects affect outbound handler usage

ctx.channel().write(msg) // Start from tail to find outbound handler (entire pipeline tail)
ctx.write(msg)           // ctx:ChannelHandlerContext
// Find previous outbound handler from current node (handler before current in pipeline)

• Although both methods write data to the channel, the range of outbound handler search is different, which needs attention in actual development.

3-2 Test Channel

package netty_components.handler;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.channel.ChannelPromise;
import io.netty.channel.embedded.EmbeddedChannel;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;

/*
*  Requirement: When network communication involves complex business requirements (multiple handler implementations), testing requires running both server and client programs, which is cumbersome.
*   EmbeddedChannel avoids this problem.
* */
@Slf4j
public class TestEmbeddedChannel {
    public static void main(String[] args) {
        ChannelInboundHandlerAdapter h1 = new ChannelInboundHandlerAdapter() {
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                log.debug("1");
                super.channelRead(ctx, msg);
            }
        };
        ChannelInboundHandlerAdapter h2 = new ChannelInboundHandlerAdapter() {
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                log.debug("2");
                super.channelRead(ctx, msg);
            }
        };

        ChannelOutboundHandlerAdapter h3 = new ChannelOutboundHandlerAdapter(){
            @Override
            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                log.debug("3");
                super.write(ctx, msg, promise);
            }
        };

        ChannelOutboundHandlerAdapter h4 = new ChannelOutboundHandlerAdapter(){
            @Override
            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                log.debug("4");
                super.write(ctx, msg, promise);
            }
        };

        EmbeddedChannel channel = new EmbeddedChannel(h1, h2, h3, h4);
        // Simulate inbound operation
        channel.writeInbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("INPUT".getBytes()));
        // Simulate outbound operation
        channel.writeOutbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("OUT".getBytes()));
    }
}

Execution Result

22:13:04 [DEBUG] [main] n.h.TestEmbeddedChannel - 1
22:13:04 [DEBUG] [main] n.h.TestEmbeddedChannel - 2
22:13:04 [DEBUG] [main] n.h.TestEmbeddedChannel - 4
22:13:04 [DEBUG] [main] n.h.TestEmbeddedChannel - 3

Summary: Netty's EmbeddedChannel allows convenient testing of complex pipelines without needing to start client and server programs for each test.

4 Data Buffering: ByteBuf

4-1 ByteBuf Dynamic Expansion

Overview: Java's native NIO API uses ByteBuffer for data buffering, while Netty uses ByteBuf, which can dynamically expand based on input data size.

ByteBuf Advantages

• Pooling - Can reuse ByteBuf instances from the pool, saving memory and reducing memory overflow possibiliteis
• Separate read/write pointers, no need to switch read/write modes like ByteBuffer
• Automatic expansion capability
• Supports chained calls for smoother usage
• Zero-copy in many places, such as slice, duplicate, CompositeByteBuf

ByteBuffer usage reference

package netty_components.buffer;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;

import static io.netty.buffer.ByteBufUtil.appendPrettyHexDump;
import static io.netty.util.internal.StringUtil.NEWLINE;

public class TestBytebuf {
    public static void main(String[] args) {
        ByteBuf buf = ByteBufAllocator.DEFAULT.buffer();
        log(buf);
        StringBuilder sb = new StringBuilder();
        for(int i = 0; i < 300; ++i) sb.append("a");
        buf.writeBytes(sb.toString().getBytes());
        log(buf);
    }

    private static void log(ByteBuf buffer) {
        int length = buffer.readableBytes();
        int rows = length / 16 + (length % 15 == 0 ? 0 : 1) + 4;
        StringBuilder buf = new StringBuilder(rows * 80 * 2)
                .append("read index:").append(buffer.readerIndex())
                .append(" write index:").append(buffer.writerIndex())
                .append(" capacity:").append(buffer.capacity())
                .append(NEWLINE);
        appendPrettyHexDump(buf, buffer);
        System.out.println(buf.toString());
    }
}

Execution Result

read index:0 write index:0 capacity:256        // Read pointer position, write pointer position, capacity

read index:0 write index:300 capacity:512
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000010| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000020| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000030| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000040| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000050| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000060| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000070| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000080| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000090| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|000000a0| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|000000b0| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|000000c0| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|000000d0| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|000000e0| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|000000f0| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000100| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000110| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000120| 61 61 61 61 61 61 61 61 61 61 61 61             |aaaaaaaaaaaa    |
+--------+-------------------------------------------------+----------------+

ByteBuf capacity expanded from 256 to 512.

4-2 ByteBuf Memory Types and Pooling

Features: ByteBuf supports heap and direct memory and memory pooling.

Direct Memory:

• High allocation and deallocation cost, but high read/write efficiency. NIO buffers typically use direct memory by default.
• Not affected by garbage collection

JVM Direct Memory

Pooling Mechanism: Pooling is essentially a resource pool that reuses resources instead of acquiring and releasing them on demand.

Pooling Concept Example: Thread pools, memory pools, connection pools, object pools

a pool is a collection of resources that are kept ready to use, rather than acquired on use and released

Enable Pooling: Reuse ByteBuf instances from the pool, using memory allocation algorithms similar to jemalloc to improve allocation efficiency

Disable Pooling: Each ByteBuf instance needs to be created, with significant overhead from repeated creation and destruction. If heap memory is used, it creates significant GC pressure.

• Whether pooling is enabled depends on the Netty version and application platform.

Enable/Disable Pooling via JVM Parameters

-Dio.netty.allocator.type={unpooled|pooled}

package netty_components.buffer;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;

public class TestBytebuf {
    public static void main(String[] args) {
        ByteBuf buf = ByteBufAllocator.DEFAULT.buffer();
        System.out.println(buf.getClass());
        StringBuilder sb = new StringBuilder();
        for(int i = 0; i < 300; ++i) sb.append("a");
        buf.writeBytes(sb.toString().getBytes());
    }
}

Output:

class io.netty.buffer.PooledUnsafeDirectByteBuf

Outputting PooledUnsafeDirectByteBuf indicates pooling is enabled and direct memory is used.

4-3 ByteBuf Operations

Features

Two Capacities: ByteBuf requires specifying capacity on creation, with maximum capacity being the system's maximum allocatable size (theoretical maximum). Between maximum and specified capacity is the expandable capacity.

Two Pointers: Read and write pointers, both at the starting position when buf is initialized.

Space Allocation: Discarded bytes (written and read), readable bytes (written but unread), writable bytes, expandable bytes

nio.bytebuffer implementation principle

nio.bytebuffer and netty.bytebuffer Comparison

Class	nio.bytebuffer	netty.buffer.ByteBuf
Capacity	Fixed capacity, cannot be changed after definition	Set initial capacity, can dynamically expand, supports pooling
Implementation Principle	Shared read/write pointers, need to switch read/write modes	Separate read/write pointers

Writing

Features:

• int type data writing supports big-endian (write high bits first) and little-endian (write low bits first), big-endian is typically used in network programming
• Supports writting StringBuffer, StringBuilder, and nio.bytebuffer.

Method Signature	Meaning	Note
writeBoolean(boolean value)	Write boolean value	01|00 represents true|false
writeByte(int value)	Write byte value
writeShort(int value)	Write short value
writeInt(int value)	Write int value	Big Endian, 0x250 writes as 00 00 02 50
writeIntLE(int value)	Write int value	Little Endian, 0x250 writes as 50 02 00 00
writeLong(long value)	Write long value
writeChar(int value)	Write char value
writeFloat(float value)	Write float value
writeDouble(double value)	Write double value
writeBytes(ByteBuf src)	Write netty ByteBuf
writeBytes(byte[] src)	Write byte[]
writeBytes(ByteBuffer src)	Write nio ByteBuffer
int writeCharSequence(CharSequence sequence, Charset charset)	Write string

Reading

buffer.readByte();                         // Read one byte from buffer, read pointer moves
// Re-read certain data
buffer.markReaderIndex();                   // Mark
buffer.readInt();                          // Read
buffer.resetReaderIndex();                  // Return to mark (can continue reading from mark)

Memory Management

ByteBuf Implementation	Recovery Method
UnpooledHeapByteBuf	JVM memory, just wait for GC
UnpooledDirectByteBuf	Direct memory needs special methods for timely recovery, although JVM uses phantom references to ensure final recovery
PooledByteBuf	Uses pooling mechanism, needs more complex rules for memory recovery

Memory management methods

buf.release();
buf.retain();

Source code different implementations recover memory through implementing protected abstract void deallocate()

Netty uses reference counting to control memory recovery. Each ByteBuf implements the ReferenceCounted interface

This interface mainly manages references

• Each ByteBuf object has an initial count of 1
• Calling release decreases count by 1, if count is 0, ByteBuf memory is recycled
• Calling retain increases count by 1, meaning the caller hasn't finished using it yet, so other handlers can't cause recycling even if they call release
• When count is 0, underlying memory is recycled, and even if the ByteBuf object exists, its methods can't be used normally

Question: When should release be called during ByteBuf usage?

Not always applicable method

ByteBuf buf = ...
try {
    ...
} finally {
    buf.release();
}

Correct approach: The last user is responsible for release. Due to the pipeline, ByteBuf usually needs to be passed to the next ChannelHandler. If release is called in finally, the pass-through is lost unless the ByteBuf is truly no longer needed.

General rule for ByteBuf release in business code:

• Starting point, for NIO implementation, in io.netty.channel.nio.AbstractNioByteChannel.NioByteUnsafe when ByteBuf is first created and placed in pipeline
• Don't process original ByteBuf, call ctx.fireChannelRead(msg) to pass backward, no need to release
• Convert original ByteBuf to other Java objects, then ByteBuf is useless and must be released
• If not calling ctx.fireChannelRead(msg) to pass backward, must release
• Note various exceptions, if ByteBuf isn't successfully passed to the next ChannelHandler, must release
• Assuming message passes all the way, TailContext is responsible for releasing unprocessed messages (original ByteBuf)
• Outbound ByteBuf handling principle
• Outbound messages eventually become ByteBuf output, passing forward, released by HeadContext after flush
• Exception handling principle
• Sometimes unsure how many times ByteBuf is referenced but must be completely released, can loop release until true is returned

Netty's pipeline underlying data structure is a doubly-linked list, with default head and tail handler nodes. Head and tail nodes have source code for ByteBuf memory release.

Pipeline default tail node source code

// final class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler
       @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 {
            // Ultimately calls release method of ReferenceCountUtil
            ReferenceCountUtil.release(msg);
        }
    }
--------------------------------------------------------------------------------
    public static boolean release(Object msg) {
        if (msg instanceof ReferenceCounted) {   // Check if it's bytebuf
            return ((ReferenceCounted) msg).release();
        }
        return false;
    }

Summary: The channelRead method in the default tail handler ultimately calls release to free ByteBuf memory

Pipeline default head node HeadContext source code

• This class implements both ChannelOutboundHandler and ChannelInboundHandler, so it can handle both inbound and outbound information

 // final class HeadContext extends AbstractChannelHandlerContext
     @Override
     public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
                unsafe.write(msg, promise);
     }
====================================================================
        @Override
        public final void write(Object msg, ChannelPromise promise) {
            assertEventLoop();

            ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
            if (outboundBuffer == null) {
                // If outboundBuffer is null, channel is closed, fail future immediately
                safeSetFailure(promise, newClosedChannelException(initialCloseCause));
                // Release message now to prevent resource leak
                ReferenceCountUtil.release(msg);     // Release memory
                return;
            }

            int size;
            try {
                msg = filterOutboundMessage(msg);
                size = pipeline.estimatorHandle().size(msg);
                if (size < 0) {
                    size = 0;
                }
            } catch (Throwable t) {
                safeSetFailure(promise, t);
                ReferenceCountUtil.release(msg);     // Release memory
                return;
            }

            outboundBuffer.addMessage(msg, size, promise);
        }

Zero-copy Operations

Operation Name	Description
slice	Slice original ByteBuf into multiple ByteBuf without memory copy, sharing memory but maintaining independent read/write pointers, and fixed capacity cannot be written.
duplicate	Uses same underlying memory as original ByteBuf, but with independent read/write pointers
CompositeByteBuf	Can logically combine multiple ByteBuf without copying

Note:

• Since these operations share memory, retain must be performed after operation to prevent other ByteBuf references from becoming invalid when original ByteBuf is released
• Sliced ByteBuf cannot be written due to memory sharing.

package netty_components.buffer;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.util.internal.SocketUtils;

import static io.netty.buffer.ByteBufUtil.appendPrettyHexDump;
import static io.netty.util.internal.StringUtil.NEWLINE;

public class TestSliceAndDuplicate {
    public static void main(String[] args) {
        ByteBuf bf = ByteBufAllocator.DEFAULT.buffer();
        bf.writeBytes(new byte[]{1, 2, 3, 4});

        System.out.println("ByteBuf obtained from slice and duplicate operations (p1, p2, p3) shares same memory as original ByteBuf (bf):");
        ByteBuf p1 = bf.slice(0,2); log(p1);
        p1.retain();
        ByteBuf p2 = bf.slice(2,2); log(p2);
        p2.retain();
        ByteBuf p3 = bf.duplicate(); log(p3); p3.retain();
        // retain increases original ByteBuf memory reference count, preventing other references from becoming invalid when original is released

        System.out.println("Modifying p1 affects other ByteBuf due to shared memory");
        p1.setByte(0,6); log(bf);
        log(p3);
        bf.release(); p1.release(); p2.release(); p3.release();
    }

    private static void log(ByteBuf buffer) {
        int length = buffer.readableBytes();
        int rows = length / 16 + (length % 15 == 0 ? 0 : 1) + 4;
        StringBuilder buf = new StringBuilder(rows * 80 * 2)
                .append("read index:").append(buffer.readerIndex())
                .append(" write index:").append(buffer.writerIndex())
                .append(" capacity:").append(buffer.capacity())
                .append(NEWLINE);
        appendPrettyHexDump(buf, buffer);
        System.out.println(buf.toString());
    }
}

Execution result: Memory is indeed shared

ByteBuf obtained from slice and duplicate operations (p1, p2, p3) shares same memory as original ByteBuf (bf):
read index:0 write index:2 capacity:2
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02                                           |..              |
+--------+-------------------------------------------------+----------------+
read index:0 write index:2 capacity:2
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 03 04                                           |..              |
+--------+-------------------------------------------------+----------------+
read index:0 write index:4 capacity:256
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04                                     |....            |
+--------+-------------------------------------------------+----------------+
Modifying p1 affects other ByteBuf due to shared memory
read index:0 write index:4 capacity:256
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 06 02 03 04                                     |....            |
+--------+-------------------------------------------------+----------------+
read index:0 write index:4 capacity:256
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 06 02 03 04                                     |....            |
+--------+-------------------------------------------------+----------------+

Method to combine two ByteBuf without copying:

ByteBuf buf1 = ByteBufAllocator.DEFAULT.buffer(5);
buf1.writeBytes(new byte[]{1, 2, 3, 4, 5});
ByteBuf buf2 = ByteBufAllocator.DEFAULT.buffer(5);
buf2.writeBytes(new byte[]{6, 7, 8, 9, 10});
CompositeByteBuf buf3 = ByteBufAllocator.DEFAULT.compositeBuffer();
// true means automatically increment write index when adding new ByteBuf, otherwise write index remains 0
buf3.addComponents(true, buf1, buf2);

Non-pooled ByteBuf combination

package netty_components.buffer;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled;

public class TestUnpooled {
    public static void main(String[] args) {
        ByteBuf buf1 = ByteBufAllocator.DEFAULT.buffer(5);
        buf1.writeBytes(new byte[]{1, 2, 3, 4, 5});
        ByteBuf buf2 = ByteBufAllocator.DEFAULT.buffer(5);
        buf2.writeBytes(new byte[]{6, 7, 8, 9, 10});

        // Unpooled uses CompositeByteBuf underneath to avoid copying
        ByteBuf buf3 = Unpooled.wrappedBuffer(buf1, buf2);
        System.out.println(ByteBufUtil.prettyHexDump(buf3));

        ByteBuf buf4 = Unpooled.wrappedBuffer(new byte[]{1, 2, 3}, new byte[]{4, 5, 6});
        System.out.println(buf4.getClass());
        System.out.println(ByteBufUtil.prettyHexDump(buf4));
    }
}

Execution Result

         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 05 06 07 08 09 0a                   |..........      |
+--------+-------------------------------------------------+----------------+
class io.netty.buffer.CompositeByteBuf
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 05 06                               |......          |
+--------+-------------------------------------------------+----------------+