RabbitMQ 是采用 Erlang 语言实现 AMQP（Advanced Message Queuing Protocol，高级消息队列协议）的消息中间件，用于在分布式系统中存储转发消息。RabbitMQ 凭借其高可靠、易扩展、高可用及丰富的功能特性受到越来越多企业的青睐。
实例要求：
消息发送端实现消息的发送、消息确认功能（交换器确认、队列确认）。
创建第一个 SpringBoot 项目（ RabbitmqProvider 消息发送项目），并整合 RabbitMQ 框架。项目结构如下图：
在pom.xml配置信息文件中，添加相关依赖文件：
org.springframework.boot
spring-boot-starter-amqp
在 application.yml 配置文件中配置 RabbitMQ 服务：
spring:
# 项目名称
application:
name: rabbitmq-provider
# RabbitMQ服务配置
rabbitmq:
host: 127.0.0.1
port: 5672
username: guest
password: guest
# 消息确认（ACK）
publisher-confirm-type: correlated #确认消息已发送到交换机(Exchange)
publisher-returns: true #确认消息已发送到队列(Queue)在com.rabbitmq.rabbitmqprovider.config包中，创建RabbitMqConfig类（RabbitMQ配置类），代码如下：
package com.rabbitmq.rabbitmqprovider.config;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
/**
* RabbitMQ配置类
* @author luoxiang
* @date 2022年01月12日 9:35 上午
*/
@Configuration
public class RabbitMqConfig {
public static final String DIRECT_QUEUE = "direct_queue"; //Direct队列名称
public static final String DIRECT_EXCHANGE = "direct_exchange"; //交换器名称
public static final String DIRECT_ROUTING_KEY = "direct_routing_key"; //路由键
public static final String DELAY_QUEUE = "delay_queue"; // 延时队列名称
public static final String DELAY_EXCHANGE = "delay_exchange"; //交换器名称
public static final String DELAY_ROUTING_KEY = "delay_routing_key"; //路由键
@Autowired
private CachingConnectionFactory connectionFactory;
public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){
RabbitTemplate rabbitTemplate = new RabbitTemplate();
rabbitTemplate.setConnectionFactory(connectionFactory);
//设置Json转换器
rabbitTemplate.setMessageConverter(jsonMessageConverter());
//设置开启Mandatory，才能触发回调函数，无论消息推送结果怎么样都强制调用回调函数
rabbitTemplate.setMandatory(true);
//确认消息送到交换器（Exchange）回调
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
System.out.println("
确认消息发送到交换机(Exchange)结果：");
System.out.println("相关数据："+correlationData);
System.out.println("是否成功："+ack);
System.out.println("错误原因："+cause);
});
//确认消息送到队列(Queue)回调
rabbitTemplate.setReturnsCallback(returnedMessage -> {
System.out.println("
确认消息送到队列(Queue)结果：");
System.out.println("发生消息:"+returnedMessage.getMessage());
System.out.println("回应码："+returnedMessage.getReplyCode());
System.out.println("回应消息："+returnedMessage.getReplyText());
System.out.println("路由键："+returnedMessage.getRoutingKey());
});
return rabbitTemplate;
}
/**
* Json转换器
* @author luoxiang
* @date 2022/1/12 10:00 上午
* @return org.springframework.amqp.support.converter.Jackson2JsonMessageConverter
*/
@Bean
public Jackson2JsonMessageConverter jsonMessageConverter() {
return new Jackson2JsonMessageConverter();
}
/**
* Direct交换器
* @author luoxiang
* @date 2022/1/12 10:01 上午
* @return org.springframework.amqp.core.DirectExchange
*/
@Bean
public DirectExchange directExchange(){
/**
* 创建交换器，参数说明：
* String name：交换器名称
* boolean durable：设置时候持久化，默认是false。durable设置为true表示持久
* boolean autoDelete：设置是否自动删除，为true则设置队列为自动删除
*/
return new DirectExchange(DIRECT_EXCHANGE,true,false);
}
/**
* 队列
* @author luoxiang
* @date 2022/1/12 10:06 上午
* @return org.springframework.amqp.core.Queue
*/
@Bean
public Queue directQueue(){
/**
* 创建队列，参数说明：
* String name：队列名称
* boolean durable：是否持久化，默认是false。durable设置为true表示持久
* boolean exclusive:设置是否排他，默认是false。为true则设置队列为排他
* boolean autoDelete：设置是否自动删除，为true则设置队列为自动删除，当没有生产者或者消费者使用队列，该队列会自动删除
* Map arguments：设置队列的其他一些参数
*/
return new Queue(DIRECT_QUEUE,true,false,false,null);
}
/**
* 绑定
* @author luoxiang
* @date 2022/1/12 10:16 上午
* @param directExchange
* @param directQueue
* @return org.springframework.amqp.core.Binding
*/
@Bean
public Binding directBinding(DirectExchange directExchange,Queue directQueue){
//将队列和交换机绑定，并设置用于匹配键：routingKey路由键
return BindingBuilder.bind(directQueue).to(directExchange).with(DIRECT_ROUTING_KEY);
}
/*******************************延时队列**********************************/
@Bean
public CustomExchange delayExchange(){
HashMap args = new HashMap<>();
args.put("x-delayed-type","direct");
return new CustomExchange(DELAY_EXCHANGE,"x-delayed-message",true,false);
}
@Bean
public Queue delayQueue(){
Queue queue = new Queue(DELAY_QUEUE, true);
return queue;
}
@Bean
public Binding delayBinding(Queue delayQueue,CustomExchange delayExchange){
return BindingBuilder.bind(delayQueue).to(delayExchange).with(DELAY_ROUTING_KEY).noargs();
}
} 在com.rabbitmq.rabbitmqprovider.entity包中，创建UserInfo(用户信息实体类)。
package com.rabbitmq.rabbitmqprovider.entity;
import lombok.Data;
/**
* 用户信息实体类
* @author luoxiang
* @date 2022年01月12日 10:25 上午
*/
@Data
public class UserInfo {
private int userId;//用户编号
private String userName;//用户姓名
private String blogUrl;//博客地址
private String blogRemark;//博客信息
}在com.rabbitmq.rabbitmqprovider.entity.UserInfo包下，创建UserSender接口（用户消息发送服务接口）。
package com.rabbitmq.rabbitmqprovider.sender;
import com.rabbitmq.rabbitmqprovider.entity.UserInfo;
import java.util.Map;
/**
* 用户消息发送服务接口
* @author luoxiang
* @date 2022年01月12日 10:29 上午
*/
public interface UserSender {
/**
* 发送用户信息Json格式数据
* @author luoxiang
* @date 2022/1/12 10:31 上午
* @param userInfo 用户信息实体类
*/
public void sendUserJson(UserInfo userInfo);
/**
* 延时发送用户信息Map格式数据
* @author luoxiang
* @date 2022/1/12 10:31 上午
* @param userMap 用户信息map
*/
public void sendDelayUserMap(Map userMap);
}在com.rabbitmq.rabbitmqprovider.sender.impl包下，创建UserSenderImpl类（用户消息发送服务类）。
package com.rabbitmq.rabbitmqprovider.sender.impl;
import com.rabbitmq.rabbitmqprovider.config.RabbitMqConfig;
import com.rabbitmq.rabbitmqprovider.entity.UserInfo;
import com.rabbitmq.rabbitmqprovider.sender.UserSender;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
/**
* 用户消息发送服务类
* @author luoxiang
* @date 2022年01月12日 10:33 上午
*/
@Service
public class UserSenderImpl implements UserSender {
@Autowired
private RabbitTemplate rabbitTemplate;
//时间格式
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
/**
* 发送用户信息Json格式数据
* @author luoxiang
* @date 2022/1/12 10:38 上午
* @param userInfo 用户信息实体类
*/
@Override
public void sendUserJson(UserInfo userInfo) {
/**
* 发送消息，参数说明：
* String exchange：交换器名称
* String routingKey：路由键
* Object object：发送内容
*/
this.rabbitTemplate.convertAndSend(RabbitMqConfig.DIRECT_EXCHANGE,RabbitMqConfig.DIRECT_ROUTING_KEY,userInfo);
System.out.println("Json格式数据消息发送成功，发送时间："+dateFormat.format(new Date()));
}
/**
* 延时发送用户信息Map格式数据
* @author luoxiang
* @date 2022/1/12 10:39 上午
* @param userMap 用户信息Map
*/
@Override
public void sendDelayUserMap(Map userMap) {
this.rabbitTemplate.convertAndSend(RabbitMqConfig.DELAY_EXCHANGE, RabbitMqConfig.DELAY_ROUTING_KEY, userMap, message -> {
message.getMessageProperties().setHeader("x-delay",5000);
return message;
});
System.out.println("Map格式数据消息发送成功。发送时间："+dateFormat.format(new Date()));
}
}消息接收端实现消息的接收、消息接收确认功能。
创建第二个SpringBoot项目（RabbitmqConsumer消息接收项目），并整合RabbitMQ框架。项目结构如下图：
在pom.xml配置信息文件中，添加相关依赖文件：
org.springframework.boot
spring-boot-starter-amqp
在application.yml配置文件中配置RabbitMQ服务：
spring:
#项目名称
application:
name: rabbitmq-consumer
#RabbitMQ服务配置
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
listener:
simple:
#重试机制
retry:
enabled: true #是否开启消费者重试
max-attempts: 5 #最大重试次数
initial-interval: 5000ms #重试间隔时间（单位毫秒）
max-interval: 1200000ms #重试最大时间间隔（单位毫秒）
multiplier: 2 #间隔时间乘子，间隔时间*乘子=下一次的时间间隔，最大不能超过设置的最大时间间隔在com.rabbitmq.rabbitmqconsumer.config包中，创建RabbitMqConfig类（RabbltMQ配置类），代码如下：
package com.rabbitmq.rabbitmqconsumer.config;
import com.rabbitmq.rabbitmqconsumer.receiver.impl.AckReceiver;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* RabbitMQ配置类
* @author luoxiang
* @date 2022年01月12日 10:55 上午
*/
@Configuration
public class RabbitMqConfig {
public static final String DIRECT_QUEUE = "direct_queue"; //Direct队列名称
public static final String DELAY_QUEUE = "delay_queue"; // 延时队列名称
/**
* 消息接收确认处理类
*/
@Autowired
private AckReceiver ackReceiver;
@Autowired
private CachingConnectionFactory connectionFactory;
@Bean
public SimpleMessageListenerContainer simpleMessageListenerContainer{
//消费者数量，默认10
int DEFAULT_CONCURRENT = 10;
//每个消费者获取最大投递数量 默认50
int DEFAULT_PREFETCH_COUNT = 50;
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
container.setConcurrentConsumers(DEFAULT_CONCURRENT);
container.setMaxConcurrentConsumers(DEFAULT_PREFETCH_COUNT);
//RabbitMQ默认是自动确认，这里改为手动确认消息
container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
//添加队列，可添加对个队列
container.addQueues(new Queue(DIRECT_QUEUE,true));
container.addQueues(new Queue(DELAY_QUEUE,true));
//设置消息处理类
container.setMessageListener(ackReceiver);
return container;
}
}在com.rabbitmq.rabbitmqconsumer.entity包中，创建UserInfo类（用户信息实体类）。
package com.rabbitmq.rabbitmqconsumer.entity;
import lombok.Data;
/**
* 用户消息实体类
* @author luoxiang
* @date 2022年01月12日 10:55 上午
*/
@Data
public class UserInfo {
private int userId;//用户编号
private String userName;//用户姓名
private String blogUrl;//博客地址
private String blogRemark;//博客信息
}在com.rabbitmq.rabbitmqconsumer.receiver包下，创建UserReceiver接口（用户消息接收接口）。
package com.rabbitmq.rabbitmqconsumer.receiver;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
/**
* 用户消息接收接口
* @author luoxiang
* @date 2022年01月12日 10:56 上午
*/
public interface UserReceiver {
/**
* 接收用户信息Json格式数据
*/
public void receiverUserJson(Message message, Channel channel) throws Exception;
/**
* 延时接收用户信息Map格式数据
*/
public void receiverDelayUserMap(Message message, Channel channel) throws Exception;
}
在com.rabbitmq.rabbitmqconsumer.receiver.impl包下，创建UserReceiverImpl类（用户消息接收类）。
package com.rabbitmq.rabbitmqconsumer.receiver.impl;
import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
import com.rabbitmq.rabbitmqconsumer.entity.UserInfo;
import com.rabbitmq.rabbitmqconsumer.receiver.UserReceiver;
import org.springframework.amqp.core.Message;
import java.util.Map;
/**
* 用户消息接收类
* @author luoxiang
* @date 2022年01月12日 10:56 上午
*/
@Service
public class UserReceiverImpl implements UserReceiver {
/**
* 接收用户信息Json格式数据
*/
@Override
public void receiverUserJson(Message message, Channel channel) throws Exception {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
//将JSON格式数据转换为实体对象
ObjectMapper mapper = new ObjectMapper();
UserInfo userInfo = mapper.readValue(message.getBody(), UserInfo.class);
System.out.println("接收者收到JSON格式消息：");
System.out.println("用户编号："+userInfo.getUserId());
System.out.println("用户名称："+userInfo.getUserName());
System.out.println("博客地址："+userInfo.getBlogUrl());
System.out.println("博客信息："+userInfo.getBlogRemark());
/**
* 确认消息，参数说明：
* long deliveryTag：唯一标识ID
* boolean multiple：是否批处理，当该参数为true时，则可以一次性确认deliveryTag小于等于传入值的所有消息
*/
channel.basicAck(deliveryTag,true);
/**
* 否定消息，参数说明：
* long deliveryTag：唯一标识ID
* boolean multiple：是否批处理，当参数为true时，则可以一次性确认deliveryTag小于等于传入值的所有消息。
* boolean requeue：如果requeue参数设置为true，则RabbitMQ会重新将这条消息存入队列，以便发送给下一个订阅的消费者；如果requeue参数设置为false，则RabbitMQ立即会把消息从队列中移除，而不会把它发送给新的消费者
*/
//channel.basicNack(deliveryTag,true,false);
}catch (Exception e){
/**
* 拒绝消息，参数说明：
* long deliveryTag：唯一标识ID
* boolean requeue：如果requeue参数设置为true，则RabbitMQ会重新将这条消息存入队列，以便发送给下一个订阅的消费者；如果requeue参数设置为false，则RabbitMQ立即会把消息从队列中移除，而不会把它发送给新的消费者
*/
channel.basicReject(deliveryTag,false);
e.printStackTrace();
}
}
/**
* 延时接收用户信息Map格式数据
*/
@Override
public void receiverDelayUserMap(Message message, Channel channel) throws Exception {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
//将JSON格式数据转换为实体类
ObjectMapper mapper = new ObjectMapper();
JavaType javaType = mapper.getTypeFactory().constructMapType(Map.class, String.class, Object.class);
Map resultMap = mapper.readValue(message.getBody(), javaType);
System.out.println("接收者收到Map格式消息：");
System.out.println("用户编号："+resultMap.get("userId"));
System.out.println("用户名称："+resultMap.get("userName"));
System.out.println("博客地址："+resultMap.get("blogUrl"));
System.out.println("博客信息："+resultMap.get("blogRemark"));
//确认消息
channel.basicAck(deliveryTag,true);
//否定消息
//channel.basicNack(deliveryTag,true,false);
}catch (Exception e){
//拒绝消息
channel.basicReject(deliveryTag,false);
e.printStackTrace();
}
}
} 在com.rabbitmq.rabbitmqconsumer.receiver.impl包下，创建AckReceiver类（消息接收确认处理类）。
package com.rabbitmq.rabbitmqconsumer.receiver.impl;
import com.rabbitmq.client.Channel;
import com.rabbitmq.rabbitmqconsumer.config.RabbitMqConfig;
import com.rabbitmq.rabbitmqconsumer.receiver.UserReceiver;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareBatchMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
/**
* 消息接收确认处理类,所有的消息，都由该类接收
* @author luoxiang
* @date 2022年01月12日 10:56 上午
*/
@Service
public class AckReceiver implements ChannelAwareBatchMessageListener {
@Autowired
private UserReceiver userReceiver;
@Override
public void onMessageBatch(List list, Channel channel) {
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
System.out.println("消息接收超过，接收时间："+dateFormat.format(new Date()));
list.forEach(message->{
//获取队列名称
String queueName = message.getMessageProperties().getConsumerQueue();
try {
//接收用户信息Json格式数据
if(queueName.equals(RabbitMqConfig.DIRECT_QUEUE)){
userReceiver.receiverUserJson(message,channel);
}
//延时接收用户信息Map格式数据
if(queueName.equals(RabbitMqConfig.DELAY_QUEUE)){
userReceiver.receiverDelayUserMap(message,channel);
}
}catch (Exception e){
e.printStackTrace();
}
});
}
@Override
public void onMessage(Message message, Channel channel) throws Exception {
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
System.out.println("消息接收成功，接收时间："+dateFormat.format(new Date()));
//获取队列名称
String queueName = message.getMessageProperties().getConsumerQueue();
//接收用户信息Json格式数据
if(queueName.equals(RabbitMqConfig.DIRECT_QUEUE)){
userReceiver.receiverUserJson(message,channel);
}
//延时接收用户信息Map格式数据
if(queueName.equals(RabbitMqConfig.DELAY_QUEUE)){
userReceiver.receiverDelayUserMap(message,channel);
}
}
} 完成上述2个项目代码后，就可以进行运行测试了。首先第二个SpringBoot项目（RabbitmqConsumer消息接收项目），让其一直运行着。
在第一个SpringBoot项目（RabbitmqProvider消息发送项目）中创建测试类，并编辑发送用户信息Json格式数据的方法。
package com.rabbitmq.rabbitmqprovider;
import com.rabbitmq.rabbitmqprovider.entity.UserInfo;
import com.rabbitmq.rabbitmqprovider.sender.UserSender;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
class RabbitmqProviderApplicationTests {
@Autowired
private UserSender userSender;
@Test
public void sendUserJson() throws Exception{
//创建用户信息
UserInfo userInfo = new UserInfo();
userInfo.setUserId(1);
userInfo.setUserName("Uncle_xiang的博客");
userInfo.setBlogUrl("https://blog.csdn.net/uncle_xiang");
userInfo.setBlogRemark("您好，欢迎访问 uncle_xiang的博客");
//执行发送
userSender.sendUserJson(userInfo);
//由于这里使用的是测试方法，当测试方法结束，RabbitMQ相关的资源也就关闭了，会导致消息确认的回调出现问题，所以加段延时，该延时与业务无关
Thread.sleep(2000);
}
}发送端执行结果：
接收端执行结果：
编写代码，测试发送用户信息Map格式数据和消息队列的延迟功能。
@Autowired
private UserSender userSender;
@Test
public void sendDelayUserMap() throws Exception{
//创建用户信息Map
Map userMap = new HashMap<>();
userMap.put("userId","1");
userMap.put("userName","uncle_xiang的博客");
userMap.put("blogUrl", "https://blog.csdn.net/uncle_xiang");
userMap.put("userRemark", "您好，欢迎访问 uncle_xiang的博客");
//执行发送
userSender.sendDelayUserMap(userMap);
//由于这里使用的是测试方法，当测试方法结束，RabbitMQ相关的资源也就关闭了，
//会导致消息确认的回调出现问题，所有加段延时，该延时与业务无关
Thread.sleep(2000);
} 发送端执行结果：
接收端执行结果：
结果说明：
在代码1.4中，设置了消息的延时为5秒钟。从上图可以看出，消息的发送时间与消息的接收时间正好间隔5秒钟，说明 RabbitMQ 消息队列的延迟功能设置成功。
| 留言与评论（共有 0 条评论） “” |
