使用 Netty 搭建只有少数客户端的服务器答案

【问题标题】：Using Netty to build a server with only few clients使用 Netty 搭建只有少数客户端的服务器
【发布时间】：2018-06-23 15:51:03
【问题描述】：

我熟悉Netty 的基础知识，并使用它构建了一个典型的运行在 TCP 上的应用程序服务器，旨在为许多客户端/连接提供服务。但是，我最近需要构建一个服务器，该服务器旨在处理少数客户端或大多数情况下仅处理一个客户端。但是客户端是许多设备的网关，因此会为我尝试设计的服务器产生大量流量。

我的问题是：

对于这个用例是否可以/推荐使用Netty？我看过here的讨论。
是否可以对管道中的通道处理程序使用多线程 EventExecutor，以便通过 EventExecutor 线程池而不是通道 EventLoop 来实现并发？它会确保来自客户端的一条消息将由一个线程通过所有处理程序处理，而下一条消息由另一个线程处理？
是否有可用的示例实现？

【问题讨论】：

标签： java multithreading netty

【解决方案1】：

根据io.netty.channel.oio 的documentation，如果您没有很多客户，您可以使用它。在这种情况下，每个连接都将在单独的线程中处理，并在后台使用 Java 旧的阻塞 IO。看看OioByteStreamChannel::activate：

/**
 * Activate this instance. After this call {@link #isActive()} will return {@code true}.
 */
protected final void activate(InputStream is, OutputStream os) {
    if (this.is != null) {
        throw new IllegalStateException("input was set already");
    }
    if (this.os != null) {
        throw new IllegalStateException("output was set already");
    }
    if (is == null) {
        throw new NullPointerException("is");
    }
    if (os == null) {
        throw new NullPointerException("os");
    }
    this.is = is;
    this.os = os;
}

如您所见，此处将使用 oio Streams。

根据您的评论。您可以在将处理程序添加到管道时指定 EventExecutorGroup，如下所示：

new ChannelInitializer<Channel> {
      public void initChannel(Channel ch) {
          ch.pipeline().addLast(new YourHandler());
      }
}

我们来看看AbstractChannelHandlerContext：

@Override
public EventExecutor executor() {
    if (executor == null) {
        return channel().eventLoop();
    } else {
        return executor;
    }
}

在这里我们看到，如果您不注册 EventExecutor，它将使用您在创建 ServerBootstrap 时指定的子事件组。

new ServerBootstrap()
      .group(new OioEventLoopGroup(), new OioEventLoopGroup())
               //acceptor group          //child group

下面是调用从通道读取的方法AbstractChannelHandlerContext::invokeChannelRead：

static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
    final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        next.invokeChannelRead(m);
    } else {
        executor.execute(new Runnable() {   //Invoked by the EventExecutor you specified
            @Override
            public void run() {
                next.invokeChannelRead(m);
            }
        });
    }
}

【讨论】：

感谢您的回答。我是否可以在使用 OIO 时将EventExecutor 用于管道中的处理程序？
@SoumyaKanti 更新了我的答案。
谢谢，对我很有帮助。
@SoumyaKanti 欢迎您。要了解EventGroup是如何为孩子Channel注册的，可以看一下private static class ServerBootstrapAcceptor
当然可以。

【解决方案2】：

即使有几个连接，我也会选择NioEventLoopGroup。

关于你的问题：

是否可以对通道使用多线程 EventExecutor 管道中的处理程序，以便代替通道 EventLoop，并发是通过EventExecutor线程池实现的？会吗确保来自客户端的一条消息将由一个线程处理通过所有处理程序，而另一个线程的下一条消息？

Netty 的Channel 保证每个入站或出站消息的处理都将在同一个线程中进行。你不必破解你自己的EventExecutor 来处理这个问题。如果提供入站消息不需要长时间处理，您的代码将看起来像ServerBootstrap 的基本用法。您可能会发现调整池中的线程数很有用。

【讨论】：