优惠券秒杀

发表于： 2025-02-01 分类于： BlackHorse Rating

字数： 6854 阅读：≈ 14分钟

全局ID生成器: Redis自增

使用数据库自增ID的缺点：

可能会暴露给用户一些信息：用户可能会根据此推断，一天之内销售出了多少张优惠券。
当优惠券太多时，会导致数据库ID过大。但如果为优惠券分表，又会出现许多优惠券共用同一个ID的情况。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
//符号位0+时间戳31bit+序列号32bit
public long nextId(String keyPrefix){
    //生成时间戳
    LocalDateTime now=LocalDateTime.now();
    long nowSecond=now.toEpochSecond(ZoneOffset.UTC);
    long timestamp=nowSecond-BEGIN_TIMESTAMP;
    //生成序列号. 每一天下的单使用同一个Key
    String date=now.format(DateTimeFormatter.ofPattern("yyyy:MM:dd"));
    long count=  stringRedisTemplate.opsForValue().increment("icr:"+keyPrefix+":"+date);
    //拼接并返回
    return timestamp<<COUNT_BITS|count;
}

这里的 keyPrefix表示一种object,比如优惠券订单

超卖问题

假设200qps，200 个请求同时抢优惠券，然而优惠券的总数只有 100 张。 JMeter 测试发现，有 45% 的失败率；数据库查看，发现此时该库存数量为 -9，出现超卖情况。

乐观锁-版本号法

先从数据库查到一个版本号 A，向数据库更新的时候确认一下自己拿到的版本号A和此时数据库里的版本号B是否一致。如果一致, set version+1; 不一致，则放弃更新操作。

乐观锁-CAS

1
2
3
4
5
//扣减库存
boolean success= seckillVoucherService.update().setSql("stock=stock-1")
        .eq("voucher_id",voucherId)
        .eq("stock",voucher.getStock()) //乐观锁CAS 方法
        .update();

1
2
3
4
UPDATE seckill_voucher
SET stock = stock - 1
WHERE voucher_id = #{voucherId}
  AND stock = #{voucher.getStock()};

Jmeter测试发现，请求失败率高达 89%，查看数据库库存，发现还剩 79 张优惠券，根本没卖完。这是因为有很多请求，在它第一遍查库存（voucher.getStock()）和发出 update 请求之间，有别的线程已经修改了数据库里的库存。

本质上是数据库的行锁。这样做的好处就是不会出现超卖现象，但是请求失败率太高了。

改进

1
2
3
4
boolean success= seckillVoucherService.update().setSql("stock=stock-1")
        .eq("voucher_id",voucherId)
        .gt("stock", 0) //把加锁操作交给数据库
        .update();

1
2
3
4
UPDATE seckill_voucher
SET stock = stock - 1
WHERE voucher_id = #{voucherId}
  AND stock > 0;

本质上是数据库的行锁。JMeter 测试，请求成功率为 50%，查询数据库发现优惠券刚好买完。

一人一单

初步想法

创建订单时，先查数据库看这个用户是否下过单；再往数据库里添加 order。

问题：多个并发请求来自于同一个用户时，“查数据库”——“给数据库里加 order”存在线程安全问题。

synchronized悲观锁

锁下单function

锁住下单 function。为什么不能用Java 实现乐观锁？乐观锁只应用于 update，而这里是新创建订单。

 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
@Transactional
public synchronized Result createVoucherOrder(Long voucherId){
    Long userId=UserHolder.getUser().getId();
    int count=query().eq("user_id",userId).eq("voucher_id", voucherId).count();
    if(count>0){
        return Result.fail("用户已经购买过一次了～");
    }
    //扣减库存
    boolean success= seckillVoucherService.update().setSql("stock=stock-1")
            .eq("voucher_id",voucherId)
            .gt("stock", 0)
            .update();
    if(!success){
        return Result.fail("库存不足，数据库更新失败");
    }
    //创建订单
    VoucherOrder voucherOrder=new VoucherOrder();
    //返回订单id
    Long orderId= redisIdWorker.nextId("order");
    voucherOrder.setId(orderId);
    voucherOrder.setUserId(userId);
    voucherOrder.setVoucherId(voucherId);
    save(voucherOrder);
    return Result.ok(orderId);
}

假设有 200 个请求同时到达服务器，都想抢优惠券，其中 100 个请求来自用户 A ，100个请求来自用户 B。 synchronized 作用于该 instance method, 它只允许同一时间，只能有一个线程执行这个方法。一个线程为一个请求执行完这个方法后，其它线程才能开始。

事务与锁的顺序：事务开始、获取锁、数据库操作、提交事务、释放锁线程 A 提交完事务了，线程 B 才有获取锁的可能，因此只要事务隔离级别在读已提交及以上，那么这种操作是线程安全的。

缺点：锁的粒度太大！我只需要同一个用户的请求被串行化，不同用户的请求其实可以被多线程并发处理。

改进：减小锁粒度

把锁的粒度换成用户ID。对于那些并发线程，首先判断它们来自于哪个用户，然后对来自于同一个用户的请求处理做串行化。

userId.toString()一般会进行new String()操作，导致同一个用户的 userId可能被创建为许多不同的字符串。而intern()方法会优先从字符串常量池中拿字符串，确保同一个用户拥有唯一的userId对象。代码如下：

1
2
3
4
5
6
7
@Transactional
public Result createVoucherOrder(Long voucherId){
    Long userId=UserHolder.getUser().getId();
    synchronized(userId.toString().intern()){
        //code 省略
    }
}

缺点：如果把 synchronized 代码块放进这个@transactional 标注的方法之内，顺序就是事务开始——获取锁——释放锁——事务提交。

如果事务隔离级别设置的是读未提交以上的，那么在线程 A 释放锁之后，事务提交之前，线程B 获取锁，然后会读到旧数据。

synchronized 代码块放在事务外面

1
2
3
4
Long userId=UserHolder.getUser().getId();
synchronized(userId.toString().intern()){
    return createVoucherOrder(voucherId);
}

把这个函数放进代码块里面，某个线程获取锁——事务提交——释放锁，严格按照这样的顺序执行。这样保证其它线程在尝试获取锁时，读到的一定是数据库被上一个线程修改过的数据。

缺点：如果启动两个服务，负载均衡把来自同一个用户的两个请求分别发到两个服务，锁则没有用了，这个用户可以下两次单。两个 JVM 不可能被同一个synchronized 锁给锁住。

Redis 互斥锁

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
//这里加个锁。预防多 JVM 情况下，来自同一用户的同一时间的多个请求，分别占用了不同 JVM 的 synchronized 锁。
SimpleRedisLock lock=new SimpleRedisLock("order:"+userId, stringRedisTemplate);
boolean isLock=lock.tryLock(1200);
if(!isLock){
    return Result.fail("不允许重复下单");
}
try{
    IVoucherOrderService proxy=(IVoucherOrderService) AopContext.currentProxy();
    return  proxy.createVoucherOrder(voucherId);
}finally{
    lock.unlock();
}

在8081和8083开启两个服务，向8082发起两个请求，分别被nginx转发给这两个服务。

POST http://localhost:8082/api/voucher-order/seckill/11

可以看到8083的服务拿到了锁 redis里面显示了它拿到的互斥锁，其值为线程ID 30 8081的服务没拿到锁

缺点：如果线程1拿到锁之后，业务阻塞了，因为锁是设置了expire time，所以在业务完成之前，它就把锁释放了。这时候线程2趁虚而入，拿到了锁，开始执行任务。线程1业务跑完，释放锁，结果把线程2的锁给释放掉了。这时候线程3又拿到锁，开始执行任务。

改进：释放锁的时候，判断一下锁是否是自己的。但是线程1和线程2同时跑业务这件事也是不对的！

线程标示：解决锁误删

获取锁的时候存入线程标示（UUID）。原来存的是线程ID，是由JVM维护的。但如果在集群运行，会有多个JVM，那么线程ID会有重复
释放锁之前比较一下线程标示
可以解决锁误删问题，但是不能解决线程 1 业务阻塞导致锁过期，线程 2 和线程 1 一起执行业务的问题。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
@Override
public void unlock(){
    //获取线程标示
    String threadId=ID_PREFIX+Thread.currentThread().getId();
    //获取锁中的标示
    String id=stringRedisTemplate.opsForValue().get(KEY_PREFIX+threadId);
    if(threadId.equals(id)){
        //线程 1判断和主动释放之间，可能会有阻塞。如果这时候锁自动过期，会有别的线程2抢锁。那么线程 1阻塞完了之后，就直接把线程2的锁给释放了
        stringRedisTemplate.delete(KEY_PREFIX+name);
    }
}

避免锁误删需原子性操作

Redis里面运行Lua脚本，可以保证该脚本是原子性操作

1
EVAL "return redis.call('set', KEYS[1], ARGV[1])" 1 name Ruoke

Redis CLI里面这样写，EVAL后面跟着脚本内容，也可以允许传参

1
2
3
4
5
6
7
8
-- 这里的 KEYS[1] 就是锁的key，这里的ARGV[1] 就是当前线程标示
-- 获取锁中的标示，判断是否与当前线程标示一致
if (redis.call('GET', KEYS[1]) == ARGV[1]) then
    -- 一致，则删除锁
    return redis.call('DEL', KEYS[1])
end
-- 不一致，则直接返回
return 0

缺点

不可重入：同一个线程不可以多次获取同一把锁。如果一个线程想要在多个方法时都保持只有它自己进入，那么需要为每个方法都设置单独的互斥锁。
不可重试：当前的实现中，获取锁失败一次就立刻返回 false，非阻塞式。
业务堵塞造成锁过期，此时会有其它线程拿到锁。
主从一致性问题：主从节点的数据同步存在延迟。假如一个线程在主节点那拿到了锁（SETNX），然后主节点宕机，从节点还没有拿到同步的数据，就被推选为新的主节点。这时候其它线程可能从新的主节点那里拿到锁。

Redisson锁

用法

1
2
3
4
5
6
7
8
9
@Configuration
public class RedissonConfig {
    @Bean
    public RedissonClient redissonClient() {
        Config config=new Config();
        config.useSingleServer().setAddress("redis://localhost:6379");
        return Redisson.create(config);
    }
}

新建一个配置类，之后调用redissonClient的方法就行。

1
2
RLock lock=redissonClient.getLock("lock:order:"+userId);
boolean isLock=lock.tryLock();

获取锁的操作在释放锁之前打个断点，可以发现redis里面出现一个 key-value

可重入锁

Key是lock:order:1011； Value是 Hash，其中field 是线程 ID，value 是重入次数

unlock()操作并不是直接删除锁，而是将重入次数减一 redisson底层用lua脚本和SETNX实现原子性的加锁/释放锁逻辑. 加锁的时候先判断锁是不是自己的，如果是，给value加一。执行完业务之后再判断一次锁是否是自己的，如果是，给value减一。

获取锁失败，重试

1
2
boolean isLock=lock.tryLock(long waitTime, long leaseTime, TimeUnit unit);
如果有一次获取锁失败，它会去重试获取锁，最多等待 waitTime时间就不等了。leaseTime表示超时释放时间。

 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
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
    long time = unit.toMillis(waitTime);
    long current = System.currentTimeMillis();
    long threadId = Thread.currentThread().getId();
    //这一步是由 lua 脚本执行 SETNX 实现的。流程如 3.4.2 所示。 tryAcquire内部调用的是tryAcquireAsync.
    Long ttl = this.tryAcquire(waitTime, leaseTime, unit, threadId);
    if (ttl == null) {
        //如果获取锁成功了，则ttl未 null
        return true;
    } else {
        //如果没有获取锁，返回剩余的锁等待
        time -= System.currentTimeMillis() - current;
        if (time <= 0L) {
            //当前耗时已经超过了waitTime, 获取锁失败
            this.acquireFailed(waitTime, unit, threadId);
            return false;
        } else {
            current = System.currentTimeMillis();
            // 订阅锁释放事件，以便其他线程释放锁时可以立即尝试获取锁
            RFuture<RedissonLockEntry> subscribeFuture = this.subscribe(threadId);
            if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {
                //在指定的 waitTime 内没有成功订阅到锁的释放事件
                //返回获取锁失败的结果
                return false;
            } else {
                //订阅到了锁释放事件！！
                try {
                    //开始do-while轮询，不停地一边检查是否超过waitTime，一边尝试获取锁
                    }
                } finally {
                    this.unsubscribe(subscribeFuture, threadId);//取消订阅
                }
        }
    }
}

Redisson如何实现锁获取失败后的等待？利用sub/pub机制，当一个线程释放锁，会发送信号给其他线程。其它线程收到信号后，开启轮询来尝试获取锁。在这期间，时刻检查是否超过waitTime.

锁超时释放

tryLock内部会调用这个函数。如果 leaseTime设置了，表明明确设置了超时释放的时间。

如果没有设置，进入 else，用看门狗默认的30秒。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
    if (leaseTime != -1L) {
        return this.tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
    } else {
        RFuture<Long> ttlRemainingFuture = this.tryLockInnerAsync(waitTime, this.commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
        ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
            if (e == null) {
                //如果锁调用成功了。（ttlRemaining代表已有锁的剩余过期时间，如果返回 null, 则代表该线程获取了锁。
                if (ttlRemaining == null) {
                    //如果没有明确设置过期时间
                    this.scheduleExpirationRenewal(threadId);
                }
            }
        });
        return ttlRemainingFuture;
    }
}

接着进入这个函数。RedissonLock类里面定义了一个Concurrent HashMap, 也就是EXPIRATION_RENEWAL_MAP, 用来存许多ExpirationEntry. ExpirationEntry 代表一个看门狗任务，该任务会定期刷新锁的过期时间。

this.getEntryName()指的是Redis连接器ID+锁的Name

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
private void scheduleExpirationRenewal(long threadId) {
    ExpirationEntry entry = new ExpirationEntry();
    ExpirationEntry oldEntry = (ExpirationEntry)EXPIRATION_RENEWAL_MAP.putIfAbsent(this.getEntryName(), entry);
    if (oldEntry != null) {
        //如果这个锁的看门狗任务已经存在了
        oldEntry.addThreadId(threadId);
    } else {
        //如果这个锁的看门狗任务还没有建立
        entry.addThreadId(threadId);
        this.renewExpiration();
    }
}

renewExpiration 方法设置了一个定时任务，每隔一段时间（锁过期时间的三分之一），就会触发 renewExpirationAsync 方法续期锁的过期时间。

我们释放锁的时候，看门狗任务会从EXPIRATION_RENEWAL_MAP中被移出

使用看门狗的场景：

业务执行时间不可预估，需动态续期。看门狗业务完成后主动释放锁，资源立即回收。如果自己设定一个时间，过长会导致锁存活时间太长，过短会有并发风险。
如果客户端宕机，则锁会快速被释放

tryLock总结

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
tryLock(long waitTime, long leaseTime, TimeUnit unit){
    //1. 尝试获取锁
    tryAcquireAsync(){
        //2.1 如果设置了leaseTime
        tryLockInnerAsync(){
            // 调用lua脚本实现可设置重入次数的SETNX
        }
        //2.2 没设置leaseTime，用看门狗的默认 30 秒
        tryLockInnerAsync();
        //拿到锁了，准备开启看门狗任务
        scheduleExpirationRenewal(threadId){
            //属于这个锁的看门狗任务不存在
            每隔一段时间（锁过期时间的三分之一），就给锁续过期时间
        }
    }
    //如果没获取锁
    订阅。其它线程一旦释放锁，它得到信号，就开始循环调用tryAcquireAsync()
}

主从一致问题：MultiLock

启动三个Redis实例，然后为每个连接都创建一把锁，最后创建一个MultiLock. MultiLock允许你同时获取/释放这几把锁，确保所有锁都同时被获取/被释放。当有其它线程想要获取这个 MultiLock 的时候，它实际上会去检查，是否这个锁在每个 Redis 实例里面有，只要有一个 Redis 实例有这把锁，它就无法获取。

同一个进程想要获取重入锁时，也必须在所有节点都获取重入锁，才算获取成功。

Redisson实现的分布式锁总结

底层基于redis的setnx命令做了改进封装，使用lua脚本保证命令的原子性
利用hash结构，记录线程标示和重入次数；
利用watchDog延续锁时间；
控制锁重试等待
Redlock红锁解决主从数据一致的问题（不推荐）性能差
如果业务非要保证强一致性，建议采用zookeeper实现的分布式锁

秒杀优化

设计

对于一次请求，要先去查优惠券库存；为了防止一个用户下多单，又去加分布式锁，完成“查用户是否下单——优惠券扣库存——添加订单到数据库”这一系列操作，并发能力比较差。优化方案：把判断库存和判断一人一单的逻辑交给 Redis.

判断库存：key-value存储优惠券 Id 和库存；一人一单：key-set存储优惠券 Id 和 userId 集合。

用Lua脚本实现判断的逻辑，当Lua脚本返回0时，说明该用户此时有下单的资格，给它生成一个 orderId, 创建一个 voucherOrder.
将该voucherOrder加入阻塞队列
开启线程任务，不断地从阻塞队列中获取信息，实现异步下单功能。

判断库存

Lua 脚本在 Redis 里是原子执行的, 脚本一旦开始执行，就会完整地执行完，不会被其他命令打断。

当 Redis 正常运行，只是在执行 Lua 脚本内部的命令时出现错误，例如内存不足、达到资源限制等，pcall 会捕获到错误，进而执行回滚操作。这样可以保证在遇到这类异常时，数据能恢复到操作前的状态，维护数据的一致性。

 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
local voucherId = ARGV[1]
local userId = ARGV[2]
local orderId = ARGV[3]

local stockKey = 'seckill:stock:' .. voucherId
local orderKey = 'seckill:order:' .. voucherId

-- 备份库存和用户是否已下单的状态
local originalStock = tonumber(redis.call('get', stockKey))
local userHasOrdered = redis.call('sismember', orderKey, userId)

-- 检查库存
if originalStock <= 0 then
    return 1
end

-- 检查用户是否已下单
if userHasOrdered == 1 then
    return 2
end

-- 尝试扣减库存和记录订单
local success, err = pcall(function()
    redis.call('incrby', stockKey, -1)
    redis.call('sadd', orderKey, userId)
    redis.call('xadd', 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', orderId)
end)

if not success then
    -- 回滚操作
    if originalStock then
        redis.call('set', stockKey, originalStock)
    end
    if userHasOrdered == 0 then
        redis.call('srem', orderKey, userId)
    end
    return -1 -- 返回 -1 表示执行异常
end

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
@PostConstruct
private void init(){
    SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
}
private class VoucherOrderHandler implements Runnable{
    @Override
    public void run() {
        while(true){
            try {
                //获取队列中的订单信息
                VoucherOrder voucherOrder=orderTasks.take();
                //创建订单
                handleVoucherOrder(voucherOrder);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }
}

private void handleVoucherOrder(VoucherOrder voucherOrder) {
    Long userId=voucherOrder.getUserId();//因为这是后台线程操作的，所以不能从ThreadLocal里面拿ID，必须从 voucherOrder里面拿 ID
    RLock lock=redissonClient.getLock("lock:order:"+userId);
    boolean isLock=lock.tryLock();
    if(!isLock){
        return;
    }
    try{
        proxy.createVoucherOrder(voucherOrder);
    }finally{
        lock.unlock();
    }
}

@PostConstruct：当这个类的 Spring Bean被创建并且完成依赖注入之后，init()方法立即被调用，也就是后台线程立即开始执行 run()任务。
orderTasks.take(): run()任务不停地尝试从阻塞队列中获取 order. 这个方法如果拿不到任务，该线程就被阻塞；如果拿到了，就开始处理数据库层面的创建账单。

消息队列实现

List结构模拟消息队列

Redis的 List 其实是一个双向链表，具有LPUSH, BRPOP功能，符合队列的要求。BRPOP代表具有阻塞作用，当没有消息可以读取的时候，阻塞并等待消息。

比起 JVM 阻塞队列的好处：

它在 JVM 之外，不用担心内存达到上限；
如果突然宕机，已经完成持久化的数据依然存在。

缺点：

无法避免消息丢失：POP 之后就从队列中走了，如果业务逻辑失败，这条消息就消失了
只支持单消费者。

Pub-Sub实现消息队列

可以允许多个生产者，多个消费者. 每个消费者可以订阅不同的频道

1
2
3
publish order.q1 hello
subscribe order.q1
psubscribe order.*

缺点：不支持数据持久化。所有发出的消息不会在redis里面保存，如果一个频道没有人订阅，发出的消息会丢失。消息堆积有上限。

Streams

1
2
3
4
127.0.0.1:6379> xadd s1 * name tom age 21
"1738826384747-0"
127.0.0.1:6379> xadd s1 * name betty age 23
"1738826467103-0"

写数据用 XADD, 往stream s1里面加入两个entry。 * 代表entry的ID由redis生成

1
2
3
4
5
6
7
127.0.0.1:6379> xread count 1 streams s1 0
1) 1) "s1"
   2) 1) 1) "1738826384747-0"
         2) 1) "name"
            2) "tom"
            3) "age"
            4) "21"

写数据用XREAD, count 指定读取的 entry数量，0表示从第一个id开始往后读,$表示读最新消息

1
2
3
4
5
6
7
8
127.0.0.1:6379> xread count 1 block 0 streams s1 $
1) 1) "s1"
   2) 1) 1) "1738826924108-0"
         2) 1) "name"
            2) "huang"
            3) "age"
            4) "23"
(33.29s)

阻塞式读数据，block后面跟一个数字表明最长可以等多久。

特点：

消息可以回溯
有漏读消息的风险。比如在循环中使用$读最新消息，可能来了好几条消息，但是处理第一条就花了很久，导致之后第二次循环读消息读到的不是第二个到的消息
一个消息可以多个消费者获取
可以阻塞读取

1
2
127.0.0.1:6379> xgroup create s1 g1 0
OK

创建一个消费者组g1, 监听s1消息队列

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
127.0.0.1:6379> xreadgroup group g1 c1 count 1 block 2000 streams s1 >
1) 1) "s1"
   2) 1) 1) "1738826384747-0"
         2) 1) "name"
            2) "tom"
            3) "age"
            4) "21"
127.0.0.1:6379> xreadgroup group g1 c1 count 1 block 2000 streams s1 >
1) 1) "s1"
   2) 1) 1) "1738826467103-0"
         2) 1) "name"
            2) "betty"
            3) "age"
            4) "23"

指定消费者组里面的c1这个consumer去s1里面读数据，’>‘表示从下一个未消费的ID开始读

1
2
127.0.0.1:6379> xack s1  g1 1738826467103-0
(integer) 1

当一个消息成功被处理，手动给它ACK，然后它就会从消息队列中被移除。

Streams实现

首先创建一个消费者组：xgroup create stream.orders g1 0 MKSTREAM
Lua脚本在判断此次请求拥有购买资格后，直接向消息队列stream.orders中添加信息，内容包含voucherId, userId, orderId
项目启动时，开启一个后台进程，不断地从stream.orders获取信息，处理订单

总结

方案对比

方案	优点	缺点	适用场景
方法级synchronized	实现简单	性能差（所有用户串行）	绝对不推荐
用户ID锁（JVM内）	细粒度、性能较好	集群环境下失效	单机部署
Redis分布式锁	支持集群、细粒度控制	实现略复杂	生产环境推荐

Redis分布式锁的问题及解决方案

问题类型	解决方案
锁误删	Lua脚本原子解锁
锁超时导致并发	Redisson看门狗