求助：如何在Hazelcast Jet Kafka中实现按时间单位限制吞吐量？

嗨，我刚好之前研究过Hazelcast Jet里给Kafka源做吞吐量限制的方案，给你分享几个可行的思路：

这个方法适合单实例或者可以按实例拆分吞吐量的场景，核心是借助限流器控制每个元素的处理速率，比如用Guava的RateLimiter：

import com.hazelcast.jet.pipeline.Pipeline;
import com.hazelcast.jet.pipeline.Sources;
import com.google.common.util.concurrent.RateLimiter;
import java.util.concurrent.CompletableFuture;

public class KafkaThrottleExample {
    public static Pipeline buildThrottledKafkaPipeline(String kafkaTopic, int maxElementsPerSecond) {
        // 初始化限流器，设置每秒允许处理的元素数
        RateLimiter rateLimiter = RateLimiter.create(maxElementsPerSecond);
        
        Pipeline pipeline = Pipeline.create();
        pipeline.readFrom(Sources.kafka(kafkaProps(), kafkaTopic))
                // 用异步步骤做限流，阻塞直到获取处理许可
                .mapAsync(event -> {
                    rateLimiter.acquire();
                    return CompletableFuture.completedFuture(event);
                })
                // 这里接你的后续处理逻辑
                .writeTo(Sinks.logger());
        
        return pipeline;
    }
    
    // 这里是你的Kafka配置方法
    private static Properties kafkaProps() {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092");
        props.setProperty("group.id", "jet-kafka-throttle");
        // 其他必要配置...
        return props;
    }
}

注意：如果是集群部署，每个Jet实例会独立使用自己的限流器，所以要根据实例总数拆分总吞吐量。比如总目标是每秒1000条，3个实例的话每个实例设置为约333条/秒。

如果需要全局统一的吞吐量控制，就需要借助Hazelcast的分布式数据结构来跟踪全局处理计数，自定义Processor实现：

import com.hazelcast.jet.core.Processor;
import com.hazelcast.jet.core.ProcessorContext;
import com.hazelcast.core.IAtomicLong;
import java.util.concurrent.locks.LockSupport;

public class GlobalThrottlingProcessor implements Processor {
    private IAtomicLong globalCounter;
    private long currentWindowStart;
    private final int maxPerSecond;
    private static final long WINDOW_MS = 1000; // 1秒时间窗口

    public GlobalThrottlingProcessor(int maxPerSecond) {
        this.maxPerSecond = maxPerSecond;
    }

    @Override
    public void init(ProcessorContext context) {
        // 用分布式原子长整型跟踪全局处理数
        globalCounter = context.hazelcastInstance().getAtomicLong("kafka-throttle-global-counter");
        currentWindowStart = System.currentTimeMillis();
    }

    @Override
    public boolean process(int ordinal, Object item) {
        long now = System.currentTimeMillis();
        // 检查是否进入新的时间窗口，重置计数器
        if (now - currentWindowStart >= WINDOW_MS) {
            globalCounter.set(0);
            currentWindowStart = now;
        }
        // 循环等待直到获取处理许可，避免CPU空转
        while (globalCounter.get() >= maxPerSecond) {
            LockSupport.parkNanos(100_000);
            now = System.currentTimeMillis();
            if (now - currentWindowStart >= WINDOW_MS) {
                globalCounter.set(0);
                currentWindowStart = now;
            }
        }
        // 拿到许可后计数+1，再传递元素
        globalCounter.incrementAndGet();
        emit(item);
        return true;
    }
}

然后在管道中引入这个自定义处理器：

pipeline.readFrom(Sources.kafka(kafkaProps(), kafkaTopic))
        .customTransform("global-throttle", 
            () -> ProcessorMetaSupplier.of(() -> new GlobalThrottlingProcessor(1000)))
        .writeTo(Sinks.logger());

如果不需要严格的限流，只是想避免突发流量过载，可以用Jet的滑动窗口统计单位时间内的元素量，超过阈值时过滤或采样：

import com.hazelcast.jet.pipeline.WindowDefinition;
import com.hazelcast.jet.aggregate.AggregateOperations;
import java.util.stream.Collectors;

pipeline.readFrom(Sources.kafka(kafkaProps(), kafkaTopic))
        // 1秒滑动窗口，每100ms统计一次
        .window(WindowDefinition.sliding(1000, 100))
        .aggregate(AggregateOperations.counting())
        .map(windowResult -> {
            long count = windowResult.result();
            if (count <= 1000) {
                return windowResult.items();
            } else {
                // 只保留前1000个元素，或者按比例采样
                return windowResult.items().stream().limit(1000).collect(Collectors.toList());
            }
        })
        .writeTo(Sinks.logger());

内容的提问来源于stack exchange，提问作者Aliksz