本節我們看看Netty的傳輸（全是干貨，自帶水杯

一、Java的NIO和OIO

流經網路的資料總是具有相同的型別：位元組，這些位元組是如何流動的主要取決于我們所說的網路傳輸，

1.1 OIO

我們先來看一段Java的阻塞應用程式程式：

package com.example.javademo;

import java.io.IOException;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.charset.Charset;

/**
 * @author lhd
 * @date 2023/05/17 13:19
 * @notes java oio演示
 */
public class PlainOioServer {

    public void serve(int port) throws IOException {
        //系結服務器到指定埠
        final ServerSocket socket = new ServerSocket(port);
        try {
            for (;;) {
                //接受連接
                final Socket clientSocket = socket.accept();
                System.out.println(
                        "Accepted connection from " + clientSocket);
                //創建一個新的執行緒來處理該連接
                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        OutputStream out;
                        try {
                            out = clientSocket.getOutputStream();
                            //將訊息寫給新連接的客戶端
                            out.write("Hi!\r\n".getBytes(Charset.forName("UTF-8")));
                            out.flush();
                            //關閉連接
                            clientSocket.close();
                        }
                        catch (IOException e) {
                            e.printStackTrace();
                        }
                        finally {
                            try {
                                clientSocket.close();
                            }
                            catch (IOException ex) {
                            // 關閉時忽略
                            }
                        }
                    }
                }).start(); //啟動執行緒
            }
        }
        catch (IOException e) {
            e.printStackTrace();
        }
    }
}

這段代碼完全可以處理中等數量的并發客戶端，但是隨著應用程式變得流行起來，你會發現它并不能很好地伸縮到支撐成千上萬的并發連入連接，

1.2 NIO

同樣，來看一段Java的非阻塞應用程式代碼：

package com.example.javademo;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

/**
 * @author lhd
 * @date 2023/05/17 13:27
 * @notes java NIO演示
 */
public class PlainNioServer {

    public void serve(int port) throws IOException {
        ServerSocketChannel serverChannel = ServerSocketChannel.open();
        serverChannel.configureBlocking(false);
        ServerSocket ssocket = serverChannel.socket();
        //將服務系結到埠
        InetSocketAddress address = new InetSocketAddress(port);
        ssocket.bind(address);
        //打開選擇器來處理Channel
        Selector selector = Selector.open();
        //將serverChannel注冊到選擇器用來接收連接
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);
        final ByteBuffer msg = ByteBuffer.wrap("Hi!\r\n".getBytes());
        for (;;) {
            try {
                //等待需要處理的新事件；阻塞 將一直持續到下一個傳入事件
                selector.select();
            } catch (IOException ex) {
                ex.printStackTrace();
           // handle exception
                break;
            }
            //獲取所有接收事件的SelectionKey 實體
            Set<SelectionKey> readyKeys = selector.selectedKeys();
            Iterator<SelectionKey> iterator = readyKeys.iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                iterator.remove();
                try {
                    //檢查事件是否是一個新的已經就緒可以被接受的連接
                    if (key.isAcceptable()) {
                        ServerSocketChannel server = (ServerSocketChannel)key.channel();
                        SocketChannel client = server.accept();
                        client.configureBlocking(false);
                        //接受客戶端，并將它注冊到選擇器
                        client.register(selector, SelectionKey.OP_WRITE |
                                SelectionKey.OP_READ, msg.duplicate());
                        System.out.println("Accepted connection from " + client);
                    }
                    //檢查套接字是否已經準備好寫資料
                    if (key.isWritable()) {
                        SocketChannel client = (SocketChannel)key.channel();
                        ByteBuffer buffer = (ByteBuffer)key.attachment();
                        while (buffer.hasRemaining()) {
                            //將資料寫到已連接的客戶端
                            if (client.write(buffer) == 0) {
                                break;
                            }
                        }
                        //關閉連接
                        client.close();
                    }
                } catch (IOException ex) {
                    key.cancel();
                    try {
                        key.channel().close();
                    } catch (IOException cex) {
                    }
                }
            }
        }
    }
}

可以看出，兩段Java的OIO和NIO都做了相同的事：連接客戶端，并發送“Hi”，區別在于一個是阻塞的，另一個是非阻塞的，但兩段代碼卻完全不同，如果為了用于非阻塞 I/O 而重新實作這個簡單的應用程式，都需要一次完全的重寫的話，那么不難想象，移植真正復雜的應用程式需要付出什么樣的努力，

二、Netty的NIO和OIO

下來我們來看以下Netty 實作該應用程式將會是什么樣子，

2.1 OIO

來看一段Netty的阻塞代碼

package com.example.netty;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.oio.OioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.oio.OioServerSocketChannel;

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

/**
 * @author lhd
 * @date 2023/05/17 13:47
 * @notes Netty 阻塞程式演示
 * ps：Netty已啟用OIO傳輸，使用 NIO / EPOLL / KQUEUE 傳輸，
 */
public class NettyOioServer {
    public void server(int port) throws InterruptedException {
        final ByteBuf buf = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("Hi!\r\n", StandardCharsets.UTF_8));
        EventLoopGroup group = new OioEventLoopGroup();
        try {
            //創建ServerBootstrap
            ServerBootstrap b = new ServerBootstrap();
            b.group(group)
                    //使用 OioEventLoopGroup以允許阻塞模式
                    .channel(OioServerSocketChannel.class)
                    .localAddress(new InetSocketAddress(port))
                    //指定 ChannelInitializer，對于每個已接受的連接都呼叫它
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch){
                            //添加一個 ChannelInboundHandlerAdapter 以攔截和處理事件
                            ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                                @Override
                                public void channelActive(ChannelHandlerContext ctx) {
                                    //將訊息寫到客戶端，并添加 ChannelFutureListener，以便訊息一被寫完就關閉連接
                                    ctx.writeAndFlush(buf.duplicate()).addListener(ChannelFutureListener.CLOSE);
                                }
                            });
                        }
                    });
            //系結服務器以接受連接
            ChannelFuture f = b.bind().sync();
            //釋放所有的資源
            f.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully().sync();
        }
    }
}

ps：這段代碼和我們上一篇文章中的有所區別，不要搞混淆了

2.2 NIO

我們再來看看Netty的非阻塞代碼：

package com.example.netty;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

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

/**
 * @author lhd
 * @date 2023/05/17 14:05
 * @notes Netty 非阻塞代碼演示
 */
public class NettyNioServer {

    public void server(int port) throws InterruptedException {
        final ByteBuf buf = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("Hi!\r\n", StandardCharsets.UTF_8));
       EventLoopGroup group = new NioEventLoopGroup(); // 1
        try {
            ServerBootstrap b = new ServerBootstrap();
            //為非阻塞模式使用NioEventLoopGroup
            b.group(group).channel(NioServerSocketChannel.class) // 2
                    .localAddress(new InetSocketAddress(port))
                    //指定 ChannelInitializer，對于每個已接受的連接都呼叫它
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) {
                            //添加 ChannelInboundHandlerAdapter 以接收和處理事件
                            ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                                @Override
                                public void channelActive(ChannelHandlerContext ctx) {
                                    //將訊息寫到客戶端，并添加ChannelFutureListener，以便訊息一被寫完就關閉連接
                                    ctx.writeAndFlush(buf.duplicate()).addListener(ChannelFutureListener.CLOSE);
                                }
                            });
                        }
                    });
            //系結服務器以接受連接
            ChannelFuture f = b.bind().sync();
            f.channel().closeFuture().sync();
        } finally {
            //釋放所有的資源
            group.shutdownGracefully().sync();
        }
    }
}

可以看到，Netty的阻塞代碼和非阻塞代碼只有兩處不同（代碼中1 2 處），這使得Netty的阻塞和非阻塞切換非常容易，Netty 為每種傳輸的實作都暴露了相同的 API，所以無論選用哪一種傳輸的實作，你的代碼都仍然幾乎不受影響，在所有的情況下，傳輸的實作都依賴于 interface Channel、ChannelPipeline 和 ChannelHandler，

三、傳輸API

傳輸 API 的核心是 interface Channel，它被用于所有的 I/O 操作，下圖是它的層次結構：

如圖所示，每個 Channel 都將會被分配一個 ChannelPipeline和ChannelConfig，ChannelConfig 包含了該 Channel 的所有配置設定，并且支持熱更新，由于特定的傳輸可能具有獨特的設定，所以它可能會實作一個 ChannelConfig 的子型別，

由于 Channel 是獨一無二的，所以為了保證順序將 Channel 宣告為 java.lang.Comparable 的一個子介面，因此，如果兩個不同的 Channel 實體都回傳了相同的散列碼，那么 AbstractChannel 中的 compareTo()方法的實作將會拋出一個 Error，

ChannelPipeline 持有所有將應用于入站和出站資料以及事件的 ChannelHandler 實體，這些 ChannelHandler 實作了應用程式用于處理狀態變化以及資料處理的邏輯，

ChannelHandler 的典型用途包括：

• 將資料從一種格式轉換為另一種格式；
• 提供例外的通知；
• 提供 Channel 變為活動的或者非活動的通知；
• 提供當 Channel 注冊到 EventLoop 或者從 EventLoop 注銷時的通知；
• 提供有關用戶自定義事件的通知，

ps:ChannelPipeline 實作了一種常見的設計模式—攔截過濾器（InterceptingFilter），UNIX 管道是另外一個熟悉的例子：多個命令被鏈接在一起，其中一個命令的輸出端將連接到命令列中下一個命令的輸入端，

也可以根據需要通過添加或者移除ChannelHandler實體來修改ChannelPipeline，
通過利用Netty的這項能力可以構建出高度靈活的應用程式，例如，STARTTLS方法名協議被請求時，你可以簡單地通過 向 ChannelPipeline 添 加 一個適當的 ChannelHandler（SslHandler）來按需地支持STARTTLS協議，除了訪問所分配的 ChannelPipeline 和 ChannelConfig 之外，也可以利用 Channel的其他方法，

比較重要的Channel方法如下：

Netty 所提供的廣泛功能只依賴于少量的介面，所以我們可以對我們的應用程式邏輯進行重大的修改，而又無需大規模地重構你的代碼庫，

下面是一個寫資料并將其沖刷到遠程節點的例子，

Channel channel = ...
//創建要發送的資料
ByteBuf buf = Unpooled.copiedBuffer("your data", CharsetUtil.UTF_8);
//寫資料并沖刷它
ChannelFuture cf = channel.writeAndFlush(buf);
//添加監聽，以便完后接收通知
cf.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) {
//如果成功，在控制臺列印成功
if (future.isSuccess()) {
System.out.println("Write successful");
} else {
//錯誤則列印錯誤和堆疊跟蹤
System.err.println("Write error");
future.cause().printStackTrace();
}
}
});

ps：Netty 的 Channel 實作是執行緒安全的，因此你可以存盤一個到 Channel 的參考，并且每當你需要向遠程節點寫資料時，都可以使用它，即使當時許多執行緒都在使用它，

上一篇博文說過，一個Channel對應一個eventLoop，一個eventLoop持有一個執行緒卻不一定只持有一個Channel，下面展示多個執行緒使用同一個Channel，

final Channel channel = ...
//創建資料
final ByteBuf buf = Unpooled.copiedBuffer("your data",
CharsetUtil.UTF_8).retain();
//創建寫資料的Runable
Runnable writer = new Runnable() {
@Override
public void run() {
channel.writeAndFlush(buf.duplicate());
}
};
//獲取執行緒池的參考
Executor executor = Executors.newCachedThreadPool();
//遞交任務給執行緒池，以便某個執行緒呼叫
executor.execute(writer);
//遞交另一個寫任務以便在另一個執行緒中執行
executor.execute(writer);
...

這些訊息將會保證按照順序發送~

四、內置的傳輸

Netty 內置了一些可開箱即用的傳輸，因為并不是它們所有的傳輸都支持每一種協議，所以你必須選擇一個和你的應用程式所使用的協議相容的傳輸，

下面是一些Netty提供的傳輸，

之后我們回詳細介紹這些傳輸，

ps:Epoll這個是 Netty 特有的實作，更加適配 Netty 現有的執行緒模型，具有更高的性能以及更低的垃圾回收壓力

4.1 NIO

NIO 提供了一個所有 I/O 操作的全異步的實作，它利用了自 NIO 子系統被引入JDK 1.4 時便可用的基于選擇器的 API，

選擇器背后的基本概念是充當一個注冊表，在那里你將可以請求在 Channel 的狀態發生變化時得到通知，可能的狀態變化有：

• 新的 Channel 已被接受并且就緒；
• Channel 連接已經完成；
• Channel 有已經就緒的可供讀取的資料；
• Channel 可用于寫資料，

選擇器運行在一個檢查狀態變化并對其做出相應回應的執行緒上，在應用程式對狀態的改變做出回應之后，選擇器將會被重置，并將重復這個程序，

下表中的常量值代表了由class java.nio.channels.SelectionKey定義的位模式，
這些位模式可以組合起來定義一組應用程式正在請求通知的狀態變化集，

內置傳輸的處理流程圖：

ps：零拷貝（zero-copy）是一種目前只有在使用 NIO 和 Epoll 傳輸時才可使用的特性，它使你可以快速高效地將資料從檔案系統移動到網路介面，而不需要將其從內核空間復制到用戶空間，其在像 FTP 或者HTTP 這樣的協議中可以顯著地提升性能，但是，并不是所有的作業系統都支持這一特性，特別地，它對于實作了資料加密或者壓縮的檔案系統是不可用的——只能傳輸檔案的原始內容，反過來說，傳輸已被加密的檔案則不是問題

4.2 Epoll—用于 Linux 的本地非阻塞傳輸

Netty 的 NIO 傳輸基于 Java 提供的異步/非阻塞網路編程的通用抽象，雖然這保證了 Netty 的非阻塞 API 可以在任何平臺上使用，但它也包含了相應的限制，因為 JDK為了在所有系統上提供相同的功能，必須做出妥協，

Netty為Linux提供了一組NIO API，其以一種和它本身的設計更加一致的方式使用epoll，并且以一種更加輕量的方式使用中斷，如果你的應用程式旨在運行于Linux系統，那么請考慮利用這個版本的傳輸；你將發現在高負載下它的性能要優于JDK的NIO實作，

這個傳輸的語意和上一節內置傳輸的處理流程圖 所示的完全相同，而且它的用法也是簡單直接的，相關示例參照上面的Netty NIO代碼，如果要在那個代碼中使用 epoll 替代 NIO，只需要將 NioEventLoopGroup替換為EpollEventLoopGroup ，并且將 NioServerSocketChannel.class 替換為EpollServerSocketChannel.class 即可，

4.3 OIO

這個可以參照Netty的OIO代碼即可，最新的Netty中拋棄了OIO的傳輸方式~

下圖是OIO的傳輸處理邏輯：

4.4 用于 JVM 內部通信的 Local 傳輸

Netty 提供了一個 Local 傳輸，用于在同一個 JVM 中運行的客戶端和服務器程式之間的異步通信，同樣，這個傳輸也支持對于所有 Netty 傳輸實作都共同的 API，

在這個傳輸中，和服務器 Channel 相關聯的 SocketAddress 并沒有系結物理網路地址；相反，只要服務器還在運行，它就會被存盤在注冊表里，并在 Channel 關閉時注銷，因為這個傳輸并不接受真正的網路流量，所以它并不能夠和其他傳輸實作進行互操作，因此，客戶端希望連接到（在同一個 JVM 中）使用了這個傳輸的服務器端時也必須使用它，除了這個限制，它的使用方式和其他的傳輸一模一樣，

4.5 Embedded 傳輸

Netty 提供了一種額外的傳輸，使得你可以將一組 ChannelHandler 作為幫助器類嵌入到其他的 ChannelHandler 內部，通過這種方式，你將可以擴展一個ChannelHandler 的功能，而又不需要修改其內部代碼，

Embedded 傳輸的關鍵是一個被稱為 EmbeddedChannel 的具體的 Channel
實作，我將在之后的文章中為大家演示這種方式，

五、傳輸的用例

Netty支持的傳輸和網路協議：

PS:UDT 協議實作了基于 UDP 協議的可靠傳輸;

在 Linux 上啟用 SCTP
SCTP 需要內核的支持，并且需要安裝用戶庫，
例如，對于 Ubuntu，可以使用下面的命令：# sudo apt-get install libsctp1
對于 Fedora，可以使用 yum：#sudo yum install kernel-modules-extra.x86_64 lksctp-tools.x86_64

雖然只有SCTP傳輸有這些特殊要求，但是其他傳輸可能也有它們自己的配置選項需要考慮，此外，如果只是為了支持更高的并發連接數，服務器平臺可能需要配置得和客戶端不一樣，

這里是一些很可能會遇到的用例

• 非阻塞代碼庫——如果你的代碼庫中沒有阻塞呼叫（或者你能夠限制它們的范圍），那么在 Linux 上使用 NIO 或者 epoll 始終是個好主意，雖然 NIO/epoll 旨在處理大量的并發連接，但是在處理較小數目的并發連接時，它也能很好地作業，尤其是考慮到它在連接之間共享執行緒的方式，
• 阻塞代碼庫——正如我們已經指出的，如果你的代碼庫嚴重地依賴于阻塞 I/O，而且你的應用程式也有一個相應的設計，那么在你嘗試將其直接轉換為 Netty 的 NIO 傳輸時，你將可能會遇到和阻塞操作相關的問題，不要為此而重寫你的代碼，可以考慮分階段遷移：先從OIO 開始，等你的代碼修改好之后，再遷移到 NIO（或者使用 epoll，如果你在使用 Linux），
• 在同一個 JVM 內部的通信——在同一個 JVM 內部的通信，不需要通過網路暴露服務，是Local 傳輸的完美用例，這將消除所有真實網路操作的開銷，同時仍然使用你的 Netty 代碼庫，如果隨后需要通過網路暴露服務，那么你將只需要把傳輸改為 NIO 或者 OIO 即可，
• 測驗你的 ChannelHandler 實作——如果你想要為自己的 ChannelHandler 實作撰寫單元測驗，那么請考慮使用 Embedded 傳輸，這既便于測驗你的代碼，而又不需要創建大量的模擬（mock）物件，你的類將仍然符合常規的 API 事件流，保證該 ChannelHandler在和真實的傳輸一起使用時能夠正確地作業，

應用程式的最佳傳輸建議：

標籤：其他

上一篇：Python工具箱系列(三十三)

下一篇：返回列表