我有一个需要维护两个主题的需求,一个是同步方法,另一个是异步方法。异步方法在调用使用者记录时按预期工作,但是在同步方法中,不会调用使用者代码。
下面是配置文件中声明的代码
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9093");
props.put(ProducerConfig.RETRIES_CONFIG, 3);
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
我在这里启用了autoflush true
@Bean( name="KafkaPayloadSyncTemplate")
public KafkaTemplate<String, KafkaPayload> KafkaPayloadSyncTemplate() {
return new KafkaTemplate<String,KafkaPayload>(producerFactory(),true);
}
在返回recordmetadataresults对象之后,控件停止对使用者进行任何调用
private List<RecordMetadata> sendPayloadToKafkaTopicInSync() throws InterruptedException, ExecutionException {
final List<RecordMetadata> recordMetadataResults = new ArrayList<RecordMetadata>();
KafkaPayload kafkaPayload = constructKafkaPayload();
ListenableFuture<SendResult<String,KafkaPayload>>
future = KafkaPayloadSyncTemplate.send(TestTopic, kafkaPayload);
SendResult<String, KafkaPayload> results;
results = future.get();
recordMetadataResults.add(results.getRecordMetadata());
return recordMetadataResults;
}
消费者代码
public class KafkaTestListener {
@Autowired
TestServiceImpl TestServiceImpl;
public final CountDownLatch countDownLatch = new CountDownLatch(1);
@KafkaListener(id="POC", topics = "TestTopic", group = "TestGroup")
public void listen(ConsumerRecord<String,KafkaPayload> record, Acknowledgment acknowledgment) {
countDownLatch.countDown();
TestServiceImpl.consumeKafkaMessage(record);
System.out.println("Acknowledgment : " + acknowledgment);
acknowledgment.acknowledge();
}
}
基于这个问题,我有两个问题
当listener类是同步生产者时,我们是否应该在其内部手动调用listen()。如果是,怎么做?
如果听众( @KafkaListener
)自动调用,我需要添加哪些其他设置/配置才能使其正常工作。
感谢您提前提供的信息
-斯里坎特
1条答案
按热度按时间j9per5c41#
你应该确定你用的是
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
消费者财产。不确定您所说的sync/async是什么意思,但produce和consume是完全不同的操作。你不能从生产者的Angular 影响消费者。因为中间有Kafka经纪人。