RabbitMQ 3.9.7 鏡像模式集群與Springboot 2.5.5 整合

更新時(shí)間：2021年10月02日 12:27:41 作者：追風(fēng)人聊Java

今天我們來(lái)聊聊 RabbitMQ 3.9.7 鏡像模式集群與Springboot 2.5.5 整合，本文通過(guò)實(shí)例代碼給大家介紹的非常詳細(xì)，需要的朋友參考下吧

1. 概述

老話說(shuō)的好：做人要懂得變通，善于思考，有時(shí)稍微轉(zhuǎn)個(gè)彎，也許問(wèn)題就解決了。

言歸正傳，之前我們聊了 RabbitMQ 3.9.7 鏡像模式集群的搭建，今天我們來(lái)聊聊 RabbitMQ 3.9.7 鏡像模式集群與Springboot 2.5.5 整合。

2. 場(chǎng)景說(shuō)明

服務(wù)器A IP：192.168.1.22

服務(wù)器B IP：192.168.1.8

服務(wù)器C IP：192.168.1.144

此三臺(tái)服務(wù)器上已搭建好了 RabbitMQ鏡像模式集群，鏡像模式集群的搭建，可參見(jiàn)我的上一篇文章。

3. 與Springboot的整合

3.1 引入依賴(lài)

<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.5.5</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

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

3.2 生產(chǎn)服務(wù)配置

spring:
  rabbitmq:
    addresses: 192.168.1.22:5672,192.168.1.8:5672,192.168.1.144:5672
    username: guest
    password: guest
    virtual-host: /
    connection-timeout: 16000

    # 啟用消息確認(rèn)模式
    publisher-confirm-type: correlated

    # 啟用 return 消息模式
    publisher-returns: true
    template:
      mandatory: true

3.3 生產(chǎn)服務(wù)代碼

import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
public class Producer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 確認(rèn)回調(diào)
     */
    final RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {

        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            // correlationData 唯一標(biāo)識(shí)
            // ack mq是否收到消息
            // cause 失敗原因
            System.out.println("correlationData：" + correlationData.getId());
            System.out.println("ack：" + ack);
            System.out.println("cause：" + cause);

        }
    };

    /**
     * 發(fā)送消息
     * @param messageBody   消息體
     * @param headers       附加屬性
     * @throws Exception
     */
    public void sendMessage(String messageBody, Map<String, Object> headers, String id) throws Exception {

        MessageHeaders messageHeaders = new MessageHeaders(headers);

        Message<String> message = MessageBuilder.createMessage(messageBody, messageHeaders);

        rabbitTemplate.setConfirmCallback(confirmCallback);

        String exchangeName = "exchange-hello";
        String routingKey = "test.123";

        CorrelationData correlationData = new CorrelationData(id);

        rabbitTemplate.convertAndSend(exchangeName, routingKey, message, new MessagePostProcessor() {

            /**
             * 發(fā)送消息后做的事情
             * @param message
             * @return
             * @throws AmqpException
             */
            @Override
            public org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message) throws AmqpException {
                return message;
            }
        }, correlationData);
    }
}

3.4 消費(fèi)服務(wù)配置

spring:
  rabbitmq:
    addresses: 192.168.1.22:5672,192.168.1.8:5672,192.168.1.144:5672
    username: guest
    password: guest
    virtual-host: /
    connection-timeout: 16000

    listener:
      simple:
        # 設(shè)置為手工ACK
        acknowledge-mode: manual
        concurrency: 5
        prefetch: 1
        max-concurrency: 10

3.5 消費(fèi)服務(wù)代碼

import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

@Component
public class Consumer {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "queue-hello", durable = "true"),
            exchange = @Exchange(value = "exchange-hello" , durable = "true", type = "topic"),
            key = "test.*"
    ))
    @RabbitHandler
    public void onMessage(Message message, Channel channel) throws Exception {

        System.out.println("收到消息：" + message.getPayload());

        Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
        channel.basicAck(deliveryTag, false);
    }
}

3.6 Rest 測(cè)試代碼

@RestController
@RequestMapping("/mq")
public class RabbitmqController {

    @Autowired
    private Producer producer;

    @GetMapping("/sendMessage")
    public String sendMessage(@RequestParam String messageBody, @RequestParam String id) throws Exception {
        Map<String, Object> headers = new HashMap<>();
        producer.sendMessage(messageBody, headers, id);
        return "success";
    }
}