尚品汇-秒杀商品定时任务存入缓存、Redis发布订阅实现状态位(五十一)
目录:
(1)秒杀业务分析
(2)搭建秒杀模块
(3)秒杀商品导入缓存
(4)redis发布与订阅实现
(1)秒杀业务分析
需求分析
所谓“秒杀”,就是网络卖家发布一些超低价格的商品,所有买家在同一时间网上抢购的一种销售方式。通俗一点讲就是网络商家为促销等目的组织的网上限时抢购活动。由于商品价格低廉,往往一上架就被抢购一空,有时只用一秒钟。
秒杀商品通常有三种限制:库存限制、时间限制、购买量限制。
(1)库存限制:商家只拿出限量的商品来秒杀。比如某商品实际库存是200件,但只拿出50件来参与秒杀。我们可以将其称为“秒杀库存”。商家赔本赚吆喝,图啥?人气!
(2)时间限制:通常秒杀都是有特定的时间段,只能在设定时间段进行秒杀活动;
(3)购买量限制:同一个商品只允许用户最多购买几件。比如某手机限购1件。张某第一次买个1件,那么在该次秒杀活动中就不能再次抢购
需求:B2B2C
- 商家提交秒杀商品申请,录入秒杀商品数据,主要包括:商品标题、原价、秒杀价、商品图片、介绍等信息
- 运营商审核秒杀申请
- 秒杀频道首页列出当天的秒杀商品,点击秒杀商品图片跳转到秒杀商品详细页。
- 商品详细页显示秒杀商品信息,点击立即抢购进入秒杀,抢购成功时预减库存。当库存为0或不在活动期范围内时无法秒杀。
- 秒杀成功,进入下单页填写收货地址、电话、收件人等信息,完成下订单,然后跳转到支付页面,支付成功,跳转到成功页,完成秒杀。
- 当用户秒杀下单30分钟内未支付,取消订单,调用微信支付或支付宝的关闭订单接口。
秒杀功能分析
列表页
详情页
排队页
下单页
支付页
数据库表
秒杀商品表seckill_goods
SeckillGoods
package com.atguigu.gmall.model.activity;
import com.atguigu.gmall.model.base.BaseEntity;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableName;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import java.math.BigDecimal;
import java.util.Date;
@Data
@ApiModel(description = "SeckillGoods")
@TableName("seckill_goods")
public class SeckillGoods extends BaseEntity {
private static final long serialVersionUID = 1L;
@ApiModelProperty(value = "spu ID")
@TableField("spu_id")
private Long spuId;
@ApiModelProperty(value = "sku ID")
@TableField("sku_id")
private Long skuId;
@ApiModelProperty(value = "标题")
@TableField("sku_name")
private String skuName;
@ApiModelProperty(value = "商品图片")
@TableField("sku_default_img")
private String skuDefaultImg;
@ApiModelProperty(value = "原价格")
@TableField("price")
private BigDecimal price;
@ApiModelProperty(value = "秒杀价格")
@TableField("cost_price")
private BigDecimal costPrice;
@ApiModelProperty(value = "添加日期")
@TableField("create_time")
private Date createTime;
@ApiModelProperty(value = "审核日期")
@TableField("check_time")
private Date checkTime;
@ApiModelProperty(value = "审核状态")
@TableField("status")
private String status;
@ApiModelProperty(value = "开始时间")
@TableField("start_time")
private Date startTime;
@ApiModelProperty(value = "结束时间")
@TableField("end_time")
private Date endTime;
@ApiModelProperty(value = "秒杀商品数")
@TableField("num")
private Integer num;
@ApiModelProperty(value = "剩余库存数")
@TableField("stock_count")
private Integer stockCount;
@ApiModelProperty(value = "描述")
@TableField("sku_desc")
private String skuDesc;
}
秒杀实现思路
- 秒杀的商品要提前放入到redis中(缓存预热),什么时间放入?凌晨放入当天的秒杀商品数据。
- 状态位控制访问请求,何为状态位?就是我们在内存中保存一个状态(Map存储状态),当抢购开始时状态为1,可以抢购,当库存为0时,状态位0,不能抢购;状态位的好处,他是在内存中判断,压力很小,可以阻止很多不必要的请求
- 用户提交秒杀请求,将秒杀商品与用户id关联发送给mq,然后返回,秒杀页面通过轮询接口查看是否秒杀成功
- 我们秒杀只是为了获取一个秒杀资格,获取秒杀资格就可以到下单页下订单,后续业务与正常订单一样
Map相当于本地缓存空间(JVM本地缓存空间),map存数据对象在本地缓存堆空间当中,本地的缓存效率高于远程的 ,没有经过磁盘IO,没有Redis远程访问请求,本地缓存直接判断,效率最高
下单我们需要注意的问题:
状态位如何同步到集群中的其他节点?
如何控制一个用户只下一个订单?
如何控制库存超买?
如何控制访问压力?
用消息队列不行,因为消息队列只能被集群中的一个监听到,只能用Redis的发布订阅,所有的消费者只要监听的队列无论你是集群还是分布式,都能够接收到
(2)搭建秒杀模块
我们先把秒杀模块搭建好,秒杀一共有三个模块,秒杀微服务模块service-activity,负责封装秒杀全部服务端业务;秒杀前端模块web-all中添加,负责前端显示业务;service-activity-client api接口模块
搭建service-activity模块
搭建方式如service-order
修改pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.atguigu.gmall</groupId>
<artifactId>service</artifactId>
<version>1.0</version>
</parent>
<version>1.0</version>
<artifactId>service-activity</artifactId>
<packaging>jar</packaging>
<name>service-activity</name>
<description>service-activity</description>
<dependencies>
<dependency>
<groupId>com.atguigu.gmall</groupId>
<artifactId>service-user-client</artifactId>
<version>1.0</version>
</dependency>
<dependency>
<groupId>com.atguigu.gmall</groupId>
<artifactId>service-product-client</artifactId>
<version>1.0</version>
</dependency>
<dependency>
<groupId>com.atguigu.gmall</groupId>
<artifactId>service-order-client</artifactId>
<version>1.0</version>
</dependency>
<dependency>
<groupId>com.atguigu.gmall</groupId>
<artifactId>rabbit-util</artifactId>
<version>1.0</version>
</dependency>
</dependencies>
<build>
<finalName>service-activity</finalName>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
添加配置bootstrap.properties
spring.application.name=service-activity
spring.profiles.active=dev
spring.cloud.nacos.discovery.server-addr=192.168.200.129:8848
spring.cloud.nacos.config.server-addr=192.168.200.129:8848
spring.cloud.nacos.config.prefix=${spring.application.name}
spring.cloud.nacos.config.file-extension=yaml
spring.cloud.nacos.config.shared-configs[0].data-id=common.yaml
启动类
package com.atguigu.gmall.activity;
@SpringBootApplication
@ComponentScan({"com.atguigu.gmall"})
@EnableDiscoveryClient
@EnableFeignClients(basePackages= {"com.atguigu.gmall"})
public class ServiceActivityApplication {
public static void main(String[] args) {
SpringApplication.run(ServiceActivityApplication.class, args);
}
}
搭建service-activity-client模块
搭建方式如service-order-client
修改pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.atguigu.gmall</groupId>
<artifactId>service-client</artifactId>
<version>1.0</version>
</parent>
<artifactId>service-activity-client</artifactId>
<version>1.0</version>
<packaging>jar</packaging>
<name>service-activity-client</name>
<description>service-activity-client</description>
</project>
添加依赖,配置网关
在web-all中引入依赖
<dependency>
<groupId>com.atguigu.gmall</groupId>
<artifactId>service-activity-client</artifactId>
<version>1.0</version>
</dependency>
在网关项目中配置秒杀服务,域名
- id: web-activity
uri: lb://web-all
predicates:
- Host=activity.gmall.com
- id: service-activity
uri: lb://service-activity
predicates:
- Path=/*/activity/** # 路径匹配
(3)秒杀商品导入缓存
缓存数据实现思路:service-task模块统一管理我们的定时任务,为了减少service-task模块的耦合度,我们可以在定时任务模块只发送mq消息,需要执行定时任务的模块监听该消息即可,这样有利于我们后期动态控制,例如:每天凌晨一点我们发送定时任务信息到mq交换机,如果秒杀业务凌晨一点需要导入数据到缓存,那么秒杀业务绑定队列到交换机就可以了,其他业务也是一样,这样就做到了很好的扩展。
上面提到我们要控制库存数量,不能超卖,那么如何控制呢?在这里我们提供一种解决方案,那就我们在导入商品缓存数据时,同时将商品库存信息导入队列{list},利用redis队列的原子性,保证库存不超卖
库存加入队列实施方案
- 如果秒杀商品有N个库存,那么我就循环往队列放入N个队列数据
- 秒杀开始时,用户进入,然后就从队列里面出队,只有队列里面有数据,说明就一点有库存(redis队列保证了原子性),队列为空了说明商品售罄
编写定时任务
在service-task模块发送消息
搭建service-task服务
搭建方式如service-mq
修改配置pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.atguigu.gmall</groupId>
<artifactId>service</artifactId>
<version>1.0</version>
</parent>
<artifactId>service-task</artifactId>
<version>1.0</version>
<packaging>jar</packaging>
<name>service-task</name>
<description>service-task</description>
<dependencies>
<!--rabbitmq消息队列-->
<dependency>
<groupId>com.atguigu.gmall</groupId>
<artifactId>rabbit-util</artifactId>
<version>1.0</version>
</dependency>
</dependencies>
<build>
<finalName>service-task</finalName>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
添加配置文件以及启动类
bootstrap.properties
spring.application.name=service-task
spring.profiles.active=dev
spring.cloud.nacos.discovery.server-addr=192.168.200.129:8848
spring.cloud.nacos.config.server-addr=192.168.200.129:8848
spring.cloud.nacos.config.prefix=${spring.application.name}
spring.cloud.nacos.config.file-extension=yaml
spring.cloud.nacos.config.shared-configs[0].data-id=common.yaml
启动类
package com.atguigu.gmall.task;
@SpringBootApplication(exclude = DataSourceAutoConfiguration.class)//取消数据源自动配置
@ComponentScan({"com.atguigu.gmall"})
@EnableDiscoveryClient
public class ServiceTaskApplication {
public static void main(String[] args) {
SpringApplication.run(ServiceTaskApplication.class, args);
}
}
添加定时任务
定义凌晨一点mq相关常量
/**
* 定时任务
*/
public static final String EXCHANGE_DIRECT_TASK = "exchange.direct.task";
public static final String ROUTING_TASK_1 = "seckill.task.1";
//队列
public static final String QUEUE_TASK_1 = "queue.task.1";
package com.atguigu.gmall.task.scheduled;
@Component
@EnableScheduling //开启定时任务的支持
@Slf4j
public class ScheduledTask {
@Autowired
private RabbitService rabbitService;//定义了消息发送的发放,下面直接调用
/**
* 每天凌晨1点执行
*
*什么时候执行注解@Scheduled
*参数值:秒 分 时 日 月 星期 年
* *:任何时间 ?:日和星期
*/
//@Scheduled(cron = "0/30 * * * * ?")
@Scheduled(cron = "0 0 1 * * ?")
public void task1() {
rabbitService.sendMessage(MqConst.EXCHANGE_DIRECT_TASK, MqConst.ROUTING_TASK_1, "");
}
}
cron表达式各占位符解释:
{秒数} {分钟} {小时} {日期} {月份} {星期} {年份(可为空)}
{秒数}{分钟} ==> 允许值范围: 0~59 ,不允许为空值,若值不合法,调度器将抛出SchedulerException异常
“*” 代表每隔1秒钟触发;
“,” 代表在指定的秒数触发,比如”0,15,45”代表0秒、15秒和45秒时触发任务
“-“代表在指定的范围内触发,比如”25-45”代表从25秒开始触发到45秒结束触发,每隔1秒触发1次
“/”代表触发步进(step),”/”前面的值代表初始值(““等同”0”),后面的值代表偏移量,比如”0/20”或者”/20”代表从0秒钟开始,每隔20秒钟触发1次,即0秒触发1次,20秒触发1次,40秒触发1次;”5/20”代表5秒触发1次,25秒触发1次,45秒触发1次;”10-45/20”代表在[10,45]内步进20秒命中的时间点触发,即10秒触发1次,30秒触发1次
{小时} ==> 允许值范围: 0~23 ,不允许为空值,若值不合法,调度器将抛出SchedulerException异常,占位符和秒数一样
{日期} ==> 允许值范围: 1~31 ,不允许为空值,若值不合法,调度器将抛出SchedulerException异常
{星期} ==> 允许值范围: 1~7 (SUN-SAT),1代表星期天(一星期的第一天),以此类推,7代表星期六(一星期的最后一天),不允许为空值,若值不合法,调度器将抛出SchedulerException异常
{年份} ==> 允许值范围: 1970~2099 ,允许为空,若值不合法,调度器将抛出SchedulerException异常
注意:日期和星期的问题
日期跟星期互斥,如果重视日期需要把星期打? 如果重视星期,把日期打?
?"与{日期}互斥,即意味着若明确指定{日期}触发,则表示{星期}无意义,以免引起冲突和混乱
常用实例:
“30 * * * * ?” 每半分钟触发任务
“30 10 * * * ?” 每小时的10分30秒触发任务
“30 10 1 * * ?” 每天1点10分30秒触发任务
“30 10 1 20 * ?” 每月20号1点10分30秒触发任务
“30 10 1 20 10 ? *” 每年10月20号1点10分30秒触发任务
“30 10 1 20 10 ? 2011” 2011年10月20号1点10分30秒触发任务
“30 10 1 ? 10 * 2011” 2011年10月每天1点10分30秒触发任务
“30 10 1 ? 10 SUN 2011” 2011年10月每周日1点10分30秒触发任务
“15,30,45 * * * * ?” 每15秒,30秒,45秒时触发任务
“15-45 * * * * ?” 15到45秒内,每秒都触发任务
“15/5 * * * * ?” 每分钟的每15秒开始触发,每隔5秒触发一次
“15-30/5 * * * * ?” 每分钟的15秒到30秒之间开始触发,每隔5秒触发一次
“0 0/3 * * * ?” 每小时的第0分0秒开始,每三分钟触发一次
“0 15 10 ? * MON-FRI” 星期一到星期五的10点15分0秒触发任务
“0 15 10 L * ?” 每个月最后一天的10点15分0秒触发任务
“0 15 10 LW * ?” 每个月最后一个工作日的10点15分0秒触发任务
“0 15 10 ? * 5L” 每个月最后一个星期四的10点15分0秒触发任务
“0 15 10 ? * 5#3” 每个月第三周的星期四的10点15分0秒触发任务
监听定时任务信息
在service-activity模块绑定与监听消息,处理 ,更新状态位
数据导入缓存
在service-util的RedisConst类中定义常量
//秒杀商品前缀
public static final String SECKILL_GOODS = "seckill:goods";
public static final String SECKILL_ORDERS = "seckill:orders";
public static final String SECKILL_ORDERS_USERS = "seckill:orders:users";
public static final String SECKILL_STOCK_PREFIX = "seckill:stock:";
public static final String SECKILL_USER = "seckill:user:";
//用户锁定时间 单位:秒
public static final int SECKILL__TIMEOUT = 60 * 60;
创建秒杀商品实体与Mapper
package com.atguigu.gmall.model.activity;
@Data
@ApiModel(description = "SeckillGoods")
@TableName("seckill_goods")
public class SeckillGoods extends BaseEntity {
private static final long serialVersionUID = 1L;
@ApiModelProperty(value = "spu ID")
@TableField("spu_id")
private Long spuId;
@ApiModelProperty(value = "sku ID")
@TableField("sku_id")
private Long skuId;
@ApiModelProperty(value = "标题")
@TableField("sku_name")
private String skuName;
@ApiModelProperty(value = "商品图片")
@TableField("sku_default_img")
private String skuDefaultImg;
@ApiModelProperty(value = "原价格")
@TableField("price")
private BigDecimal price;
@ApiModelProperty(value = "秒杀价格")
@TableField("cost_price")
private BigDecimal costPrice;
@ApiModelProperty(value = "添加日期")
@TableField("create_time")
private Date createTime;
@ApiModelProperty(value = "审核日期")
@TableField("check_time")
private Date checkTime;
@ApiModelProperty(value = "审核状态")
@TableField("status")
private String status;
@ApiModelProperty(value = "开始时间")
@TableField("start_time")
private Date startTime;
@ApiModelProperty(value = "结束时间")
@TableField("end_time")
private Date endTime;
@ApiModelProperty(value = "秒杀商品数")
@TableField("num")
private Integer num;
@ApiModelProperty(value = "剩余库存数")
@TableField("stock_count")
private Integer stockCount;
@ApiModelProperty(value = "描述")
@TableField("sku_desc")
private String skuDesc;
}
package com.atguigu.gmall.activity.mapper;
@Mapper
public interface SeckillGoodsMapper extends BaseMapper<SeckillGoods> {
}
监听消息
导入工具包{redis,util}到service-activity 项目中!
package com.atguigu.gmall.activity.receiver;
@Component
public class SeckillReceiver {
@Autowired
private RedisTemplate redisTemplate;
@Autowired
private SeckillGoodsMapper seckillGoodsMapper;
@SneakyThrows
@RabbitListener(bindings = @QueueBinding(
value = @Queue(value = MqConst.QUEUE_TASK_1,durable = "true",autoDelete = "false"),
exchange = @Exchange(value = MqConst.EXCHANGE_DIRECT_TASK),
key = {MqConst.ROUTING_TASK_1}
))
public void importToRedis(Message message, Channel channel){
try {
//查询Mysql中符合条件的数据 时间 库存 状态
// 将当天的秒杀商品放入缓存!通过mapper 执行sql 语句!
// 条件当天 ,剩余库存>0 , 审核状态 = 1
QueryWrapper<SeckillGoods> seckillGoodsQueryWrapper = new QueryWrapper<>();
//状态库存
seckillGoodsQueryWrapper.eq("status","1").gt("stock_count",0);
// select DATE_FORMAT(start_time,'%Y-%m-%d') from seckill_goods; yyyy-mm-dd
//时间
seckillGoodsQueryWrapper.eq("DATE_FORMAT(start_time,'%Y-%m-%d')", DateUtil.formatDate(new Date()));
// 获取到当天秒杀的商品列表!
List<SeckillGoods> seckillGoodsList = seckillGoodsMapper.selectList(seckillGoodsQueryWrapper);
// 将seckillGoodsList 这个集合数据放入缓存!
for (SeckillGoods seckillGoods : seckillGoodsList) {
// 考虑使用哪种数据类型,以及缓存的key!使用hash! hset key field value hget key field
// 定义key = SECKILL_GOODS field = skuId value = seckillGoods
// 判断当前缓存key 中是否有 秒杀商品的skuId
Boolean flag = redisTemplate.boundHashOps(RedisConst.SECKILL_GOODS).hasKey(seckillGoods.getSkuId().toString());
// 判断
if (flag){
// 表示缓存中已经当前的商品了。
continue;
}
// 没有就放入缓存!
redisTemplate.boundHashOps(RedisConst.SECKILL_GOODS).put(seckillGoods.getSkuId().toString(),seckillGoods);
//防止库存超卖,存储Redis的List
// 将每个商品对应的库存剩余数,放入redis-list 集合中!
for (Integer i = 0; i < seckillGoods.getStockCount(); i++) {
// 放入list key = seckill:stock:skuId;
String key = RedisConst.SECKILL_STOCK_PREFIX+seckillGoods.getSkuId();
redisTemplate.opsForList().leftPush(key,seckillGoods.getSkuId().toString());
// redisTemplate.boundListOps(key).leftPush(seckillGoods.getSkuId());
}
// 秒杀商品在初始化的时候:状态位初始化 1(可以抢购) seckillpush后面定义的发送消息主题
// publish seckillpush 46:1 | 后续业务如果说商品被秒杀完了! publish seckillpush 46:0
redisTemplate.convertAndSend("seckillpush",seckillGoods.getSkuId()+":1");
}
} catch (Exception e) {
e.printStackTrace();
}
// 手动确认消息
//消息确认 // 参数一:消息的唯一标识,参数二:是否批量确认 false 确认一个消息,true 批量确认
channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
}
}
(4)redis发布与订阅实现
更新状态位
由于我们的秒杀服务时集群部署service-activity的,我们面临一个问题?RabbitMQ 如何实现对同一个应用的多个节点进行广播呢?
RabbitMQ 只能对绑定到交换机上面的不同队列实现广播,对于同一队列的消费者他们存在竞争关系,同一个消息只会被同一个队列下其中一个消费者接收,达不到广播效果;
我们目前的需求是定时任务发送消息,我们将秒杀商品导入缓存,同事更新集群的状态位,既然RabbitMQ 达不到广播的效果,我们就放弃吗?当然不是,我们很容易就想到一种解决方案,通过redis的发布订阅模式来通知其他兄弟节点,这不问题就解决了吗?
过程大致如下
应用启动,多个节点监听同一个队列(此时多个节点是竞争关系,一条消息只会发到其中一个节点上)
消息生产者发送消息,同一条消息只被其中一个节点收到
收到消息的节点通过redis的发布订阅模式来通知其他兄弟节点
这两个订阅
第一个发送
另外两个都接受到了
接下来配置redis发布与订阅:代码实现
package com.atguigu.gmall.activity.redis;
@Configuration
public class RedisChannelConfig {
/*
docker exec -it e222dac4e559 redis-cli
subscribe seckillpush // 订阅 接收消息
publish seckillpush admin // 发布消息
*/
/**
*监听器
*
* 注入订阅主题
* @param connectionFactory redis 链接工厂
* @param listenerAdapter 消息监听适配器
* @return 订阅主题对象
*/
@Bean
RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
MessageListenerAdapter listenerAdapter) {
//连接对象
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
//谁知工厂
container.setConnectionFactory(connectionFactory);
//设置适配器 订阅主题
container.addMessageListener(listenerAdapter, new PatternTopic("seckillpush"));
//这个container 可以添加多个 messageListener
return container;
}
/**
* 返回消息适配器
* @param receiver 创建接收消息对象
* @return 适配器
*/
@Bean
MessageListenerAdapter listenerAdapter(MessageReceive receiver) {
//这个地方 是给 messageListenerAdapter 传入一个消息接受的处理器,利用反射的方法调用“receiveMessage”
//也有好几个重载方法,这边默认调用处理器的方法 叫handleMessage 可以自己到源码里面看
//参数一:处理器 参数二:处理的方法 这里是通过反射方式执行处理器
return new MessageListenerAdapter(receiver, "receiveMessage");
}
@Bean //注入操作数据的template
StringRedisTemplate template(RedisConnectionFactory connectionFactory) {
return new StringRedisTemplate(connectionFactory);
}
}
StringRedisTemplate,这个有 重载的工厂的方法
消息处理器:
package com.atguigu.gmall.activity.redis;
@Component
public class MessageReceive {
/**接收消息的方法*/
public void receiveMessage(String message){
System.out.println("----------收到消息了message:"+message);
if(!StringUtils.isEmpty(message)) {
/*
消息格式
skuId:0 表示没有商品
skuId:1 表示有商品
*/
// 因为传递过来的数据为 ""6:1""
message = message.replaceAll("\"","");
String[] split = StringUtils.split(message, ":");
if (split == null || split.length == 2) {
CacheHelper.put(split[0], split[1]);
}
}
}
}
CacheHelper类本地缓存类
package com.atguigu.gmall.activity.util;
/**
* 系统缓存类
*/
public class CacheHelper {
/**
* 缓存容器 ConcurrentHashMap这个是线程安全的
*/
private final static Map<String, Object> cacheMap = new ConcurrentHashMap<String, Object>();
/**
* 加入缓存
*
* @param key
* @param cacheObject
*/
public static void put(String key, Object cacheObject) {
cacheMap.put(key, cacheObject);
}
/**
* 获取缓存
* @param key
* @return
*/
public static Object get(String key) {
return cacheMap.get(key);
}
/**
* 清除缓存
*
* @param key
* @return
*/
public static void remove(String key) {
cacheMap.remove(key);
}
public static synchronized void removeAll() {
cacheMap.clear();
}
}
说明:
- RedisChannelConfig 类配置redis监听的主题和消息处理器
- MessageReceive 类为消息处理器,消息message为:商品id与状态位,如:1:1 表示商品id为1,状态位为1
redis发布消息
监听已经配置好,接下来我就发布消息,更改秒杀监听器{ SeckillReceiver },如下
完整代码如下
@RabbitListener(bindings = @QueueBinding(
value = @Queue(value = MqConst.QUEUE_TASK_1, durable = "true"),
exchange = @Exchange(value = MqConst.EXCHANGE_DIRECT_TASK, type = ExchangeTypes.DIRECT, durable = "true"),
key = {MqConst.ROUTING_TASK_1}
))
public void importItemToRedis(Message message, Channel channel) throws IOException {
//Log.info("importItemToRedis:");
QueryWrapper<SeckillGoods> queryWrapper = new QueryWrapper<>();
queryWrapper.eq("status", 1);
queryWrapper.gt("stock_count", 0);
//当天的秒杀商品导入缓存
queryWrapper.eq("DATE_FORMAT(start_time,'%Y-%m-%d')", DateUtil.formatDate(new Date()));
List<SeckillGoods> list = seckillGoodsMapper.selectList(queryWrapper);
//把数据放在redis中
for (SeckillGoods seckillGoods : list) {
if (redisTemplate.boundHashOps(RedisConst.SECKILL_GOODS).hasKey(seckillGoods.getSkuId().toString()))
continue;
redisTemplate.boundHashOps(RedisConst.SECKILL_GOODS).put(seckillGoods.getSkuId().toString(), seckillGoods);
//根据每一个商品的数量把商品按队列的形式放进redis中
for (int i = 0; i < seckillGoods.getStockCount(); i++) {
redisTemplate.boundListOps(RedisConst.SECKILL_STOCK_PREFIX + seckillGoods.getSkuId()).leftPush(seckillGoods.getSkuId().toString());
}
//通知添加与更新状态位,更新为开启
redisTemplate.convertAndSend("seckillpush", seckillGoods.getSkuId()+":1");
}
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}
说明:到目前我们就实现了商品信息导入缓存,同时更新状态位的工作