我想创建一个基于 Kafka 流的应用程序，该应用程序处理一个主题并批量接收大小为 X(即 50)的消息，但如果流的流量较低，则在 Y 秒内(即 5 秒)给我流的任何内容。
因此，我不是逐个处理消息，而是处理列表大小为 50(或可能更小)的List[Record]。
这是为了使某些 I/O 绑定处理更有效率。
我知道这可以使用经典的 Kafka API 实现，但正在寻找一种基于流的实现，该实现也可以在本地处理偏移量提交，同时考虑到错误/故障。 我在文档中或通过搜索找不到任何相关内容，并且想知道是否有人可以解决此问题。
@Matthias J. Sax 的答案很好，我只想为此添加一个例子，我认为它可能对某人有用。 假设我们要将传入的值组合成以下类型：
public class MultipleValues { private List<String> values; }
要将消息收集到具有最大大小的批处理中，我们需要创建转换器：
public class MultipleValuesTransformer implements Transformer<String, String, KeyValue<String, MultipleValues>> {
private ProcessorContext processorContext;
private String stateStoreName;
private KeyValueStore<String, MultipleValues> keyValueStore;
private Cancellable scheduledPunctuator;
public MultipleValuesTransformer(String stateStoreName) {
this.stateStoreName = stateStoreName;
}
@Override
public void init(ProcessorContext processorContext) {
this.processorContext = processorContext;
this.keyValueStore = (KeyValueStore) processorContext.getStateStore(stateStoreName);
scheduledPunctuator = processorContext.schedule(Duration.ofSeconds(30), PunctuationType.WALL_CLOCK_TIME, this::doPunctuate);
}
@Override
public KeyValue<String, MultipleValues> transform(String key, String value) {
MultipleValues itemValueFromStore = keyValueStore.get(key);
if (isNull(itemValueFromStore)) {
itemValueFromStore = MultipleValues.builder().values(Collections.singletonList(value)).build();
} else {
List<String> values = new ArrayList<>(itemValueFromStore.getValues());
values.add(value);
itemValueFromStore = itemValueFromStore.toBuilder()
.values(values)
.build();
}
if (itemValueFromStore.getValues().size() >= 50) {
processorContext.forward(key, itemValueFromStore);
keyValueStore.put(key, null);
} else {
keyValueStore.put(key, itemValueFromStore);
}
return null;
}
private void doPunctuate(long timestamp) {
KeyValueIterator<String, MultipleValues> valuesIterator = keyValueStore.all();
while (valuesIterator.hasNext()) {
KeyValue<String, MultipleValues> keyValue = valuesIterator.next();
if (nonNull(keyValue.value)) {
processorContext.forward(keyValue.key, keyValue.value);
keyValueStore.put(keyValue.key, null);
}
}
}
@Override
public void close() {
scheduledPunctuator.cancel();
}
}
我们需要创建键值存储，将其添加到StreamsBuilder，并使用transform方法构建KStream流
Properties props = new Properties();
...
Serde<MultipleValues> multipleValuesSerge = Serdes.serdeFrom(new JsonSerializer<>(), new JsonDeserializer<>(MultipleValues.class));
StreamsBuilder builder = new StreamsBuilder();
String storeName = "multipleValuesStore";
KeyValueBytesStoreSupplier storeSupplier = Stores.persistentKeyValueStore(storeName);
StoreBuilder<KeyValueStore<String, MultipleValues>> storeBuilder =
Stores.keyValueStoreBuilder(storeSupplier, Serdes.String(), multipleValuesSerge);
builder.addStateStore(storeBuilder);
builder.stream("source", Consumed.with(Serdes.String(), Serdes.String()))
.transform(() -> new MultipleValuesTransformer(storeName), storeName)
.print(Printed.<String, MultipleValues>toSysOut().withLabel("transformedMultipleValues"));
KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), props);
kafkaStreams.start();
通过这种方法，我们使用了我们为其进行聚合的传入密钥。 如果您需要收集消息不是按键，而是按某些消息的字段收集消息，则需要以程来触发 KStream 上的重新平衡(通过使用中间主题)：
.selectKey(..)
.through(intermediateTopicName)
.transform( ..)
最简单的方法可能是使用有状态transform()操作。每次收到记录时，您都会将其放入商店。收到 50 条记录后，执行处理、发出输出并从存储中删除记录。
若要在一定时间内未读取限制时强制执行处理，可以注册挂钟标点符号。
似乎没有必要使用Processors 或Transformers 和transform()按计数批处理事件。常规groupBy()和reduce()/aggregate()应该可以解决问题：
KeyValueSerde keyValueSerde = new KeyValueSerde(); // simple custom Serde
final AtomicLong batchCount = new AtomicLong(0L);
myKStream
.groupBy((k,v) -> KeyValue.pair(k, batchCount.getAndIncrement() / batchSize),
Grouped.keySerde(keyValueSerde))
.reduce(this::windowReducer) // <-- how you want to aggregate values in batch
.toStream()
.filter((k,v) -> /* pass through full batches only */)
.selectKey((k,v) -> k.key)
...
您还需要为标准 KeyValue<String、Long> 添加简单的Serde。
此选项显然仅在您不需要"标点符号"在超时时发出不完整的批处理时才有用。在分布式处理的情况下，它也不保证批处理中元素的顺序。
您还可以将计数连接到键字符串以形成新键(而不是使用 KeyValue)。这将进一步简化示例(使用 Serdes.String())。