当消费者在 spring-cloud-stream 应用程序中启动时接收消息

Question

我正在玩 Spring-cloud-stream 和 RabbitMQ。

我有一个产生消息的 REST 端点。

@SpringBootApplication
@EnableBinding(MyProcessor.class)
public class ProducerDemoApplication {

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

@RestController
class ProducerController {

    @Autowired
    MyProcessor myProcessor;

    @RequestMapping(value = "sendmessage/{message}", method = RequestMethod.GET)
    public String sendMessage(@PathVariable("message") String message) {
        myProcessor.anOutput().send(MessageBuilder.withPayload(message).build());
        return "sent";  
    }

}

interface MyProcessor {
    String INPUT = "myInput";

    @Output("myOutput")
    MessageChannel anOutput();
}

通过另一个应用程序，我正在使用这些消息。

@StreamListener(MyProcessor.INPUT)
public void eventHandler(String message) {
    System.out.println("**************  Message received => "+message);
}

当两个应用程序都已启动并正在运行时。我能够发布消息并在消费者处使用它。

我在以下场景中面临的问题：

我故意让消费者失望并通过生产者发布消息。现在，当消费者启动时，它没有收到任何消息。

我想 RabbitMQ 保证消息传递。

Github 链接
https://github.com/govi20/producer-demo
https://github.com/govi20/consumer-demo

Answer 1

正如我之前提到的，您已经在“myInput”中配置了错误配置，因为您没有 @Input 配置，这会导致消费者启动期间出现 A component required a bean named 'myInput' that could not be found. 错误。 因此，在消费者方面需要这样的东西

 interface MyProcessor {
    String INPUT = "myInput";

    @Input("myInput")
    MessageChannel myInput();
}

如果您没有定义group，则更多信息会导致 Rabbit 端出现匿名队列（类似于 myInput.anonymous.pZg03h0zQ2-SHLh1_QL8DQ），这实际上会导致每次启动时队列的名称都不同，所以

spring.cloud.stream.bindings.myInput.destination=myInput
spring.cloud.stream.bindings.myInput.group=myGroup

将导致队列名称myInput.myGroup 在启动之间保持一致。

另外，在生产者端myOutput 会导致创建一个没有路由到上述（或任何其他）队列的 Rabbit Exchange，因此 Rabbit 会丢弃消息，因此您不可能接收来自生产者，直到您在 myOutput 交换和 myInput.myGroup 队列之间创建路由。 但是，如果您按照我上面描述的方式配置输入，spring-cloud-stream 还将创建一个名为 myInput 的交换，该交换将自动路由到 myInput.myGroup，因此如果您将生产者更改为发送到该目的地，您将收到消费者的消息。

Answer 2

您需要一个关于消费者输入绑定的组。否则它是匿名的并绑定一个自动删除队列，该队列仅在消费者运行时存在。