Kafka Producer 回调性能

Question

我有 Kafka Produce，它将消息发送到 kafka。我使用帮助存储过程在 onsucess 和 onFailure 中将消息记录在数据库中。如代码所示，我使用的是异步

1. 我应该将存储库中的 callStoredProcedure 方法标记为同步以避免死锁吗？我相信不需要同步，因为回调将在单个线程中按顺序执行。

2. 来自下面的链接

   https://kafka.apache.org/10/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html

请注意，回调通常会在 生产者，因此应该相当快，否则他们会延迟 从其他线程发送消息。如果你想执行 建议阻塞或计算昂贵的回调 在回调主体中使用您自己的 Executor 来并行化处理。

我应该在其他线程中执行回调吗？ 并且你可以分享代码sn-p如何在其他线程中执行回调。比如在 3 个线程中并行回调

我的代码sn-p

@Autowired
private Myrepository  myrepository;

public void sendMessageToKafka(List<String> message) {
    
            for (String s : message) {
    
                future = kafkaTemplate.send(topicName, message);
    
                future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
    
                    @Override
                    public void onSuccess(SendResult<String, String> result) {
    
                        System.out.println("Message Sent  " + result.getRecordMetadata().timestamp());
                        
                        myrepository.callStoredProcedure(result,"SUCCESS");
    
                    }
    
                    @Override
                    public void onFailure(Throwable ex) {
    
                        System.out.println(" sending failed  ");
                        myrepository.callStoredProcedure(result,"FAILED");
    
                    }
                });
    
            }

Answer 1

private final ExecutorService exec = Executors.newSingleThreadExecutor();


...

this.exec.submit(() -> myrepository.callStoredProcedure(result,"SUCCESS"));

任务仍将在单个线程上运行（但不是 Kafka IO 线程）。

如果它跟不上您的发布速度，您可能需要使用不同的执行器，例如缓存线程池执行器或 Spring 的 ThreadPoolTaskExecutor。