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

Question

我在玩春云流和RabbitMQ。

我有一个生成消息的RESTendpoint。

@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

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

Answer 2

正如我之前提到的，您在“myInput”中已经有了未命中配置，因为您没有@Input配置，这会导致组件需要一个名为“myInput”的bean，但找不到。消费者启动期间的错误。因此，在消费者端需要这样的东西

 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，因此如果您将生产者更改为发送到该目的地，您将收到有关消费者的消息。