redis、RabbitMq整合【四 优惠卷秒杀（下）——集群下的线程并发问题，秒杀优化（异步） 】
·
前言：
👏作者简介：我是笑霸final，一名热爱技术的在校学生。
📝个人主页：个人主页1 || 笑霸final的主页2
📕系列专栏：《项目专栏》
📧如果文章知识点有错误的地方，请指正！和大家一起学习，一起进步👀
🔥如果感觉博主的文章还不错的话，👍点赞👍 + 👀关注👀 + 🤏收藏🤏
集群下的线程并发问题目录
🐉上节回顾🐉
上节讲了 乐观锁和悲观锁，并实现了乐观锁。
仿黑马点评-redis整合【四 优惠卷秒杀（上） 】
分布式锁
不同的分布式锁实现方案
基于redis（setnx）的分布式锁
实现分布式锁需要两个基本方法
- 索取锁
SETNX lock thread1
互斥：确保只有一个线程获取锁EXPIRE lock 10设置过期时间 【EXPIRE key 时间】单位秒
想一想：如果还没来得及执行设置过期时间就宕机了怎么办？使用下面的命令来解决SET lock thread EX 10 NX
- 释放锁
DEL lock
手动释放、超时释放
流程图
redis分布式锁初级版本1
public interface ILock {
/**
* 尝试获取锁
* @param timeoutSec
* @return
*/
boolean tryLock(Long timeoutSec);
void unLock();
}
设计一个类SimpleRedisLock去实现ILock接口
public class SimpleRedisLock implements ILock{
private StringRedisTemplate stringRedisTemplate;
//不同的业务使用不同的锁
private String name;
private static final String KEY_PREFIX="lock:";
public SimpleRedisLock(StringRedisTemplate stringRedisTemplate, String name) {
this.stringRedisTemplate = stringRedisTemplate;
this.name = name;
}
@Override
public boolean tryLock(Long timeoutSec) {
//获取线程的标识
long threadId = Thread.currentThread().getId();
Boolean success = stringRedisTemplate
.opsForValue()
.setIfAbsent(KEY_PREFIX + name, threadId + "", timeoutSec, TimeUnit.SECONDS);
return Boolean.TRUE.equals(success);//拆箱防止空指针
}
@Override
public void unLock() {
stringRedisTemplate.delete(KEY_PREFIX + name);
}
}
调用
@Override
public Result seckillVoucher(Long voucherId) {
//1.查询id
SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
//2.判断是否开始
if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
//没有开始
return Result.fail("时间还没开始");
}
//3.判断是否结束
if (voucher.getEndTime().isBefore(LocalDateTime.now())) {
//已经结束
return Result.fail("已经结束");
}
//4.判断库存是否充足
if (voucher.getStock()<1) {
//库存不足
return Result.fail("库存不足");
}
Long userId = UserHolder.getUser().getId();
//intern()在常量池里先去找一样的地址返回
// synchronized(userId.toString().intern()) {
// VoucherOrderServiceImpl proxy
// =(VoucherOrderServiceImpl)AopContext.currentProxy();//代理对象
// //因为没有加事务 事务用的代理对象 可能存在事务失效
// //没有用代理对象默认是 this.creteVoucherOrder(voucherId)
// return proxy.creteVoucherOrder(voucherId);
// }
//尝试自己加锁
//创建锁对象
SimpleRedisLock lock =
new SimpleRedisLock(stringRedisTemplate, "order:" + userId);
//获取锁
boolean isLock = lock.tryLock(5L);
//判断是否获取成功
if (!isLock) {
return Result.fail("不允许重复下单");
}
try{
VoucherOrderServiceImpl proxy
=(VoucherOrderServiceImpl)AopContext.currentProxy();//代理对象
//因为没有加事务 事务用的代理对象 可能存在事务失效
//没有用代理对象默认是 this.creteVoucherOrder(voucherId)
return proxy.creteVoucherOrder(voucherId);
}finally {
lock.unLock();
}
}
redis分布式锁初级版本2
普通setnx分布式锁出现的问题
- 在某个线程获取锁执行业务时若发生阻塞，且阻塞过程中锁超时，此时另一个线程同样来请求锁，发现可以获取锁，但实际上前一个线程还没执行完。
解决方案：
- 在获取锁时存入线程标示（可以用UUID表示）
- 在释放锁时先获取锁中的线程标示，判断是否与当前线程标示一致
- 如果一致则释放锁
- 如果不一致则不释放锁
更新后的流程图
代码
public class SimpleRedisLock implements ILock{
private StringRedisTemplate stringRedisTemplate;
//不同的业务使用不同的锁
private String name;
private static final String KEY_PREFIX="lock:";
private static final String ID_PREFIX= UUID.fastUUID().toString(true)+"-";
public SimpleRedisLock(StringRedisTemplate stringRedisTemplate, String name) {
this.stringRedisTemplate = stringRedisTemplate;
this.name = name;
}
@Override
public boolean tryLock(Long timeoutSec) {
//获取线程的标识
String threadId = ID_PREFIX+Thread.currentThread().getId();
Boolean success = stringRedisTemplate
.opsForValue()
.setIfAbsent(KEY_PREFIX + name, threadId , timeoutSec, TimeUnit.SECONDS);
return Boolean.TRUE.equals(success);//拆箱防止空指针
}
@Override
public void unLock() {
//获取线程标识
String threadId = ID_PREFIX+Thread.currentThread().getId();
//获取锁中的标识
String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
//判断
if (threadId.equals(id)){
stringRedisTemplate.delete(KEY_PREFIX + name);
}
}
}
类加载的顺序：
- 1.加载静态成员/代码块：
先递归地加载父类的静态成员/代码块(Object的最先)；再依次加载到本类的静态成员。
同一个类里的静态成员/代码块，按写代码的顺序加载。
如果其间调用静态方法，则调用时会先运行静态方法，再继续加载。同一个类里调用静态方法时，可以不理会写代码的顺序。
调用父类的静态成员，可以像调用自己的一样；但调用其子类的静态成员，必须使用“子类名.成员名”来调用。 - 2.加载非静态成员/代码块：（实例块在创建对象时才会被加载。而静态成员在不创建对象时可以加载）
先递归地加载父类的非静态成员/代码块(Object的最先)；再依次加载到本类的非静态成员。
同一个类里的非静态成员/代码块，按写代码的顺序加载。同一个类里调用方法时，可以不理会写代码的顺序。
但调用属性时，必须注意加载顺序。一般编译不通过，如果能在加载前调用，值为默认初始值(如：null 或者 0)。
调用父类的非静态成员(private 除外)，也可以像调用自己的一样。 - 3.调用构造方法：
先递归地调用父类的构造方法(Object的最先)也就是上溯下行；默认调用父类空参的，也可在第一行写明调用父类某个带参的。
再依次到本类的构造方法；构造方法内，也可在第一行写明调用某个本类其它的构造方法。 - 注意：如果加载时遇到 override 的成员，可看作是所需创建的类型赋值给当前类型。
其调用按多态用法：只有非静态方法有多态；而静态方法、静态属性、非静态属性都没有多态。
假设子类override父类的所有成员，包括静态成员、非静态属性和非静态方法。
由于构造子类时会先构造父类；而构造父类时，其所用的静态成员和非静态属性是父类的，但非静态方法却是子类的；
由于构造父类时，子类并未加载；如果此时所调用的非静态方法里有成员，则这个成员是子类的，且非静态属性是默认初始值的。
java调用lua脚本改造分布式锁
lua脚本
-- 比较线程标示与锁中的标示是否一致
if(redis.call('get', KEYS[1]) == ARGV[1]) then
-- 释放锁 del key
return redis.call('del', KEYS[1])
end
return 0
// @Override
// public void unLock() {
// //获取线程标识
// String threadId = ID_PREFIX+Thread.currentThread().getId();
// //获取锁中的标识
// String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
// //判断
// if (threadId.equals(id)){
// stringRedisTemplate.delete(KEY_PREFIX + name);
// }
// }
//提前加载lua脚本
public static final DefaultRedisScript<Long> UNLOCK_SCRIPT;
static {
UNLOCK_SCRIPT=new DefaultRedisScript<>();
UNLOCK_SCRIPT.setLocation(new ClassPathResource("un;ock.lua"));//加载文件
UNLOCK_SCRIPT.setResultType(Long.class);//设置返回类型
}
/**
* 基于lua脚本
*/
@Override
public void unLock() {
//调用脚本
List<String> list=new ArrayList<>();
list.add(KEY_PREFIX + name);
stringRedisTemplate.execute(UNLOCK_SCRIPT,list,
ID_PREFIX+Thread.currentThread().getId());
}
基于redis（setnx）的分布式锁
setnx遇到的问题，Redisson的引入
Redisson是架设在Redis基础上的一个Java驻内存数据网格（In-Memory Data Grid）。充分的利用了Redis键值数据库提供的一系列优势，基于Java实用工具包中常用接口，为使用者提供了一系列具有分布式特性的常用工具类。使得原本作为协调单机多线程并发程序的工具包获得了协调分布式多机多线程并发系统的能力，大大降低了设计和研发大规模分布式系统的难度。同时结合各富特色的分布式服务，更进一步简化了分布式环境中程序相互之间的协作。
Redisson官网 点此跳转 ：
https://redisson.org/
github地址 点此跳转：https://github.com/redisson/redisson
Redisson入门
Redisson官网 点此跳转 ：
https://redisson.org/
github地址 点此跳转：https://github.com/redisson/redisson
我们看官方的快速入门
1.导入坐标
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.18.0</version>
</dependency>
2.配置Redisson客户端
@Configuration
public class RedissonConfig {
@Bean
public RedissonClient redissonClient(){
//配置类
Config config=new Config();
//添加redis的地址
config.useSingleServer()
.setAddress("redis://101.200.230.213:6379")
.setPassword("0615");
return Redisson.create(config);
}
}
3.使用
@Resource
private RedissonClient redissonClient;
@Test
void testRedisson()throws InterruptedException{
//获取锁（可重入），指定名称
RLock lock =redissonClient.getLock("anyLock");
//尝试获取锁，参数分别是 获取锁的最大等待时间 ，锁自动释放时间，时间单位
boolean tryLock = lock.tryLock(1, 10, TimeUnit.SECONDS);
//判断锁 获取成功
if (tryLock) {
try {
log.info("执行的业务");
}finally {
//释放锁
lock.unlock();
}
}
}
分布式锁的代码
@Resource
private RedissonClient redissonClient;
@Override
public Result seckillVoucher(Long voucherId) {
//1.查询id
SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
//2.判断是否开始
if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
//没有开始
return Result.fail("时间还没开始");
}
//3.判断是否结束
if (voucher.getEndTime().isBefore(LocalDateTime.now())) {
//已经结束
return Result.fail("已经结束");
}
//4.判断库存是否充足
if (voucher.getStock()<1) {
//库存不足
return Result.fail("库存不足");
}
Long userId = UserHolder.getUser().getId();
RLock lock = redissonClient.getLock("order:" + userId);
//获取锁
boolean isLock = lock.tryLock();
//判断是否获取成功
if (!isLock) {
return Result.fail("不允许重复下单");
}
try{
VoucherOrderServiceImpl proxy
=(VoucherOrderServiceImpl)AopContext.currentProxy();//代理对象
//因为没有加事务 事务用的代理对象 可能存在事务失效
//没有用代理对象默认是 this.creteVoucherOrder(voucherId)
return proxy.creteVoucherOrder(voucherId);
}finally {
lock.unlock();
}
}
@Transactional//两张表 加上事务
public Result creteVoucherOrder(Long voucherId) {
//5一人一单
Long userId = UserHolder.getUser().getId();
//5.1查询订单
Integer count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
//5.2查询是否存在
if (count > 0) {
//存在就不能下单了
return Result.fail("你已经购买过了，只能买一次哟");
}
//6.扣减库存 乐观锁 在这一刻去判断更新时和查询到的库存是否一
boolean success = seckillVoucherService
.update()
.setSql("stock=stock-1")
.eq("voucher_id", voucherId)
// .eq("stock",voucher.getStock())
.gt("stock", 0)//判断 stock>0可行
.update();
if (!success) {
//库存不足
return Result.fail("库存不足");
}
//7.创建订单
VoucherOrder voucherOrder = new VoucherOrder();
//7.1订单id 用id全局唯一生成器
Long orderId = redisIdWorker.nexId("order");
voucherOrder.setId(orderId);
//7.2用户id
voucherOrder.setUserId(userId);
//7.3代金卷id
voucherOrder.setVoucherId(voucherId);
save(voucherOrder);
//8.返回id
return Result.ok(orderId);
}
🐉秒杀优化，异步秒杀🐉
分析
分析
查询 减库存 创建订单都走数据库，数据库并发能力较差
我们可以开多个线程来完成不同效率的步骤
redis优化流程
redis完成秒杀资格判断
1保存秒杀库存到redis中
代码：VoucherController
/**
* 新增秒杀券
* @param voucher 优惠券信息，包含秒杀信息
* @return 优惠券id
*/
@PostMapping("seckill")
public Result addSeckillVoucher(@RequestBody Voucher voucher) {
voucherService.addSeckillVoucher(voucher);
return Result.ok(voucher.getId());
}
2.基于lua脚本判断秒杀库存
lua脚本
-- 1.参数列表
-- 1.1.优惠券id
local voucherId = ARGV[1]
-- 1.2.用户id
local userId = ARGV[2]
-- 1.3.订单id
local orderId = ARGV[3]
-- 2.数据key
-- 2.1.库存key
local stockKey = 'seckill:stock:' .. voucherId
-- 2.2.订单key
local orderKey = 'seckill:order:' .. voucherId
-- 3.脚本业务
-- 3.1.判断库存是否充足 get stockKey
if(tonumber(redis.call('get', stockKey)) <= 0) then
-- 3.2.库存不足，返回1
return 1
end
-- 3.2.判断用户是否下单 SISMEMBER orderKey userId
if(redis.call('sismember', orderKey, userId) == 1) then
-- 3.3.存在，说明是重复下单，返回2
return 2
end
-- 3.4.扣库存 incrby stockKey -1
redis.call('incrby', stockKey, -1)
-- 3.5.下单（保存用户）sadd orderKey userId
redis.call('sadd', orderKey, userId)
-- 3.6.发送消息到队列中， XADD stream.orders * k1 v1 k2 v2 ...
redis.call('xadd', 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', orderId)
return 0
JAVA代码
public static final DefaultRedisScript<Long> SECKILL_SCRIPT;
static {
SECKILL_SCRIPT=new DefaultRedisScript<>();
SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));//加载文件
SECKILL_SCRIPT.setResultType(Long.class);//设置返回类型
}
/**
* 秒杀下单
* @param voucherId
* @return
*/
@Override
public Result seckillVoucher(Long voucherId) {
//获取用户
Long userId = UserHolder.getUser().getId();
//1.执行lua脚本
Long result = stringRedisTemplate
.execute(SECKILL_SCRIPT,
Collections.emptyList(),
voucherId.toString(),
userId
);
//2.判断结果是否为0
int value = result.intValue();
if(value!=0){
return Result.fail(value==1?"库存不足":"不能重复下单");
}
//2.2为零 有购买资格，把下单信息保存到阻塞队列
//.创建订单
VoucherOrder voucherOrder = new VoucherOrder();
//.1订单id 用id全局唯一生成器
Long orderId = redisIdWorker.nexId("order");
voucherOrder.setId(orderId);
//.2用户id
voucherOrder.setUserId(userId);
//.3代金卷id
voucherOrder.setVoucherId(voucherId);
blockingQueue.add(voucherOrder);//后面讲
//返回id
Long Id = redisIdWorker.nexId("order");
return Result.ok(Id);
}
阻塞队列 && 异步下单
阻塞队列
//阻塞队列
private BlockingQueue<VoucherOrder> blockingQueue
=new ArrayBlockingQueue<>(1024*1024);
异步下单
- 1.创建线程池
//线程池
private static final ExecutorService SECKILL_ORDER_EXECUTOR
= Executors.newSingleThreadExecutor();
- 2.异步执行
private class VoucherOrderHandler implements Runnable{
@Override
public void run() {
while(true){
//1.获取队列中得信息
try{
VoucherOrder take = blockingQueue.take();
save(take);//直接保存 不需要加锁
}catch (Exception e){
log.info("处理订单异常：原因"+e);
}
}
}
}
🐉整合RabbitMq实现异步秒杀🐉
上一节我们用jvm得阻塞队列，他其实也有问题，数据不能持久化！
导入依赖
<!--rabbitmq-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
启动类
加上
@EnableRabbit
配置类代码
声明x交换机声明y交换机声明队列A声明死信队列DA队列绑定X交换机d队列绑定y交换机
package com.hmdp.config;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import java.util.HashMap;
public class TtlQueueConfig {
//普通交换机名称
public static final String X_EXCHANGE="X";
//死信交换机名称
public static final String Y_DEAD_LETTER_EXCHANGE="Y";
//普通队列名称
public static final String QUEUE_A="QA";
//死信队列名称
public static final String DEAD_LETTER_QUEUE_D="QD";
/**
* 声明x交换机
* @return
*/
@Bean("xExchange")//别名和方法名取一样
public DirectExchange xExchange(){
return new DirectExchange(X_EXCHANGE);
}
/**
* 声明y交换机
* @return
*/
@Bean("yExchange")//别名和方法名取一样
public DirectExchange yExchange(){
return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
}
/**
* 声明队列A
* @return
*/
@Bean("queueA")
public Queue queueA(){
final HashMap<String, Object> arguments
= new HashMap<>();
//设置死信交换机
arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
//设置死信RoutingKey
arguments.put("x-dead-letter-routing-key","YD");
//设置TTL设置10秒过期
arguments.put("x-message-ttl",10000);
return QueueBuilder.durable(QUEUE_A)
.withArguments(arguments)
.build();
}
/**
* 声明死信队列D
* @return
*/
@Bean("queueD")
public Queue queueD(){
return QueueBuilder.durable(DEAD_LETTER_QUEUE_D)
.build();
}
/**
* A队列绑定X交换机
* @param queueA
* @return
*/
@Bean
public Binding queueABindingX(@Qualifier("queueA")Queue queueA,
@Qualifier("xExchange") DirectExchange xExchange){
return BindingBuilder.bind(queueA).to(xExchange).with("XA");
}
/**
* d队列绑定y交换机
* @param queueD
* @return
*/
@Bean
public Binding queueDBindingY(@Qualifier("queueD")Queue queueD,
@Qualifier("yExchange") DirectExchange yExchange
){
return BindingBuilder.bind(queueD).to(yExchange).with("YD");
}
}
生产者代码
@Resource
private RabbitTemplate rabbitTemplate;
/**
* 秒杀下单
* @param voucherId
* @return
*/
@Override
public Result seckillVoucher(Long voucherId) {
//获取用户
Long userId = UserHolder.getUser().getId();
//1.执行lua脚本
Long result = stringRedisTemplate
.execute(SECKILL_SCRIPT,
Collections.emptyList(),
voucherId.toString(),
userId
);
//2.判断结果是否为0
int value = result.intValue();
if(value!=0){
return Result.fail(value==1?"库存不足":"不能重复下单");
}
//2.2为零 有购买资格，把下单信息保存到阻塞队列
//.创建订单
VoucherOrder voucherOrder = new VoucherOrder();
//.1订单id 用id全局唯一生成器
Long orderId = redisIdWorker.nexId("order");
voucherOrder.setId(orderId);
//.2用户id
voucherOrder.setUserId(userId);
//.3代金卷id
voucherOrder.setVoucherId(voucherId);
//blockingQueue.add(voucherOrder);//后面讲
//放入mq
String jsonStr = JSONUtil.toJsonStr(voucherOrder);
rabbitTemplate.convertAndSend("X","XA",jsonStr );
//返回id
Long Id = redisIdWorker.nexId("order");
return Result.ok(Id);
}
消息消费者代码
/**
* sheng 消费者1
* @param message
* @param channel
* @throws Exception
*/
@RabbitListener(queues = "QA")
public void receivedA(Message message, Channel channel)throws Exception{
String msg=new String(message.getBody());
log.info("正常队列:");
VoucherOrder voucherOrder = JSONUtil.toBean(msg, VoucherOrder.class);
log.info(voucherOrder.toString());
save(voucherOrder);//保存到数据库
//数据库秒杀库存减一
Long voucherId=voucherOrder.getVoucherId();
boolean success = seckillVoucherService.update()
.setSql("stock = stock - 1") // set stock = stock - 1
.eq("voucher_id", voucherId).gt("stock", 0) // where id = ? and stock > 0
.update();
}
/**
* sheng 消费者2
* @param message
* @throws Exception
*/
@RabbitListener(queues = "QD")
public void receivedD(Message message)throws Exception{
log.info("死信队列:");
String msg=new String(message.getBody());
VoucherOrder voucherOrder = JSONUtil.toBean(msg, VoucherOrder.class);
log.info(voucherOrder.toString());
save(voucherOrder);
Long voucherId=voucherOrder.getVoucherId();
boolean success = seckillVoucherService.update()
.setSql("stock = stock - 1") // set stock = stock - 1
.eq("voucher_id", voucherId).gt("stock", 0) // where id = ? and stock > 0
.update();
}
华为开发者空间，是为全球开发者打造的专属开发空间，汇聚了华为优质开发资源及工具，致力于让每一位开发者拥有一台云主机，基于华为根生态开发、创新。
更多推荐
20
0- 0
已为社区贡献3条内容
所有评论(0)