Spring Boot+RabbitMQ 通過fanout模式實作訊息接收（支持消費者多實體部署）-有解無憂

　　本文章適用的場景：同一條訊息可以被多個消費者同時消費，注意：當消費者多實體部署時，會輪詢消費訊息，網上有大量的的案例展示：P生產一條訊息，消費者服務C中建立Q1和Q2兩個佇列共同消費，但極少的材料展示：P生產一條訊息后M1，消費者C1和C2可以同時消費M1，如下圖所示，案例基于Spring Boot以及RabbitMQ的“fanout”型別exchange，已經實測可放心使用，

1、引入基本依賴，專案不同請您按自己的情況引入合適的依賴

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

2、rabbitmq密碼、用戶名等配置

　　內容過于基礎，請網路上自行搜索相關內容，

3、Exchange配置

    @Bean
    public FanoutExchange myFanoutExchange() {
    　　 //“my.fanout”僅用于演示，請按自己的專案情況選擇合適的、有意義的命名
        return new FanoutExchange("my.fanout", true, false);
    }
    
    @Bean
    public Queue anonymousQueue() {
        //注意：必須使用匿名佇列才能達到本文目標，此匿名佇列實體的名稱在消費端會被使用
        return new AnonymousQueue();
    }
    
    @Bean
    public Binding bindAnonymousQueue() {
        return BindingBuilder.bind(anonymousQueue()).to(myFanoutExchange());
    }

4、發送端代碼

        //構建回呼回傳的資料
        CorrelationData correlationData = https://www.cnblogs.com/skevin/p/new CorrelationData(UUIDUtils.getUUID());

        Message msg = MessageBuilder.withBody(message.getBytes())
                .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
                .setCorrelationId(correlationData.getId())
                .build();
        try {
            rabbitTemplate.convertAndSend(exchangeName, null, msg, correlationData);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }

注意：“fanout"模式直接將訊息發送到“exchange”，routing key應為null或“”，如果您的代碼只需要發送訊息而無接收，請忽略第3小節“Exchange配置”中所示代碼

5、消費者代碼

@Service
public class MyListener {
    @RabbitListener(queues = "#{anonymousQueue.name}")
    public void receive(String message, Message message1, Channel channel) {
        try {
            //your code
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            long deliverTag = message1.getMessageProperties().getDeliveryTag();
            channel.basicAck(deliverTag, false);
        }
    }
}

　　代碼“@RabbitListener…”中的“#{anonymousQueue.name}”中的“anonymousQueue”對應第3小節“Exchange配置”中所宣告的匿名佇列實體，

　　~本文完~

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/436312.html

標籤：Java

上一篇：9-Java泛型

下一篇：Spring系列22：Spring AOP 概念與快速入門篇