我想同时使用Spring Kafka的ReplyingKafkaTemplate和Kafka事务。响应应用程序的代码如下所示:

@SpringBootApplication(scanBasePackages= {"com.example.myapp"})
public class ResponseApplication {
    private static final Logger LOGGER = LoggerFactory.getLogger(ResponseApplication.class);

    public static void main(String[] args) {
        SpringApplication.run(ResponseApplication.class, args);
    }

    @KafkaListener(id="${myapp.kafka.groupId}", topics="kRequests1")
    @SendTo(value= {"kReplies1"})
    @Transactional
    public String listen(String in) {
        return in;
    }

    @Bean
    public NewTopic kRequests1() {
        return new NewTopic("kRequests1", 2, (short) 1);
    }
}


配置类如下所示:

@Configuration
@EnableKafka
@EnableTransactionManagement
public class KafkaConfig {

    @Value( "${myapp.kafka.groupId}" )
    private String groupId;

    @Bean
    public KafkaTransactionManager<String, String> KafkaTransactionManager(
            ProducerFactory<String, String> producerFactory) {
        return new KafkaTransactionManager<String, String>(producerFactory);
    }

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        DefaultKafkaProducerFactory<String, String> producerFactory = new DefaultKafkaProducerFactory<>(producerConfigs());
        producerFactory.setTransactionIdPrefix("myapp");
        return producerFactory;
    }

    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, System.getProperty("kafka.host", "localhost") + ":9092");
        props.put(ProducerConfig.RETRIES_CONFIG, 0);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        props.put(ProducerConfig.RETRIES_CONFIG, 1);
        return props;
    }

    @Bean
    public ContainerProperties containerProperties(KafkaTransactionManager<String, String> kafkaTransactionManager) {
        ContainerProperties containerProperties = new ContainerProperties("kRequests1");
        containerProperties.setTransactionManager(kafkaTransactionManager);
        containerProperties.setGroupId(groupId);
        return containerProperties;
    }

    @Bean("kafkaListenerContainerFactory")
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>>
                        kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.getContainerProperties().setPollTimeout(3000);
        KafkaTemplate<?, ?> replyTemplate = new KafkaTemplate<>(producerFactory());
        factory.setReplyTemplate(replyTemplate);
        return factory;
    }

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, System.getProperty("kafka.host", "localhost") + ":9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "iprs");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> producerFactory) {
        return new KafkaTemplate<>(producerFactory);
    }
}


每次应用程序收到消息时,都会导致以下异常:

java.lang.IllegalStateException: No transaction is in process


Spring容器是否不应该启动事务,因为该方法用@Transactional注释,启用了事务管理并且ProducerFactory具有事务ID前缀?

最佳答案

我看到您正在使用回复模板;在任何情况下,该模板上的send都将超出@Transactional的范围。

在此用例中,不应使用@Transactional;将事务管理器注入到侦听器容器工厂中,以便在侦听器退出时容器可以将偏移量发送到事务。

08-15 19:23
查看更多