每个 Http 请求的多个 Kafka 生产者实例

Question

我有一个可以同时被多个用户调用的休息端点。这个休息端点调用一个事务性 Kafka 生产者。我的理解是，如果我们使用 Transaction，我不能同时使用同一个 Kafka Producer 实例。

如何高效地为每个 HTTP 请求创建一个新的 Kafka 生产者实例？

//Kafka Transaction enabled
producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
producerProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "prod-1-" );

@Service
public class ProducerService {
    @Autowired
    private KafkaTemplate<Object, Object> kafkaTemplate;

    public void postMessage(final MyUser message) {
        // wrapping the send method in a transaction
        this.kafkaTemplate.executeInTransaction(kafkaTemplate -> {
              kafkaTemplate.send("custom", null, message);
        }

    }

Answer 1

请参阅 DefaultKafkaProducerFactory 的 javadoc。它为生产者发起的交易维护生产者缓存。

/**
 * The {@link ProducerFactory} implementation for a {@code singleton} shared {@link Producer} instance.
 * <p>
 * This implementation will return the same {@link Producer} instance (if transactions are
 * not enabled) for the provided {@link Map} {@code configs} and optional {@link Serializer}
 * implementations on each {@link #createProducer()} invocation.

...

 * Setting {@link #setTransactionIdPrefix(String)} enables transactions; in which case, a
 * cache of producers is maintained; closing a producer returns it to the cache. The
 * producers are closed and the cache is cleared when the factory is destroyed, the
 * application context stopped, or the {@link #reset()} method is called.

...

 */

每个 Http 请求的多个 Kafka 生产者实例

Multiple Kafka Producer Instance for each Http Request

spring-boot

apache-kafka

spring-kafka

kafka-producer-api