赞
踩
转载:springboot集成rabbitmq商品秒杀业务实战(流量削峰)
消息队列如何实现流量削峰?
要对流量进行削峰,最容易想到的解决方案就是用消息队列来缓冲瞬时流量,把同步的直接调用转换成异步的间接推送,中间通过一个队列在一端承接瞬时的流量洪峰,在另一端平滑地将消息推送出去。
这里就不讲springboot
和rabbitmq
如何集成了,参考文章https://www.cnblogs.com/fantongxue/p/12493497.html
数据库有一张商品表,库存量是100。现在有1000个消费者准备开抢这100个库存。
t_product
表维护商品编号与商品库存剩余数量。编号No123321的这种商品的库存量有100个。
t_product_record
维护抢到商品的用户ID。理论上t_product
表开抢后的 记录数量应该是100条(共有100个人抢到了商品)。
我们使用压力测试工具jweter
对其进行并发性测试。
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
</dependency>
spring:
datasource:
driver-class-name: com.mysql.cj.jdbc.Driver
url: jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=utf-8&useSSL=true&serverTimezone=UTC
username: root
password: 1234
rabbitmq:
host: 101.201.101.206
username: guest
password: guest
publisher-confirms: true # 开启Rabbitmq发送消息确认机制,发送消息到队列并触发回调方法
publisher-returns: true
listener:
simple:
concurrency: 10 #消费者数量
max-concurrency: 10 #最大消费者数量
prefetch: 1 #限流(消费者每次从队列获取的消息数量)
auto-startup: true #启动时自动启动容器
acknowledge-mode: manual #开启ACK手动确认模式
1, 定义消息转换实例 ,转化成 JSON
传输
2 , 配置启用rabbitmq
事务
package com.aaa.springredis.controller;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
@Component
public class RabbitConfig {
<span class="token comment">/** * 定义消息转换实例 ,转化成 JSON传输 * * @return Jackson2JsonMessageConverter */</span> <span class="token annotation punctuation">@Bean</span> <span class="token keyword">public</span> MessageConverter <span class="token function">integrationEventMessageConverter</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token punctuation">{<!-- --></span> <span class="token keyword">return</span> <span class="token keyword">new</span> <span class="token class-name">Jackson2JsonMessageConverter</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span> <span class="token punctuation">}</span> <span class="token comment">/** * 配置启用rabbitmq事务 * * @param connectionFactory connectionFactory * @return RabbitTransactionManager */</span> <span class="token annotation punctuation">@Bean</span> <span class="token keyword">public</span> RabbitTransactionManager <span class="token function">rabbitTransactionManager</span><span class="token punctuation">(</span>CachingConnectionFactory connectionFactory<span class="token punctuation">)</span> <span class="token punctuation">{<!-- --></span> <span class="token keyword">return</span> <span class="token keyword">new</span> <span class="token class-name">RabbitTransactionManager</span><span class="token punctuation">(</span>connectionFactory<span class="token punctuation">)</span><span class="token punctuation">;</span> <span class="token punctuation">}</span>
}
说明:被@PostConstruct
修饰的方法会在服务器加载Servlet
的时候运行,并且只会被服务器执行一次。如果想在生成对象时完成某些初始化操作,而偏偏这些初始化操作又依赖于依赖注入,那么久无法在构造函数中实现。为此,可以使用@PostConstruct
注解一个方法来完成初始化,@PostConstruct
注解的方法将会在依赖注入完成后被自动调用。
回调函数的使用前提是配置文件中开启了rabitmq
消息确认机制
Constructor >> @Autowired >> @PostConstruct
@Autowired RabbitTemplate rabbitTemplate;
<span class="token keyword">private</span> <span class="token keyword">static</span> <span class="token keyword">final</span> Logger LOGGER <span class="token operator">=</span> LoggerFactory<span class="token punctuation">.</span><span class="token function">getLogger</span><span class="token punctuation">(</span>RabbitController<span class="token punctuation">.</span><span class="token keyword">class</span><span class="token punctuation">)</span><span class="token punctuation">;</span> <span class="token annotation punctuation">@PostConstruct</span> <span class="token keyword">private</span> <span class="token keyword">void</span> <span class="token function">init</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">{<!-- --></span> <span class="token comment">/** * 消息发送到交换器Exchange后触发回调。 * 使用该功能需要开启确认,spring-boot中配置如下: * spring.rabbitmq.publisher-confirms = true */</span> rabbitTemplate<span class="token punctuation">.</span><span class="token function">setConfirmCallback</span><span class="token punctuation">(</span><span class="token keyword">new</span> <span class="token class-name">RabbitTemplate<span class="token punctuation">.</span>ConfirmCallback</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token punctuation">{<!-- --></span> <span class="token annotation punctuation">@Override</span> <span class="token keyword">public</span> <span class="token keyword">void</span> <span class="token function">confirm</span><span class="token punctuation">(</span>CorrelationData correlationData<span class="token punctuation">,</span> <span class="token keyword">boolean</span> b<span class="token punctuation">,</span> String s<span class="token punctuation">)</span> <span class="token punctuation">{<!-- --></span> <span class="token keyword">if</span> <span class="token punctuation">(</span>b<span class="token punctuation">)</span> <span class="token punctuation">{<!-- --></span> System<span class="token punctuation">.</span>out<span class="token punctuation">.</span><span class="token function">println</span><span class="token punctuation">(</span><span class="token string">"消息已确认 cause:{"</span><span class="token operator">+</span>s<span class="token operator">+</span><span class="token string">"} - {"</span><span class="token operator">+</span>correlationData<span class="token operator">+</span><span class="token string">"}"</span><span class="token punctuation">)</span><span class="token punctuation">;</span> <span class="token punctuation">}</span> <span class="token keyword">else</span> <span class="token punctuation">{<!-- --></span> System<span class="token punctuation">.</span>out<span class="token punctuation">.</span><span class="token function">println</span><span class="token punctuation">(</span><span class="token string">"消息未确认 cause:{"</span><span class="token operator">+</span>s<span class="token operator">+</span><span class="token string">"} - {"</span><span class="token operator">+</span>correlationData<span class="token operator">+</span><span class="token string">"}"</span><span class="token punctuation">)</span><span class="token punctuation">;</span> <span class="token punctuation">}</span> <span class="token punctuation">}</span> <span class="token punctuation">}</span><span class="token punctuation">)</span><span class="token punctuation">;</span> <span class="token comment">/** * 通过实现ReturnCallback接口, * 如果消息从交换器发送到对应队列失败时触发 * 比如根据发送消息时指定的routingKey找不到队列时会触发 * 使用该功能需要开启确认,spring-boot中配置如下: * spring.rabbitmq.publisher-returns = true */</span> rabbitTemplate<span class="token punctuation">.</span><span class="token function">setReturnCallback</span><span class="token punctuation">(</span><span class="token keyword">new</span> <span class="token class-name">RabbitTemplate<span class="token punctuation">.</span>ReturnCallback</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token punctuation">{<!-- --></span> <span class="token annotation punctuation">@Override</span> <span class="token keyword">public</span> <span class="token keyword">void</span> <span class="token function">returnedMessage</span><span class="token punctuation">(</span>Message message<span class="token punctuation">,</span> <span class="token keyword">int</span> replyCode<span class="token punctuation">,</span> String replyText<span class="token punctuation">,</span> String exchange<span class="token punctuation">,</span> String routingKey<span class="token punctuation">)</span> <span class="token punctuation">{<!-- --></span> LOGGER<span class="token punctuation">.</span><span class="token function">error</span><span class="token punctuation">(</span><span class="token string">"消息被退回:{}"</span><span class="token punctuation">,</span> message<span class="token punctuation">)</span><span class="token punctuation">;</span> LOGGER<span class="token punctuation">.</span><span class="token function">error</span><span class="token punctuation">(</span><span class="token string">"消息使用的交换机:{}"</span><span class="token punctuation">,</span> exchange<span class="token punctuation">)</span><span class="token punctuation">;</span> LOGGER<span class="token punctuation">.</span><span class="token function">error</span><span class="token punctuation">(</span><span class="token string">"消息使用的路由键:{}"</span><span class="token punctuation">,</span> routingKey<span class="token punctuation">)</span><span class="token punctuation">;</span> LOGGER<span class="token punctuation">.</span><span class="token function">error</span><span class="token punctuation">(</span><span class="token string">"描述:{}"</span><span class="token punctuation">,</span> replyText<span class="token punctuation">)</span><span class="token punctuation">;</span> <span class="token punctuation">}</span> <span class="token punctuation">}</span><span class="token punctuation">)</span><span class="token punctuation">;</span> <span class="token punctuation">}</span>
写抢单测试类,我们使用jweter
压力测试工具开启1000个线程进行测试(开启多线程并发测试),所以为了区别每一个模拟的用户,使用userId
累加的方式进行区分。
private int userId=0;
//开始抢单
@RequestMapping("/begin")
@ResponseBody
public void begin(){
userId++;
this.send(userId);
}
而上面的send
方法就是把接收到的用户请求发送到rabbitmq
消息中间件中。
@RequestMapping("/send")
@ResponseBody
public String send(Integer messge){
//第一个参数:交换机名字 第二个参数:Routing Key的值 第三个参数:传递的消息对象
rabbitTemplate.convertAndSend("test.direct","test",messge);
return "发送消息成功";
}
rabbitmq
监听上篇文章也说过了,作用就是监听指定队列中收到来自交换机的消息,来一条收一条,收完为止!
通过ACK
确认是否被正确接收,每个Message
都要被确认,可以手动去 ACK
或自动ACK
,如果信息消费失败的话会拒绝当前消息,并把消息返回原队列。
从队列中收到用户的userId
,然后进行购买商品模拟操作(减少一个库存,新增一条购买记录)
@Autowired RabbitController controller;
<span class="token comment">/** * @RabbitListener 可以标注在类上面,需配合 @RabbitHandler 注解一起使用 * @RabbitListener 标注在类上面表示当有收到消息的时候,就交给 @RabbitHandler 的方法处理,具体使用哪个方法处理, * 根据 MessageConverter 转换后的参数类型 * * 使用 @Payload 和 @Headers 注解可以消息中的 body 与 headers 信息 * * 通过 ACK 确认是否被正确接收,每个 Message 都要被确认(acknowledged),可以手动去 ACK 或自动 ACK */</span> <span class="token annotation punctuation">@RabbitListener</span><span class="token punctuation">(</span>queues <span class="token operator">=</span> <span class="token string">"test"</span><span class="token punctuation">)</span> <span class="token comment">//指定监听的队列名</span> <span class="token keyword">public</span> <span class="token keyword">void</span> <span class="token function">receiver</span><span class="token punctuation">(</span><span class="token annotation punctuation">@Payload</span> Integer userId<span class="token punctuation">,</span> <span class="token annotation punctuation">@Headers</span> Channel channel<span class="token punctuation">,</span> Message message<span class="token punctuation">)</span> <span class="token keyword">throws</span> IOException <span class="token punctuation">{<!-- --></span> LOGGER<span class="token punctuation">.</span><span class="token function">info</span><span class="token punctuation">(</span><span class="token string">"用户{}开始抢单"</span><span class="token punctuation">,</span> userId<span class="token punctuation">)</span><span class="token punctuation">;</span> <span class="token keyword">try</span> <span class="token punctuation">{<!-- --></span> <span class="token comment">//处理消息</span> controller<span class="token punctuation">.</span><span class="token function">robbingProduct</span><span class="token punctuation">(</span>userId<span class="token punctuation">)</span><span class="token punctuation">;</span>
// 确认消息已经消费成功
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
LOGGER.error(“消费处理异常:{} - {}”, userId, e);
// 拒绝当前消息,并把消息返回原队列
channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
}
}
购买商品的方法
public void robbingProduct(Integer userId){ Product product = testDao.selectProductByNo("No123321"); if (product != null && product.getTotal() > 0) { //更新库存表,库存量减少1。返回1说明更新成功。返回0说明库存已经为0 int i = testDao.updateProduct("No123321"); if(i>0){ //插入记录 testDao.insertProductRecord(new ProductRecord("No123321", userId)); //发送短信 LOGGER.info("用户{}抢单成功", userId); }else { LOGGER.error("用户{}抢单失败", userId); } } else { LOGGER.error("用户{}抢单失败", userId); }
<span class="token punctuation">}</span>
jweter
压力测试工具如何使用百度吧,这里忽略!
控制台打印
而数据库中的库存变成了0
购买记录中存放了抢单成功的用户id
(100条记录)
当然,剩下的900个用户都抢单失败了!
rabbitmq
队列是先进先出的顺序,先来后到,1000个请求你也得给我排队,前100个请求抢单成功之后就注定了后900个请求是抢单失败的!
使用RabbitMQ
的最主要变化就是:以前抢单操作请求直接由我们抢单应用程序执行,现在请求被转移到了RabbitMQ
服务器中。RabbitMQ
服务器把接收到的抢单请求进行排队,最后由RabbitMQ
服务器把抢单请求转发到我们的抢单应用程序,这样的好处就是避免我们的抢单应用程序短时间直接处理大量请求。RabbitMQ
服务器主要作用是减缓抢单应用程序的并发压力,相当于在我们的抢单程序之前加了一道请求缓冲区。
实战结束!
工程地址:https://github.com/fantongxue666/rabbitmq-seckill
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。