简单的SpringBoot整合redis和RabbitMQ秒杀商品学习笔记

实现思路

当多个用户同时点击下单时，假设商品库存量为100个，但同时下单用户却远远大于100，所以这时就需要MQ消息队列来控制，为了防止单一用户抢多次，就将用户存在redis缓存中，大概是这么个意思。

这也有效防止了电商项目的超卖现象。

首先我们可以先去安装redis服务和rabbitmq服务（省略），然后在数据库中建立相应表来模拟下单减库存的操作。

数据库信息

这里就用笔者借鉴上篇博客的数据库来实现。

创建商品详情表

DROP TABLE IF EXISTS `goods_info`;
CREATE TABLE `goods_info`  (
`id` varchar(64) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '主键id',
`goods_name` varchar(500) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '商品名称',
`goods_stock` int(8) NULL DEFAULT NULL COMMENT '商品剩余库存',
`create_time` datetime(0) NULL DEFAULT NULL,
`update_time` datetime(0) NULL DEFAULT NULL,
PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Dynamic;

SET FOREIGN_KEY_CHECKS = 1;

CREATE TABLE `seckill`.`Untitled`  (
`id` varchar(64) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '主键id',
`user_id` varchar(64) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '订单所属用户id',
`goods_id` varchar(64) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '商品id',
`pay_status` int(1) NULL DEFAULT NULL COMMENT '支付状态 0-超时未支付  1-已支付  2-待支付',
`create_time` datetime(0) NULL DEFAULT NULL,
`update_time` datetime(0) NULL DEFAULT NULL
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Dynamic;

创建抢购信息表

表创建完成后，在dao层写好相关操作

（库存减一，插入下单记录，查询订单状态，更新状态等）

具体实现

接下来是rabbitmq的配置类

/**
* rabbitmq配置
*/
@Configuration
public class OrderRabbitmqConfig {

private static final Logger logger = LoggerFactory.getLogger(OrderRabbitmqConfig.class);

@Autowired
private Environment env;

/**
* channel链接工厂
*/
@Autowired
private CachingConnectionFactory connectionFactory;

/**
* 监听器容器配置
*/
@Autowired
private SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;

/**
* 声明rabbittemplate
* @return
*/
@Bean
public RabbitTemplate rabbitTemplate(){
//消息发送成功确认，对应application.properties中的spring.rabbitmq.publisher-confirms=true
connectionFactory.setPublisherConfirms(true);
//消息发送失败确认，对应application.properties中的spring.rabbitmq.publisher-returns=true
connectionFactory.setPublisherReturns(true);
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
//设置消息发送格式为json
rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
rabbitTemplate.setMandatory(true);
//消息发送到exchange回调 需设置：spring.rabbitmq.publisher-confirms=true
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
logger.info("消息发送成功:correlationData({}),ack({}),cause({})",correlationData,ack,cause);
}
});
//消息从exchange发送到queue失败回调  需设置：spring.rabbitmq.publisher-returns=true
rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
logger.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}",exchange,routingKey,replyCode,replyText,message);
}
});
return rabbitTemplate;
}

//---------------------------------------订单队列------------------------------------------------------

/**
* 声明订单队列的交换机
* @return
*/
@Bean("orderTopicExchange")
public TopicExchange orderTopicExchange(){
//设置为持久化 不自动删除
return new TopicExchange(env.getProperty("order.mq.exchange.name"),true,false);
}

/**
* 声明订单队列
* @return
*/
@Bean("orderQueue")
public Queue orderQueue(){
return new Queue(env.getProperty("order.mq.queue.name"),true);
}

/**
* 将队列绑定到交换机
* @return
*/
@Bean
public Binding simpleBinding(){
return BindingBuilder.bind(orderQueue()).to(orderTopicExchange()).with(env.getProperty("order.mq.routing.key"));
}

/**
* 注入订单对列消费监听器
*/
@Autowired
private OrderListener orderListener;

/**
* 声明订单队列监听器配置容器
* @return
*/
@Bean("orderListenerContainer")
public SimpleMessageListenerContainer orderListenerContainer(){
//创建监听器容器工厂
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
//将配置信息和链接信息赋给容器工厂
factoryConfigurer.configure(factory,connectionFactory);
//容器工厂创建监听器容器
SimpleMessageListenerContainer container = factory.createListenerContainer();
//指定监听器
container.setMessageListener(orderListener);
//指定监听器监听的队列
container.setQueues(orderQueue());
return container;
}

}

由于在MQ这方面咱没怎么了解过，为了下次方便自己观看，咱们baidu一下。

baidu到的RabbitMQ的大致流程如下。（感谢带佬的图）

• 生产者把消息投递到exchange交换机
• exchange接受到消息后,根据消息的key和已经设置好的binging,进行消息路由,将消息投递到一个或者多个queue里面
• 消费者从queue中取出消息,开始执行相应的代码

生产者，在这里就是指消息的创建者，负责创建和推送数据到消息服务器。

问题来了，exchange交换机和queue分别是什么？

继续。

• Exchange（交换器）：用于接受、分配消息；
• Queue（队列）：用于存储生产者的消息；

Exchange 与 Queue 是通过 RoutingKye 来进行关联的，且 Exchange 与 Queue 是多对多的关系。

生产者通过交换机将消息存在队列中以后，就需要消费者从中获取消息。

看看带佬对于消费者的解释。

消费者很容易理解。他们连接到代理服务器上，并订阅到队列（queue）上。把消息队列想象成一个具名邮箱。每当消息到达特定的邮箱时，RabbitMQ会将其发送给其中一个订阅的/监听的消费者。当消费者接收到消息时，他只得到消息的一部分：有效荷载。在消息路由过程中，消息的标签并没有随着有效载荷一同传递。RabbitMQ甚至不会告诉你是谁生产/发送了消息。就好比你拿起信件时，却发现所有的信封都是空白的。想要知道这条消息是否是从Millie姑妈发来的唯一方式是他在信里签了名。同理，如果需要明确知道是谁生产的AMQP消息的话，就要看生产者是否把发送信息放入有效载荷中。

代码依然在上面的Service类里。

所以这里自己总结一下大致流程大概是：

生产程序 产生消息 => mq服务器 => 交换机 => 队列 => 消费者 => 取出

交换机设置好以后，我们来设置生产程序。

先看看消费者代码。

/**
* 消息监听器（消费者）
*/
@Component
public class OrderListener implements ChannelAwareMessageListener {

private static final Logger logger = LoggerFactory.getLogger(OrderListener.class);

@Autowired
private OrderService orderService;
/**
* 处理接收到的消息
* @param message 消息体
* @param channel 通道，确认消费用
* @throws Exception
*/
@Override
public void onMessage(Message message, Channel channel) throws Exception {
try{
//获取交付tag
long tag = message.getMessageProperties().getDeliveryTag();
String str = new String(message.getBody(),"utf-8");
logger.info("接收到的消息：{}",str);
JSONObject obj = JSONObject.parseObject(str);
//下单，操作数据库
orderService.order(obj.getString("userId"),obj.getString("goodsId"));
//确认消费
channel.basicAck(tag,true);
}catch(Exception e){
logger.error("消息监听确认机制发生异常：",e.fillInStackTrace());
}
}
}

生产者代码

@Service
public class OrderService {

@Resource
private SeckillMapper seckillMapper;

@Autowired
private RabbitTemplate rabbitTemplate;

@Autowired
private Environment env;

/**
* 下单，操作数据库
* @param userId
* @param goodsId
*/
@Transactional()
public void order(String userId,String goodsId){
//该商品库存-1（当库存>0时）
int count = seckillMapper.reduceGoodsStockById(goodsId);
//更新成功，表明抢单成功，插入下单记录，支付状态设为2-待支付
if(count > 0){
OrderRecord orderRecord = new OrderRecord();
orderRecord.setId(CommonUtils.createUUID());
orderRecord.setGoodsId(goodsId);
orderRecord.setUserId(userId);
orderRecord.setPayStatus(2);
seckillMapper.insertOrderRecord(orderRecord);
//将该订单添加到支付队列
rabbitTemplate.setExchange(env.getProperty("pay.mq.exchange.name"));
rabbitTemplate.setRoutingKey(env.getProperty("pay.mq.routing.key"));
rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
String json = JSON.toJSONString(orderRecord);
Message msg = MessageBuilder.withBody(json.getBytes()).build();
rabbitTemplate.convertAndSend(msg);
}
}
}

在上面说过为了防止单一用户，需要将消息存入redis，但是，如果已经存过了，就不需要再存一次。

贴出代码。

/**
* 校验同一用户对同一商品是否重复下单
* @param order 下单参数封装
* @return true-通过校验  false-未通过
*/
public boolean checkSeckillUser(OrderRequest order) {

String key = env.getProperty("seckill.redis.key.prefix") + order.getUserId() + order.getGoodsId();
/**
* 1.返回ture,表明该用户与该商品第一次匹配进redis,即该用户首次秒杀该商品
* 2.利用redis的setnx原理，避免多服务、高并发带来的线程安全问题
* 3.一行代码搞定
* 4.没有太具体的业务场景，不设置过期时间
*/
return redisTemplate.opsForValue().setIfAbsent(key,"1");

/* 老代码，代码冗余，未加锁，线程不安全
String key = env.getProperty("seckill.redis.key.prefix")+userId;
//用户秒杀过的商品id集合
Set<String> goodsIdSet = redisTemplate.opsForSet().members(key);
if(goodsIdSet != null && goodsIdSet.contains(goodsId)){
//不是第一次秒杀，返回false
return false;
}else{
//是第一次秒杀，将该id添加到该用户的集合下面
redisTemplate.opsForSet().add(key,goodsId);
return true;
}
*/
}

下面有参考代码的博文地址，这里不详细贴出了。

在带佬们的博文里，又看见了对消息的解释，这里也稍微CV一下。

消息包含两部分内容：有效载荷（payload）和标签（label）。有效载荷就是你想要传输的数据。他可以是任何内容，一个JSON数组或者是你最喜欢的iguana Ziggy的MPEG-4。RabbitMQ不会在意这些。标签就更有趣了。他描述了有效载荷，并且RabbitMQ用他来决定谁将获得消息的拷贝。举例来说，不同于TCP协议的是，当你明确指定发送方和接收方时，AMQP只会用标签表述这条消息（一个交换器的名称和可选的主题标记），然后把消息交由Rabbit。Rabbit会根据标签把消息发送给感兴趣的接收方。这种通信方式是一种“发后即忘”（fire-and-forget）的单向方式。

参考代码博文：https://blog.csdn.net/qq_36798969/article/details/93594354

• 点赞
• 收藏
• 分享
• 文章举报