RabbitMQ集成Webflux消费消息

雨点打透心脏的1/2处 2022-01-22 13:43 437阅读 0赞

pom.xml：

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>

application.yml：

spring:
      rabbitmq:
        host: localhost
        port: 5672
        username: guest
        password: guest

MessageListenerContainerFactory.java：

@Component
    public class MessageListenerContainerFactory { 
    
        private final ConnectionFactory connectionFactory;
    
        @Autowired
        public MessageListenerContainerFactory(ConnectionFactory connectionFactory) { 
            this.connectionFactory = connectionFactory;
        }
    
        public MessageListenerContainer createMessageListenerContainer(String queueName) { 
            SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
            container.addQueueNames(queueName);
            container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
            return container;
        }
    }

AmqpReactiveController.java：

@RestController
    public class AmqpReactiveController { 
    
        private final MessageListenerContainerFactory messageListenerContainerFactory;
    
        @Autowired
        public AmqpReactiveController(MessageListenerContainerFactory messageListenerContainerFactory) { 
            this.messageListenerContainerFactory = messageListenerContainerFactory;
        }
    
        @GetMapping(value = "listener", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
        public Flux<String> receiveMessagesFromQueue(@RequestParam String queueName) { 
            MessageListenerContainer container = messageListenerContainerFactory.createMessageListenerContainer(queueName);
            return Flux.create(emitter -> { 
                container.setupMessageListener((ChannelAwareMessageListener) (message, channle) -> { 
                    if (emitter.isCancelled()) { 
                        container.stop();
                        return;
                    }
                    String payload = new String(message.getBody());
                    System.out.println("收到来自" + queueName + "的消息：" + payload);
                    emitter.next(payload);
                    channle.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                });
                emitter.onRequest(v -> container.start());
                emitter.onDispose(() -> { 
                    container.stop();
                });
            });
        }
    }

调用方通过WebClient调用接口：

String baseUrl = "http://localhost:8080";
    WebClient webClient = WebClient.create(baseUrl);
    Flux<String> flux = webClient.get().uri("listener?queueName=queue_name1").retrieve().bodyToFlux(String.class);
    flux.subscribe(System.out::println);

参考链接：  
[https://www.baeldung.com/spring-amqp-reactive][https_www.baeldung.com_spring-amqp-reactive]

[https_www.baeldung.com_spring-amqp-reactive]: https://www.baeldung.com/spring-amqp-reactive