Получать сообщения, когда потребитель подключен к приложению spring-cloud-stream - PullRequest
Новая
       6

Получать сообщения, когда потребитель подключен к приложению spring-cloud-stream

0 голосов
/ 14 октября 2018

Я играю с 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 гарантирует доставку сообщений.

Ссылки на Githubhttps://github.com/govi20/producer-demo
https://github.com/govi20/consumer-demo

Ответы [ 2 ]

0 голосов
/ 14 октября 2018

Как я упоминал ранее, у вас уже есть неправильная конфигурация в 'myInput', так как у вас нет конфигурации @Input, которая приводит к ошибке A component required a bean named 'myInput' that could not be found. при запуске клиента.Таким образом, что-то подобное потребуется на стороне потребителя 

 interface MyProcessor {
    String INPUT = "myInput";

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

Более того, если вы не определите group, это приведет к анонимной очереди на стороне Кролика (что-то вроде 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, поэтому, если вы измените своего производителя для отправки в этот пункт назначения, вы получитесообщения на потребителя.

0 голосов
/ 14 октября 2018

Вам нужна группа для привязки ввода потребителя.В противном случае он является анонимным и связывает очередь автоматического удаления, которая существует только во время работы потребителя.

...