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

Question

我有一个StreamListener，我想用新的功能模型和消费者替换它

@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();
}