Spring Cloud Stream：如何将@Transactional与新的消费者<>函数式编程模型一起使用

Question

我有StreamListener，我想使用新的功能模型和Consumer <>来替换它。不幸的是，我不知道如何将@Transactional转移到新模型:

@Transactional
@StreamListener(PaymentChannels.PENDING_PAYMENTS_INPUT)
public void executePayments(PendingPaymentEvent event) throws Exception {
paymentsService.triggerInvoicePayment(event.getInvoiceId());
}

我厌倦了某些事情。下面是示例代码。我将日志记录消息添加到另一个测试队列中。然后抛出一个异常来触发回滚。不幸的是，即使在方法完成之前消息并不存在，消息也会排队(我使用断点进行了测试)。尽管出现了错误，但似乎事务还是自动提交了。

@Transactional
@RequiredArgsConstructor
@Component
public class functionalPayment implements Consumer<PendingPaymentEvent> {
private final PaymentsService paymentsService;
private final StreamBridge streamBridge;
public void accept(PendingPaymentEvent event) {
paymentsService.triggerInvoicePayment(event.getInvoiceId());
streamBridge.send("log-out-0",event);
throw new RuntimeException("Test exception to rollback message from log-out-0");
}
}

配置:

spring.cloud.stream.rabbit.bindings.functionalPayment-in-0.consumer.queue-name-group-only=true
spring.cloud.stream.rabbit.bindings.functionalPayment-in-0.consumer.declare-exchange=true
spring.cloud.stream.rabbit.bindings.functionalPayment-in-0.consumer.bind-queue=true
spring.cloud.stream.rabbit.bindings.functionalPayment-in-0.consumer.transacted=true
spring.cloud.stream.source=log
spring.cloud.stream.bindings.log-out-0.content-type=application/json
spring.cloud.stream.bindings.log-out-0.destination=log_a
spring.cloud.stream.bindings.log-out-0.group=log_a
spring.cloud.stream.rabbit.bindings.log-out-0.producer.declare-exchange=true
spring.cloud.stream.rabbit.bindings.log-out-0.producer.bind-queue=true
spring.cloud.stream.rabbit.bindings.log-out-0.producer.queue-name-group-only=true
spring.cloud.stream.rabbit.bindings.log-out-0.producer.binding-routing-key=log
spring.cloud.stream.rabbit.bindings.log-out-0.producer.transacted=true
spring.cloud.stream.rabbit.bindings.log-out-0.producer.exchange-type=direct
spring.cloud.stream.rabbit.bindings.log-out-0.producer.routing-key-expression='log'

Answer 1

您是否尝试过

@Transactional
public class ExecutePaymentConsumer implements Consumer<PendingPaymentEvent> {
public void accept(PendingPaymentEvent event) {
paymentsService.triggerInvoicePayment(event.getInvoiceId());
}
}
. . .
@Bean
public ExecutePaymentConsumer executePayments() {
return new ExecutePaymentConsumer();
}