SpringBoot（三十九）SpringBoot集成RabbitMQ实现流量削峰添谷
前边我们有具体的学习过RabbitMQ的安装和基本使用的情况。
但是呢，没有演示具体应用到项目中的实例。
这里使用RabbitMQ来实现流量的削峰添谷。
一：添加pom依赖
<!--rabbitmq-需要的 AMQP 依赖-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>二：yml配置
spring:
#配置rabbitmq 服务器
rabbitmq:
virtual-host: /
host: 1.15.157.156
port: 5672
username: xxxxx
password: xxxxx
# 开启发布确认机制
#SIMPLE, // 使用 RabbitTemplate#waitForConfirms() 或 waitForConfirmsOrDie()
#CORRELATED, // 使用 CorrelationData 关联确认与发送的消息
#NONE // 不启用发布确认
publisher-confirm-type: correlated
# publisher-confirms 消息的可靠投递, confirm 确认模式 默认为false
#publisher-confirms: true
# 添加发布确认返回, return 回退模式 默认为false
publisher-returns: true
### listener
listener:
# 每次从队列中预取5条消息
prefetch: 20
# 最小消费者数量
concurrency: 1
# 最大的消费者数量
max-concurrency: 10
simple:
# 设置预取数量为1 每次取一个
prefetch: 1
# manual：手动 ack，需要在业务代码结束后，调用 api 发送 ack，虽灵活但会提高编码复杂度。
# auto：自动 ack，没有异常则返回 ack；抛出异常则返回 nack，消息重新入队，一直到没有异常为止，也可以设置最大重试次数，超过次数后发送到专门收集错误消息的队列进一步处理
# none：关闭ack，MQ 假定消费者获取消息后会成功处理，因此消息投递后立即被删除（消息投递是不可靠的，可能丢失）
acknowledge-mode: manual
# 失败重试
retry:
# 开启消费者失败重试
enabled: true
# 初始的失败等待时长为1秒
initial-interval: 1000
# 失败的等待时长倍数，下次等待时长 = multiplier * last-interval
multiplier: 3
# 最大重试次数
max-attempts: 4
# true无状态；false有状态。如果业务中包含事务，这里改为false
stateless: true具体的配置都有对应的注释，参照即可。
三：编写config配置类
package com.modules.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
@Configuration
public class RabbitMQConfig
{
@Value("${spring.rabbitmq.host}")
private String host;
@Value("${spring.rabbitmq.port}")
private int port;
@Value("${spring.rabbitmq.username}")
private String userName;
@Value("${spring.rabbitmq.password}")
private String password;
@Value("${spring.rabbitmq.listener.prefetch}")
private int prefetch;
@Value("${spring.rabbitmq.listener.concurrency}")
private int concurrentConsumers;
@Value("${spring.rabbitmq.listener.max-concurrency}")
private int maxConcurrentConsumers;
/**
* 链接RabbitMQ
* @return
*/
@Bean
public ConnectionFactory connectionDirectFactory()
{
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setHost(host);
connectionFactory.setPort(port);
connectionFactory.setUsername(userName);
connectionFactory.setPassword(password);
connectionFactory.setPublisherConfirms(true); //必须要设置
return connectionFactory;
}
/**
* 配置RabbitMQ参数
* @return
*/
@Bean
public SimpleRabbitListenerContainerFactory rabbitDirectListenerContainerFactory()
{
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionDirectFactory());
//设置最小并发的消费者数量
factory.setConcurrentConsumers(concurrentConsumers);
//设置最大并发的消费者数量
factory.setMaxConcurrentConsumers(maxConcurrentConsumers);
//限流，单位时间内消费多少条记录
factory.setPrefetchCount(prefetch);
// json转消息
//factory.setMessageConverter(new Jackson2JsonMessageConverter());
//设置rabbit 确认消息的模式，默认是自动确认
//factory.setAcknowledgeMode(AcknowledgeMode.AUTO);
//设置rabbit 确认消息的模式，默认是自动确认
factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
return factory;
}
/**
* 回调函数
* @param connectionFactory
* @return
*/
@Bean
public RabbitTemplate createDirectRabbitTemplate(ConnectionFactory connectionFactory)
{
RabbitTemplate rabbitTemplate = new RabbitTemplate();
rabbitTemplate.setConnectionFactory(connectionFactory);
//设置开启Manatory，才能触发回调函数，无论消息推送结果怎么样都会强制调用回调函数
rabbitTemplate.setMandatory(true);
// 设置确认发送到交换机的回调函数 =》 消息推送到server，但是在server里找不到交换机 / 消息推送到sever，交换机和队列啥都没找到 / 消息推送到server，找到交换机了，但是没找到队列 / 消息推送成功
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
if(ack)
{
System.out.println("发送者消息确认成功！");
}
else
{
System.out.println("发送者消息确认是呗，考虑重发："+cause);
}
//System.out.println("相关数据："+correlationData);
//System.out.println("确认情况："+ack);
//System.out.println("原因："+cause);
//System.out.println("===============================");
});
//设置确认消息已发送到队列的回调 =》 消息推送到server，找到交换机了，但是没找到队列 触发这个回调函数
rabbitTemplate.setReturnsCallback(returnedMessage -> {
System.out.println("交换机为："+returnedMessage.getExchange());
System.out.println("返回消息为："+returnedMessage.getMessage());
System.out.println("路由键为："+returnedMessage.getRoutingKey());
System.out.println("回应消息为："+returnedMessage.getReplyText());
System.out.println("回应代码为："+returnedMessage.getReplyCode());
System.out.println("===============================");
});
return rabbitTemplate;
}
@Bean
Queue trafficSpikedQueue()
{
return new Queue("trafficSpikedQueue", true);
}
@Bean
DirectExchange trafficSpikedExchange()
{
return new DirectExchange("trafficSpikedExchange");
}
@Bean
Binding binding(Queue trafficSpikedQueue, DirectExchange trafficSpikedExchange)
{
return BindingBuilder.bind(trafficSpikedQueue).to(trafficSpikedExchange).with("trafficSpikedKey");
}//*/
}四：创建生产者
package com.modules.controller.rabbitmq;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class TrafficController {
@Autowired
private RabbitTemplate rabbitTemplate;
@GetMapping("/java/traffic")
public String sendTrafficMessage(@RequestParam String message)
{
for (int i = 1; i <= 100; i++)
{
// 使用java多线程来模拟多用户并发请求
final int temp = i;
new Thread(
()->{
// 给RabbitMQ发送消息
rabbitTemplate.convertAndSend(
"trafficExchange",
"trafficKey",
"hello world:"+temp,
new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException
{
// System.out.println("发送回调:"+temp);
System.out.println(message);
return message;
}
});
}
).start();
}
// rabbitTemplate.convertAndSend("trafficSpikedExchange", "trafficSpikedKey", message);
return "Message sent";
}
}五：创建消费者
package com.modules.controller.rabbitmq;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import org.springframework.amqp.core.Message;
import com.rabbitmq.client.*;
import java.io.IOException;
@Component
public class TrafficSpikedConsumer {
@RabbitListener(queues = "trafficQueue")
public void receiveMessage(Message message, Channel channel) throws InterruptedException, IOException
{
// 为了演示一个一个消费的情况，这里使用线程暂停来延迟控制台输出
Thread.sleep(100);
// =========================================
// 处理消息，例如写入数据库或进行计算
System.out.println("Received message: " + new String(message.getBody()));
//System.out.println("channel: " + channel);
// =========================================
// 成功处理后手动确认消息
long deliveryTag = message.getMessageProperties().getDeliveryTag();
//System.out.println("deliveryTag:"+deliveryTag);
channel.basicAck(deliveryTag, false);
}
}控制台输出的数据比较多。我这里就不做展示了。
PS：我这里测试的时候遇到一个小问题，发现消费者最后消费的数量跟生产者生产的数量对不上。我百思不得其解。这问题出在哪里呢？
后来，我才发现，我测试是在本地做的测试，对应的代码，我服务器端打包的jar里边也有一份，也就是说，我一个生产者，对应两个消费者（本地+服务器）这也是我本地消费者消费的数量跟生产数量不一致的原因。
以上大概就是Springboot集成RabbitMQ实现流量削峰添谷的一个小例子。
通过RabbitMQ的队列机制，可以有效地缓解高峰期的流量压力。
有好的建议，请在下方输入你的评论。