Spring云流：StreamBridge和事务

Question

我希望有一个Spring云流侦听器处理有关其中发送的所有消息的完整事务。或者，即使之后有异常，也会提交函数中使用StreamBridge手动发送的所有消息。

这是我的lib版本：

spring : 2.5.5
spring cloud stream : 3.1.4
spring cloud stream rabbit binder : 3.1.4

我的Spring云流形态：

spring:
  cloud:
    function:
      definition: test
    stream:
      rabbit:
        default:
          producer:
            transacted: true
          consumer:
            transacted: true
        bindings:
          test-in-0:
            consumer:
              queueNameGroupOnly: true
              receive-timeout: 500
              transacted: true
          test-out-0:
            producer:
              queueNameGroupOnly: true
              transacted: true
          other-out-0:
            producer:
              queueNameGroupOnly: true
              transacted: true
      bindings:
        test-in-0:
          destination: test.request
          group: test.request
          consumer:
            requiredGroups: test.request
            maxAttempts: 1
        test-out-0:
          destination: test.response
          group:  test.response
          producer:
            requiredGroups:  test.response
        other-out-0:
          destination: test.other.request
          group: test.other.request
          producer:
            requiredGroups: test.other.request

我的测试java代码：

@Configuration
public class TestSender {
    @Bean
    public Function<Message<TestRequest>, Message<String>> test(Service service) {
        return (request) -> service.run(request.getPayload().getContent());
    }
}

@Component
@Transactional
public class Service {
    private static final Logger LOGGER = LoggerFactory.getLogger(Service.class);
    StreamBridge bridge;
    IWorker worker;
    public Service(StreamBridge bridge, IWorker worker) {
        this.bridge = bridge;
        this.worker = worker;
    }

    @Transactional
    public Message<String> run(String message) {
        LOGGER.info("Processing {}", message);
        bridge.send("other-out-0", MessageBuilder.withPayload("test")
                .setHeader("toto", "titi").build());
        if (message.equals("error")) {
            throw new RuntimeException("test error");
        }
        return MessageBuilder.withPayload("test")
                .setHeader("toto", "titi").build();
    }
}

要运行的测试类：

@SpringBootApplication
public class EmptyWorkerApplication {

    private static final Logger LOGGER = LoggerFactory.getLogger(EmptyWorkerApplication.class);

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

    @Bean
    public ApplicationRunner runner(RabbitTemplate template) {
        return args -> {
            LOGGER.info("Sending messages ...");
template.convertAndSend("test.request", "#",
                                org.springframework.amqp.core.MessageBuilder.withBody(
                                                "{\"content\":\"toto\"}".getBytes(StandardCharsets.UTF_8))
                                        .setContentType("application/json")
                                        .build());
        template.convertAndSend("test.request", "#",
                                org.springframework.amqp.core.MessageBuilder.withBody(
                                        "{\"content\":\"error\"}".getBytes(StandardCharsets.UTF_8))
                                        .setContentType("application/json")
                                        .build());
        template.convertAndSend("test.request", "#",
                                org.springframework.amqp.core.MessageBuilder.withBody(
                                        "{\"content\":\"titi\"}".getBytes(StandardCharsets.UTF_8))
                                        .setContentType("application/json")
                                        .build());
        };
    }

我还添加了TransactionManager：

@Configuration
@EnableTransactionManagement
public class TransactionManagerConfiguration {

    @Bean(name = "transactionManager")
    public RabbitTransactionManager rabbitTransactionManager(ConnectionFactory cf) {
        RabbitTransactionManager manager = new RabbitTransactionManager(cf);
        return manager;
    }

}

在最后的这个示例中，我的兔子队列中有：

或者我应该只有两条消息test.other.request.我做错了什么？

编辑1

试验代码：

@Component("myfunction")
public class Myfunction implements Consumer<String> {

    private final StreamBridge streamBridge;

    public Myfunction(StreamBridge streamBridge) {
        this.streamBridge = streamBridge;
    }

    @Override
    @Transactional
    public void accept(String request) {
        this.streamBridge.send("myfunction-out-0", request);
        if (request.equals("error")) {
            throw new RuntimeException("test error");
        }
    }
}

@SpringBootApplication
public class EmptyWorkerApplication {

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

    @Bean
    public RabbitTransactionManager rabbitTransactionManager(ConnectionFactory cf) {
        RabbitTransactionManager manager = new RabbitTransactionManager(cf);
        return manager;
    }

    @Bean
    public ApplicationRunner runner(RabbitTemplate template) {
        return args -> {
template.convertAndSend("test.request", "#",
                                org.springframework.amqp.core.MessageBuilder.withBody(
                                                "test".getBytes(StandardCharsets.UTF_8))
                                        .setContentType("text/plain")
                                        .build());
        template.convertAndSend("test.request", "#",
                                org.springframework.amqp.core.MessageBuilder.withBody(
                                        "error".getBytes(StandardCharsets.UTF_8))
                                        .setContentType("text/plain")
                                        .build());
        template.convertAndSend("test.request", "#",
                                org.springframework.amqp.core.MessageBuilder.withBody(
                                        "test".getBytes(StandardCharsets.UTF_8))
                                        .setContentType("text/plain")
                                        .build());
        };
    }
}

spring:
  rabbitmq:
    host: xx
    port: xx
    username: xx
    password: xx
    virtual-host: xx
  cloud:
    function:
      definition: myfunction
    stream:
      rabbit:
         bindings:
           myfunction-in-0:
             queueNameGroupOnly: true
           myfunction-out-0:
             queueNameGroupOnly: true
             transacted: true
      bindings:
        myfunction-in-0:
          destination: test.request
          group: test.request
          consumer:
            requiredGroups: test.request
            autoBindDlq: true
            maxAttempts: 1
        myfunction-out-0:
          destination: test.response
          group:  test.response
          producer:
            requiredGroups:  test.response

编辑2：

我终于成功了。我的错误是设置了属性spring。云流动兔子绑定。myfunction-in-0。消费者事务处理=true，而不是spring。云流动兔子绑定。中的myfunction。消费者交易=真

事实上，我不理解两者的区别，在spring cloud stream和spring cloud rabbit活页夹文档中也没有找到任何解释。

Answer 1

请看这个答案-Spring Cloud stream：如何使用@Transactional与新消费者