快捷導(dǎo)航

在netty中使用native傳輸協(xié)議的方法

更新時(shí)間：2022年05月26日 17:02:59 作者：程序那些事

這篇文章主要介紹了在netty中使用native傳輸協(xié)議,這里我們只以Kqueue為例介紹了netty中native傳輸協(xié)議的使用，需要的朋友可以參考下

簡(jiǎn)介

對(duì)于IO來(lái)說(shuō)，除了傳統(tǒng)的block IO,使用最多的就是NIO了，通常我們?cè)趎etty程序中最常用到的就是NIO,比如NioEventLoopGroup,NioServerSocketChannel等。

我們也知道在IO中有比NIO更快的IO方式，比如kqueue和epoll，但是這兩種方式需要native方法的支持，也就是說(shuō)需要在操作系統(tǒng)層面提供服務(wù)。

如果我們?cè)谥С諯queue或者epoll的服務(wù)器上，netty是否可以提供對(duì)這些優(yōu)秀IO的支持呢？

答案是肯定的。但是首先kqueue和epoll需要JNI支持，也就是說(shuō)JAVA程序需要調(diào)用本地的native方法。

native傳輸協(xié)議的依賴

要想使用kequeue和epoll這種native的傳輸方式，我們需要額外添加項(xiàng)目的依賴,如果是linux環(huán)境，則可以添加如下的maven依賴環(huán)境：

  <dependencies>
    <dependency>
      <groupId>io.netty</groupId>
      <artifactId>netty-transport-native-epoll</artifactId>
      <version>${project.version}</version>
      <classifier>linux-x86_64</classifier>
    </dependency>
    ...
  </dependencies>

其中version需要匹配你所使用的netty版本號(hào)，否則可能出現(xiàn)調(diào)用異常的情況。

classifier表示的是系統(tǒng)架構(gòu)，它的值可以是linux-x86_64，也可以是linux-aarch_64.

如果你使用的mac系統(tǒng)，那么可以這樣引入：

  <dependencies>
    <dependency>
      <groupId>io.netty</groupId>
      <artifactId>netty-transport-native-kqueue</artifactId>
      <version>${project.version}</version>
      <classifier>osx-x86_64</classifier>
    </dependency>
    ...
  </dependencies>

netty除了單獨(dú)的個(gè)體包之外，還有一個(gè)all in one的netty-all包，如果你使用了這個(gè)all in one的包，那么不需要額外添加native的依賴。

如果netty提供的系統(tǒng)架構(gòu)并沒有你正在使用的，那么你需要手動(dòng)進(jìn)行編譯，以下是編譯所依賴的程序包, 如果是在RHEL/CentOS/Fedora系統(tǒng)中，則使用：

sudo yum install autoconf automake libtool make tar \
                 glibc-devel \
                 libgcc.i686 glibc-devel.i686

如果是在Debian/Ubuntu系統(tǒng)中，則使用：

sudo apt-get install autoconf automake libtool make tar \
                     gcc

如果是在MacOS/BSD系統(tǒng)中，則使用：

brew install autoconf automake libtool

netty本地傳輸協(xié)議的使用

安裝好依賴包之后，我們就可以在netty中使用這些native傳輸協(xié)議了。

native傳輸協(xié)議的使用和NIO的使用基本一致，我們只需要進(jìn)行下面的替換即可。

如果是在liunx系統(tǒng)中，則進(jìn)行下面的替換：

    NioEventLoopGroup → EpollEventLoopGroup
    NioEventLoop → EpollEventLoop
    NioServerSocketChannel → EpollServerSocketChannel
    NioSocketChannel → EpollSocketChannel

如果是在mac系統(tǒng)中，則進(jìn)行下面的替換：

    NioEventLoopGroup → KQueueEventLoopGroup
    NioEventLoop → KQueueEventLoop
    NioServerSocketChannel → KQueueServerSocketChannel
    NioSocketChannel → KQueueSocketChannel

這里還是使用我們熟悉的聊天服務(wù)為例，首先看下基于Kqueue的netty服務(wù)器端應(yīng)該怎么寫：

EventLoopGroup bossGroup = new KQueueEventLoopGroup(1);
        EventLoopGroup workerGroup = new KQueueEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(KQueueServerSocketChannel.class)
             .handler(new LoggingHandler(LogLevel.INFO))
             .childHandler(new NativeChatServerInitializer());

            Channel channel = b.bind(PORT).sync().channel();
            log.info("server channel:{}", channel);
            channel.closeFuture().sync();

和NIO一樣，在服務(wù)器端我們需要使用KQueueEventLoopGroup創(chuàng)建兩個(gè)EventLoopGroup，一個(gè)是bossGroup, 一個(gè)是workerGroup。

然后將這兩個(gè)group傳入到ServerBootstrap中，并且添加KQueueServerSocketChannel作為channel。

其他的內(nèi)容和NIO server的內(nèi)容是一樣的。

接下來(lái)我們看下基于Kqueue的netty客戶端改如何跟server端建立連接：

EventLoopGroup group = new KQueueEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
             .channel(KQueueSocketChannel.class)
             .handler(new NativeChatClientInitializer());

            // 建立連接
            Channel ch = b.connect(HOST, PORT).sync().channel();
            log.info("client channel: {}", ch);

這里使用的是KQueueEventLoopGroup，并將KQueueEventLoopGroup放到Bootstrap中，并且為Bootstrap提供了和server端一致的KQueueSocketChannel。

然后就是客戶端向channel中寫消息，這里我們直接從命令行輸入：

// 從命令行輸入
            ChannelFuture lastWriteFuture = null;
            BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
            for (;;) {
                String line = in.readLine();
                if (line == null) {
                    break;
                }
                // 將從命令行輸入的一行字符寫到channel中
                lastWriteFuture = ch.writeAndFlush(line + "\r\n");
                // 如果輸入'再見'，則等待server端關(guān)閉channel
                if ("再見".equalsIgnoreCase(line)) {
                    ch.closeFuture().sync();
                    break;
                }
            }

上面代碼的意思是將命令行收到的消息寫入到channel中，如果輸入的是’再見’，則關(guān)閉channel。

為了能夠處理字符串，這里用到了三個(gè)編碼解碼器：

        // 添加行分割器
        pipeline.addLast(new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
        // 添加String Decoder和String Encoder,用來(lái)進(jìn)行字符串的轉(zhuǎn)換
        pipeline.addLast(new StringEncoder());
        pipeline.addLast(new StringDecoder());

分別是行分割器，字符編碼器和字符解碼器。

運(yùn)行一下看，程序運(yùn)行沒問(wèn)題，客戶端和服務(wù)器端可以進(jìn)行通訊。