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.156port: 5672username: xxxxxpassword: xxxxx# 开启发布确认机制#SIMPLE,     // 使用 RabbitTemplate#waitForConfirms() 或 waitForConfirmsOrDie()#CORRELATED, // 使用 CorrelationData 关联确认与发送的消息#NONE        // 不启用发布确认publisher-confirm-type: correlated# publisher-confirms 消息的可靠投递, confirm 确认模式 默认为false#publisher-confirms: true# 添加发布确认返回, return 回退模式 默认为falsepublisher-returns: true### listenerlistener:# 每次从队列中预取5条消息prefetch: 20# 最小消费者数量concurrency: 1# 最大的消费者数量max-concurrency: 10simple:# 设置预取数量为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-intervalmultiplier: 3# 最大重试次数max-attempts: 4# true无状态；false有状态。如果业务中包含事务，这里改为falsestateless: 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*/@Beanpublic 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*/@Beanpublic 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*/@Beanpublic 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;}@BeanQueue trafficSpikedQueue(){return new Queue("trafficSpikedQueue", true);}@BeanDirectExchange trafficSpikedExchange(){return new DirectExchange("trafficSpikedExchange");}@BeanBinding 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 {@Autowiredprivate 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() {@Overridepublic 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的队列机制，可以有效地缓解高峰期的流量压力。

有好的建议，请在下方输入你的评论。