RabbitMQ 3.9.7 镜像模式集群与Springboot 2.5.5 整合

<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>

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

     # 启用消息确认模式

     publisher-confirm-type: correlated

     # 启用 return 消息模式

     publisher-returns: true

     template:

       mandatory: true

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;

     /**

      * 确认回调

      */

     final RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {

         @Override

         public void confirm(CorrelationData correlationData, boolean ack, String cause) {

             // correlationData 唯一标识

             // ack mq是否收到消息

             // cause 失败原因

             System.out.println( "correlationData：" + correlationData.getId());

             System.out.println( "ack：" + ack);

             System.out.println( "cause：" + cause);

         }

     };

     /**

      * 发送消息

      * @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() {

             /**

              * 发送消息后做的事情

              * @param message

              * @return

              * @throws AmqpException

              */

             @Override

             public org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message) throws AmqpException {

                 return message;

             }

         }, correlationData);

     }

}

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:

         # 设置为手工ACK

         acknowledge-mode: manual

         concurrency: 5

         prefetch: 1

         max-concurrency: 10

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 );

     }

}

@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" ;

     }

}