通道池中的 Netty 负载均衡器答案

【问题标题】：Netty load balancer in channels pool通道池中的 Netty 负载均衡器
【发布时间】：2013-06-23 10:23:40
【问题描述】：

我将protobuf-rpc-pro 用于我的应用程序（两个系统集成）。 protobuf-rpc-pro 基于 Netty，使用了 Netty 的这个依赖：

io.netty
netty-all
4.0.0.CR1版本>
依赖>

我需要为两个系统之间的通信实现通道池以获得高性能。 IE。在我的实现中，我想获得多个连接，这些连接将并行发送消息（非阻塞通信）。这里的问题是如何检查发送缓冲区是否已满并切换到另一个连接（如负载平衡）。 Netty 或某些外部实现中是否有任何机制？在网上没有找到任何东西...

【问题讨论】：

标签： netty

【解决方案1】：

你看过Channel#isWritable() 方法吗？据我了解，如果写缓冲区太满，那将返回 false。

编辑：添加了使用 Channel#isWritebale() 和 highWaterMark/lowWaterMark 的简单演示。

static InetSocketAddress ADDRESS = new InetSocketAddress("localhost", 4711);

static ChannelPipeline createPipeline() {

    ChannelPipeline pipeline = Channels.pipeline();
    pipeline.addLast("frameDecoder", new DelimiterBasedFrameDecoder(100,
            Delimiters.lineDelimiter()));
    pipeline.addLast("stringDecoder", new StringDecoder(CharsetUtil.UTF_8));

    pipeline.addLast("stringEncoder", new StringEncoder(CharsetUtil.UTF_8));
    return pipeline;

}

public static void main(String[] args) throws Throwable {
    int suspendedWrites = 0;
    int N = 1000;
    final CountDownLatch ready = new CountDownLatch(N);

    ConnectionlessBootstrap server = new ConnectionlessBootstrap(
            new NioDatagramChannelFactory());
    server.setPipeline(createPipeline());
    server.getPipeline().addLast("printer",
            new SimpleChannelUpstreamHandler() {
                @Override
                public void messageReceived(ChannelHandlerContext ctx,
                        MessageEvent e) throws Exception {
                    System.out.println((String) e.getMessage());
                    ready.countDown();
                }
            });
    server.bind(ADDRESS);

    ClientBootstrap client = new ClientBootstrap(
            new NioDatagramChannelFactory());
    client.setPipeline(createPipeline());
    Channel clientChannel = client.connect(ADDRESS).sync().getChannel();

    NioDatagramChannelConfig config = (NioDatagramChannelConfig) clientChannel
            .getConfig();

    config.setWriteBufferLowWaterMark(500);
    config.setWriteBufferHighWaterMark(1000);

    for (int i = 0; i < N; ++i) {
        String message = "Hello number " + (i + 1) + " from client\n";
        if (clientChannel.isWritable())
            clientChannel.write(message);
        else {
            clientChannel.write(message).await();
            ++suspendedWrites;
        }
    }

    ready.await(1, TimeUnit.SECONDS);
    client.releaseExternalResources();
    client.shutdown();
    server.releaseExternalResources();
    server.shutdown();
    System.out.println("Suspended writes: " + suspendedWrites);
    System.out.println("Missed reads: " + ready.getCount());
}

运行这将显示 Channel#isWriteable() 取决于当前的写入缓冲区大小以及高水位线和低水位线的值。

更详细的实现可以监控兴趣操作的变化，并在此基础上提供可写通道。

【讨论】：

谢谢你的回答，我去看看。
对我不起作用。我试图减少缓冲区大小。我有 1000 个执行请求的线程。请求消息大小大于缓冲区大小。听起来它会将所有请求放在某个队列中，并长时间发送所有这些消息。并且在这个发送期间，isWritable 标志始终为真。
也许我遗漏了一些设置？
您的样本看起来很棒！我试试看。
就我而言，WriteBufferLowWaterMark 和 WriteBufferHighWaterMark 设置的值太大。现在它工作正常。谢谢。