Rabbit MQ整合springBoot

一、pom依赖

        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!--spring整合MQ--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId></dependency>

二、消费端

2.1、application.properties 配置文件

server.port=8002
#上下文路径
server.servlet.context-path=/
spring.application.name=rabbit_consumer# MQ配置
spring.rabbitmq.addresses=192.168.220.3:5672
spring.rabbitmq.username=root
spring.rabbitmq.password=root
# 虚拟主机
spring.rabbitmq.virtual-host=/
# 连接超时 15秒
spring.rabbitmq.connection-timeout=15000
# 设置消费端消费成功消息后手动签收消息，默认auto自动签收
spring.rabbitmq.listener.simple.acknowledge-mode=manual
spring.rabbitmq.listener.simple.concurrency=6
# 最大消费线程数，并发数
spring.rabbitmq.listener.simple.max-concurrency=11
# prefetch为限制一次传送给消费者的消息数
spring.rabbitmq.listener.simple.prefetch=1# 自定义属性配置 MQ
spring.rabbitmq.listener.test.exchange=test_topic_exchange
spring.rabbitmq.listener.test.exchange.type=topic
spring.rabbitmq.listener.test.queue=test_topic1
spring.rabbitmq.listener.test.key=test_topic1.*

2.2、消费端核心组件

package com.xiao.component;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;
import java.io.IOException;@Component
public class RabbitMQReceived {@RabbitListener(bindings = @QueueBinding(exchange = @Exchange(name = "${spring.rabbitmq.listener.test.exchange}",type = "${spring.rabbitmq.listener.test.exchange.type}",durable = "true",ignoreDeclarationExceptions = "true"),value = @Queue(value = "${spring.rabbitmq.listener.test.queue}",durable = "true"),key = "${spring.rabbitmq.listener.test.key}"/*,admins = "root"*/))/*** 监听消息* @param message   消息* @param channel   通道*/@RabbitHandlerpublic void onMessage(Message message, Channel channel) throws IOException {System.err.println("=====================================");System.err.println("消费端 RabbitMQReceived 消费消息：" + message.getPayload());Long deliveryTag = (Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);//由于消费端配置手动消费消息后签收机制 spring.rabbitmq.listener.simple.acknowledge-mode=manual
//        channel.basicAck(deliveryTag,false);System.err.println("消费端 RabbitMQReceived ack：yes deliveryTag:" + deliveryTag);}
}

三、生产端

3.1、application.properties 配置文件

server.port=8001
#上下文路径
server.servlet.context-path=/
spring.application.name=rabbit_produce# MQ配置
spring.rabbitmq.addresses=192.168.220.3:5672
spring.rabbitmq.port=5672
spring.rabbitmq.username=root
spring.rabbitmq.password=root
# 虚拟主机
spring.rabbitmq.virtual-host=/
# 连接超时 15秒
spring.rabbitmq.connection-timeout=15000
# 开启produce发送给broker的消息确认模式，可靠性投递
spring.rabbitmq.publisher-confirms=true
#spring.rabbitmq.publisher-confirm-type=true  #有点问题
# 针对于broker未接收的消息return机制，需要结合mandatory一起使用
#spring.rabbitmq.template.mandatory=true
#spring.rabbitmq.publisher-returns=true

2.2、生产者 MQ消息发送组件

package com.xiao.component;import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Correlation;
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;
import java.util.UUID;@Component
public class RabbitMQSender {@Autowiredprivate RabbitTemplate rabbitTemplate;//生产者发送消息到broker确认回调接口private final RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {/*** @param correlationData   消息的唯一标识* @param ack broke         broker是否签收成功* @param cause             失败异常信息*/@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {String formatStr = String.format("生产端 confirmCallback 相关数据：%s，" +"broker签收情况 ack=%s，异常信息：%s" ,correlationData.toString(),ack,cause);System.err.println(formatStr);/*System.out.println("生产端 confirmCallback 相关数据：" + correlationData);System.out.println("生产端 confirmCallback broker签收情况：" + ack);System.out.println("生产端 confirmCallback 异常信息：" + cause);*/}};/*** 发送消息* @param message       消息* @param properties    消息对应的属性，如时间*/public void send(Object message, Map<String,Object> properties) {MessageHeaders messageHeaders = new MessageHeaders(properties);Message<?> msg = MessageBuilder.createMessage(message, messageHeaders);rabbitTemplate.setConfirmCallback(confirmCallback);//消息发送完后置处理器MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {@Overridepublic org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message) throws AmqpException {System.err.println("生产端 RabbitMQSender send后置处理：" + message);return message;}@Overridepublic org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message, Correlation correlation) {System.err.println("生产端 RabbitMQSender send后置处理：" + message+" 消息标识：" + correlation);return message;}};//消息唯一属性CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());rabbitTemplate.convertAndSend("test_topic_exchange",//exchange,"test_topic1.xiao",// routingKey,msg, //message,messagePostProcessor,correlationData);}
}

四、测试

package com.xiao;import com.xiao.component.RabbitMQSender;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import java.util.HashMap;
import java.util.Map;@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest
public class SendMessageTest {@Autowiredprivate RabbitMQSender rabbitMQSender;@Testpublic void send() throws InterruptedException {Map<String,Object> properties = new HashMap<>(2);properties.put("userName","xiao");rabbitMQSender.send("hello world!",properties);Thread.sleep(5000);//10秒}
}

1、生产端控制台

生产端 RabbitMQSender send后置处理：(Body:'[B@3a6045c6(byte[535])' MessageProperties [headers={}, contentType=application/x-java-serialized-object, contentLength=535, deliveryMode=PERSISTENT, priority=0, deliveryTag=0]) 消息标识：CorrelationData [id=8c78e89d-80f3-4f3d-ba8b-13e863c6295c]
2023-07-21 20:05:37.611  INFO 4536 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Attempting to connect to: [192.168.220.3:5672]
2023-07-21 20:05:37.653  INFO 4536 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory#6f38a289:0/SimpleConnection@6215366a [delegate=amqp://root@192.168.220.3:5672/, localPort= 4712]
生产端 confirmCallback 相关数据：CorrelationData [id=8c78e89d-80f3-4f3d-ba8b-13e863c6295c]，broker签收情况 ack=true，异常信息：null

2、消费端控制台

=====================================
消费端 RabbitMQReceived 消费消息：hello world!
消费端 RabbitMQReceived ack：yes deliveryTag:1