Spring Cloud Stream：如何在新的Consumer&lt；&gt；函数编程模型中使用@Transaction

问题描述

我有StreamListener，我想使用新的功能模型和Consumer<；>；替换它。遗憾的是，我不知道如何将@Transaction转换到新模式：

@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'

推荐答案

您是否尝试过

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