一、Redis优化秒杀

1. 压测秒杀业务

• 模拟1000请求，秒杀100张优惠券
• 1000个不同请求，则需要1000个不同token

编写测试方法生成token，存到token.txt 文件中：

@SpringBootTest
public class GenerateTokenTest {

    @Resource
    private UserServiceImpl userService;

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Test
    public void test() throws IOException {
        List<User> userList = userService.query().list();
        FileWriter fw = new FileWriter("e:/token.txt");
        BufferedWriter bw = new BufferedWriter(fw);
        for (User user : userList) {
            String token = UUID.randomUUID().toString(true);
            UserDTO userDTO = BeanUtil.copyProperties(user, UserDTO.class);
            Map<String, Object> userMap = BeanUtil.beanToMap(userDTO, new HashMap<>(),
                CopyOptions.create()
                    .setIgnoreNullValue(true)
                    .setFieldValueEditor((fieldName, fieldValue) -> fieldValue.toString()));
            stringRedisTemplate.opsForHash().putAll(RedisConstants.LOGIN_USER_KEY + token, userMap);
            stringRedisTemplate.expire(RedisConstants.LOGIN_USER_KEY + token, RedisConstants.LOGIN_USER_TTL, TimeUnit.MINUTES);

            bw.write(token);
            bw.newLine(); // 换行符
            bw.flush();  // 清除缓存
        }
        // 关闭
        bw.close();
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

生成结果：

使用JMeter测试：

测试结果：

• 由结果可知，平均耗时1675毫秒，耗时较长，吞吐量362.2/s，吞吐量也不高

2. 优化秒杀思路

• 没优化之前的业务流程都是串行执行的，并且还会频繁和数据库交互，所以这就是耗时长，吞吐量不高的原因
  

• 判断秒杀库存、校验一人一单的业务都是查询操作，耗时较短，可以将该业务放到Redis中执行

• 减库存、创建订单业务会涉及到数据库的写操作，耗时长，可以单独开一个线程处理

• 主线程负责校验，校验成功后将订单id存到阻塞队列，并且将订单id返回给用户，用户可以接着执行其他业务

• 新启的线程异步读取阻塞队列中的信息，判断是否需要创建订单
  

• 由上面分析可知，需要提前将库存信息和优惠券订单信息缓存到Redis

• 库存信息可以用String类型存储，只要库存大于0就表示有库存；判断完之后还需要在Redis中预减库存；

• 优惠券订单信息需要用set集合存储，key是优惠券订单id，value是用户id，通过判断value是否存在就知道该用户是否下单过，保证了一人一单

• 为了确保判断秒杀库存、校验一人一单的原子性，所以需要使用Lua脚本
  

3. 改进秒杀业务

需求：

• 新增秒杀优惠券的同时，将优惠券信息保存到Redis中
• 基于Lua脚本，判断秒杀库存、一人一单，决定用户是否抢购成功
• 如果抢购成功，将优惠券id和用户id封装后存入阻塞队列
• 开启线程任务，不断从阻塞队列中获取信息，实现异步下单功能

新增秒杀优惠券的同时，将优惠券信息保存到Redis中：

@Override
@Transactional
public void addSeckillVoucher(Voucher voucher) {
    // 保存优惠券
    save(voucher);
    // 保存秒杀信息
    SeckillVoucher seckillVoucher = new SeckillVoucher();
    seckillVoucher.setVoucherId(voucher.getId());
    seckillVoucher.setStock(voucher.getStock());
    seckillVoucher.setBeginTime(voucher.getBeginTime());
    seckillVoucher.setEndTime(voucher.getEndTime());
    seckillVoucherService.save(seckillVoucher);
    // 保存秒杀库存到Redis中
    stringRedisTemplate.opsForValue().set(SECKILL_STOCK_KEY + voucher.getId(), voucher.getStock().toString());
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

• 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，2个..表示拼接字符串
local stockKey = 'seckill:stock:' .. voucherId
-- 2.2.订单key
local orderKey = 'seckill:order:' .. voucherId

-- 3.脚本业务
-- 3.1.判断库存是否充足 get stockKey
-- tonumber是将字符串转为数字
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
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

改进的代码：

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {

    @Resource
    private ISeckillVoucherService seckillVoucherService;

    @Resource
    private RedisIdWorker redisIdWorker;

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Resource
    private RedissonClient redissonClient;

    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;

    static {
        SECKILL_SCRIPT = new DefaultRedisScript<>();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);
    }

    // 创建阻塞队列
    private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);
    // 用线程池创建独立线程
    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();

    // 项目启动后，就应该开启线程，异步从阻塞队列中获取信息
    @PostConstruct
    private void init() {
        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
    }

    private class VoucherOrderHandler implements Runnable {

        @Override
        public void run() {
            while (true) {
                try {
                    // 1.获取队列中的订单信息
                    VoucherOrder voucherOrder = orderTasks.take();
                    // 2.创建订单
                    handleVoucherOrder(voucherOrder);
                } catch (Exception e) {
                    log.error("处理订单异常", e);
                }
            }
        }
    }

    // 将代理对象作为全局变量，供所有线程使用
    private IVoucherOrderService proxy;

    private void handleVoucherOrder(VoucherOrder voucherOrder) {
        // 1.获取用户，注意，这里是单独的线程，所以不能从主线程的ThreadLocal获取userId
        Long userId = voucherOrder.getUserId();
        // 2.创建锁对象
        RLock redisLock = redissonClient.getLock("lock:order:" + userId);
        // 3.尝试获取锁，这里加锁是兜底方案，可以不用再加锁，因为前面执行过lua脚本校验过一人一单
        boolean isLock = redisLock.tryLock();
        // 4.判断是否获得锁成功
        if (!isLock) {
            // 获取锁失败，直接返回失败或者重试
            log.error("不允许重复下单！");
            return;
        }
        try {
            // 注意：不能通过 IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy(); 获取代理对象
            // 因为AopContext.currentProxy();底层内部也有个ThreadLocal，但是此时的线程是新开启的线程，所以不能获取不到主线程中的代理对象
            // 所以需要在主线程中先获取到代理对象，保存到全局变量供所有线程使用
            proxy.createVoucherOrder(voucherOrder);
        } finally {
            // 释放锁
            redisLock.unlock();
        }
    }

    @Override
    public Result seckillVoucher(Long voucherId) {
        Long userId = UserHolder.getUser().getId();
        long orderId = redisIdWorker.nextId("order");
        // 1.执行lua脚本
        Long result = stringRedisTemplate.execute(
                SECKILL_SCRIPT,
                Collections.emptyList(),
                voucherId.toString(), userId.toString(), String.valueOf(orderId)
        );
        int r = result.intValue();
        // 2.判断结果是否为0
        if (r != 0) {
            // 2.1.不为0 ，代表没有购买资格
            return Result.fail(r == 1 ? "库存不足" : "不能重复下单");
        }
        // 2.2 为0 ，代表有购买资格，把下单信息保存到阻塞队列
        VoucherOrder voucherOrder = new VoucherOrder();
        // 2.3.订单id
        voucherOrder.setId(orderId);
        // 2.4.用户id
        voucherOrder.setUserId(userId);
        // 2.5.代金券id
        voucherOrder.setVoucherId(voucherId);
        // 2.6.放入阻塞队列
        orderTasks.add(voucherOrder);
        // 3.在主线程中获取代理对象
        proxy = (IVoucherOrderService) AopContext.currentProxy();
        // 4.返回订单id
        return Result.ok(orderId);
    }

    @Transactional
    @Override
    public void createVoucherOrder(VoucherOrder voucherOrder) {

        Long userId = voucherOrder.getUserId();
        // 5.1.查询订单
        int count = query().eq("user_id", userId).eq("voucher_id", voucherOrder.getVoucherId()).count();
        // 5.2.判断是否存在
        if (count > 0) {
            // 用户已经购买过了
            log.error("用户已经购买过了");
            return;
        }

        // 6.扣减库存
        boolean success = seckillVoucherService.update()
                .setSql("stock = stock - 1") // set stock = stock - 1
                .eq("voucher_id", voucherOrder.getVoucherId()).gt("stock", 0) // where id = ? and stock > 0
                .update();
        if (!success) {
            // 扣减失败
            log.error("库存不足");
            return;
        }
        // 7.创建订单
        save(voucherOrder);
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

测试结果：

• 相比优化前，平均耗时缩短，吞吐量提高

4. 小结

秒杀业务的优化思路是什么？

• 先利用Redis完成库存余量、一人一单判断，完成抢单业务
• 再将下单业务放入阻塞队列，利用独立线程异步下单

基于阻塞队列的异步秒杀存在哪些问题？

• 内存限制问题（如果不对BlockingQueue做大小限制，则会有内存溢出问题）
• 数据安全问题（如果服务宕机，则内存的数据将会丢失）