Reactor Kafka:恰好一次处理样本 [英] Reactor Kafka: Exactly Once Processing Sample
本文介绍了Reactor Kafka:恰好一次处理样本的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!
问题描述
我读过很多文章,其中有许多不同的配置可以实现一次处理.
I've read many articles where there are many different configurations to achieve exactly once processing.
这是我的生产者配置:
final Map<String, Object> props = Maps.newConcurrentMap();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "producer-tx-1");
这是我的消费者配置:
final Map<String, Object> props = Maps.newHashMap();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
我读过这个[示例场景][1]
I read this [sample scenarios][1]
我尝试遵循但我遇到了一些问题:
And I try to follow but i'm getting some issues:
这是我的生产者代码:
@Override
public Mono<SenderResult<Void>> buy(Message msg) {
final ReactiveKafkaProducerTemplate kafkaProducerTemplate = kafkaConfig.getKafkaProducerTemplate();
return kafkaProducerTemplate.transactionManager().begin().then(kafkaProducerTemplate.send(mytopic, msg));
}
我的消费者代码:
@Override
public void run(ApplicationArguments arg0) throws Exception {
final ReactiveKafkaProducerTemplate kafkaProducerTemplate = kafkaConfig.getKafkaProducerTemplate();
final ReactiveKafkaConsumerTemplate kafkaConsumerTemplate = kafkaConfig.getKafkaConsumerTemplate(mytopic, Message.class);
final Flux<ConsumerRecord<String, Message>> flux = kafkaConsumerTemplate.receiveExactlyOnce(kafkaProducerTemplate.transactionManager())
.concatMap(receiverRecordFlux -> receiverRecordFlux );
flux.subscribe(record -> {
final Message message = record.value();
System.out.printf("received message: timestamp=%s key=%d value=%s\n",
dateFormat.format(new Date(record.timestamp())),
record.key(),
message);
transactionService.processAndSendToNextTopic(message)
.doOnSuccess(aVoid -> kafkaProducerTemplate.transactionManager().commit())
.subscribe();
});
}
我在尝试生成和使用消息时总是遇到以下错误:
I'm always getting following error when trying to produce and consume message:
Invalid transition attempted from state IN_TRANSACTION to state IN_TRANSACTION
推荐答案
查看 receiveExactlyOnce
/**
* Returns a {@link Flux} of consumer record batches that may be used for exactly once
* delivery semantics. A new transaction is started for each inner Flux and it is the
* responsibility of the consuming application to commit or abort the transaction
* using {@link TransactionManager#commit()} or {@link TransactionManager#abort()}
* after processing the Flux.
begin()
已经被调用,所以你不需要调用它.
begin()
has already been called so you don't need to call it.
@Override
public Flux<Flux<ConsumerRecord<K, V>>> receiveExactlyOnce(TransactionManager transactionManager) {
this.ackMode = AckMode.EXACTLY_ONCE;
Flux<ConsumerRecords<K, V>> flux = withDoOnRequest(createConsumerFlux());
return flux.map(consumerRecords -> transactionManager.begin()
.then(Mono.fromCallable(() -> awaitingTransaction.getAndSet(true)))
.thenMany(transactionalRecords(transactionManager, consumerRecords)))
.publishOn(transactionManager.scheduler());
}
这篇关于Reactor Kafka:恰好一次处理样本的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!
查看全文