Netty全局流量整形

Question

提问：Java如何在netty服务器实现全局流量整形
因为我的请求Channel都是短请求，对单个Channel限流不适用，我需要对全局的所有Channel进行统一限流
我现在已有一个netty服务器代码 如下图

根据已有教程添加GlobalTrafficShapingHandler或GlobalChannelTrafficShapingHandler均无法实现功能，添加后导致客户端请求服务端接口不通，debug查看发现Channel被remove了。
且我还需要额外实现对流量整形速率的动态调整，从redis获取数据，比较已有限制速率和新限制速率是否相同，不同则更新限制速率。

Answer 1

引自免费微信小程序：皆我百晓生

首先，你需要确保你已经配置好了NioEventLoopGroup，它是Netty的核心事件循环组，用于管理多线程并发的活动。

下面是一个简单的示例代码片段，展示如何使用Java 8的.nio.NioServerSocketChannel创建一个Nio服务器，并通过全局流量整形进行处理：

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;

public class GlobalTrafficShapingHandler implements ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ch.pipeline()
                .addLast(new HttpServerCodec())
                .addLast(new HttpObjectAggregator(maxcontentLength: 512 * 1024))
                .addLast(new GlobalTrafficShapingHandler());
    }

    @Override
    public void addLast(ChannelPipeline pipeline) throws Exception {
        pipeline.addLast(new GlobalTrafficShapingHandler());
    }
}

class GlobalTrafficShapingHandler extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ch.pipeline()
                .addLast(new HttpServerCodec())
                .addLast(new GlobalTrafficShapingHandler());
    }

    @Override
    public void addLast(ChannelPipeline pipeline) throws Exception {
        pipeline.addLast(new GlobalTrafficShapingHandler());
    }
}

在这个示例中，我们定义了一个名为GlobalTrafficShapingHandler的类，它继承自ChannelInitializer接口。initChannel方法中，我们指定了一个新的管道，并在管道中添加了一个全局流量整形器GlobalTrafficShapingHandler。这个处理器将接收来自客户端的HTTP对象，并将其聚合到总的最大内容长度为512KB的分组中。然后，这个最大分组会被添加到全球的总流量中。

你需要设置正确的IP地址、端口和请求通道的大小。这可以通过修改ServerBootstrap对象中的相关参数来完成。例如，如果你想使用1234567890的IP地址和8080端口，你可以在ServerBootstrap对象中这样设置：

ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
    .channel(NioServerSocketChannel.class)
    .childOption(ChannelOption.SO_REUSEADDR, true);
b.bind(host, port);

然后，你可以启动你的服务器并开始监听请求通道：

ChannelFuture f = b.bind(host, port).sync();
f.channel().closeFuture().sync();

关于如何获取并比较限制速率，你可以在客户端发送请求时使用Java 8的HttpRequestExecutor类。例如：

HttpRequestExecutor executor = Executors.newFixedThreadPool(nThreads);
executor.execute(new RequestHandler());

HttpResponse<String> response = executor.submit(new ResponseHandler());
String content = response.body();

System.out.println(content);

executor.shutdownNow();

这里，我们创建了一个固定大小的线程池，然后使用它提交了一个RequestHandler任务。然后，我们调用submit方法将另一个ResponseHandler任务提交给线程池。这两个任务都会阻塞，直到它们完成。

在每个任务完成后，我们打印出响应的内容。

注意，你需要确保HttpRequestExecutor不会阻塞超过指定数量的线程池工作。在上面的例子中，我们将线程池的大小设置为nThreads（即等于CPU核心数）。你可能需要根据你的应用程序的具体需求来调整这个值。

如果你想要在收到请求时立即更新限制速率，你可以在InitHandler方法中添加一个新的操作。例如：

private int rateLimit = 100; // 假设这是一个512KB的最大限制速率

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    ...
    ch.pipeline()
            .addLast(new GlobalTrafficShapingHandler())
            .addLast(new InitHandler());
}

在InitHandler方法中，我们可以创建一个新的操作，用于检查当前的速率限制是否已经被消耗完。如果是，我们可以返回一个错误消息，否则我们可以尝试增加速率限制。