前言

最近学习了利用redis解决许多业务问题的方法，优惠券秒杀是其中一个综合性较强的业务问题。

全局唯一id

雪花算法：雪花算法的原理就是生成一个的 64 位比特位的 long 类型的唯一 id。id组成为符号位（1bit）、时间戳（41bits）、工作机器ID（10bits）、序列号（12bits）。雪花算法能基于时间戳生成自增的id，但正是因为基于时间戳，雪花算法有可能出现时间（时钟）回拨的问题。

组成：

1：最高 1 位是符号位，固定值 0，表示id 是正整数
41：接下来 41 位存储毫秒级时间戳，2^41/(1000606024365)=69，大概可以使用 69 年。
10：再接下 10 位存储机器码，包括 5 位 datacenterId 和 5 位 workerId。最多可以部署 2^10=1024 台机器。
12：最后 12 位存储序列号。同一毫秒时间戳时，通过这个递增的序列号来区分。即对于同一台机器而言，同一毫秒时间戳下，可以生成 2^12=4096 个不重复 id。

基于雪花算法生成id：

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
/**
 * 雪花算法分布式ID生成器
 *
 * 结构：
 * 0 - 41位时间戳 - 10位工作机器ID - 12位序列号
 * 总计64位
 */
public class SnowflakeIdGenerator {
    // 起始时间戳（2023-01-01 00:00:00 UTC）
    private static final long EPOCH = 1672531200000L;

    // 机器ID位数
    private static final long WORKER_ID_BITS = 10L;
    // 序列号位数
    private static final long SEQUENCE_BITS = 12L;

    // 最大机器ID（1023）
    private static final long MAX_WORKER_ID = ~(-1L << WORKER_ID_BITS);
    // 最大序列号（4095）
    private static final long MAX_SEQUENCE = ~(-1L << SEQUENCE_BITS);

    // 时间戳左移位数（22位：10位机器ID + 12位序列号）
    private static final long TIMESTAMP_SHIFT = WORKER_ID_BITS + SEQUENCE_BITS;
    // 机器ID左移位数（12位序列号）
    private static final long WORKER_ID_SHIFT = SEQUENCE_BITS;

    private final long workerId;      // 工作机器ID
    private long sequence = 0L;       // 序列号
    private long lastTimestamp = -1L; // 上次生成ID的时间戳

    // 时钟回拨容忍阈值（5毫秒）
    private static final long MAX_CLOCK_BACKWARD_MS = 5;

    /**
     * 构造函数
     * @param workerId 工作机器ID (0-1023)
     */
    public SnowflakeIdGenerator(long workerId) {
        if (workerId < 0 || workerId > MAX_WORKER_ID) {
            throw new IllegalArgumentException(
                    String.format("Worker ID must be between 0 and %d", MAX_WORKER_ID));
        }
        this.workerId = workerId;
    }

    /**
     * 生成唯一ID（线程安全）
     * @return 64位唯一ID
     */
    public synchronized long nextId() {
        long timestamp = currentTime();

        // 处理时钟回拨
        if (timestamp < lastTimestamp) {
            long offset = lastTimestamp - timestamp;
            if (offset <= MAX_CLOCK_BACKWARD_MS) {
                // 容忍小范围回拨，等待时间追赶
                try {
                    wait(offset);
                    timestamp = currentTime();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    throw new RuntimeException("Clock moved backwards, wait interrupted", e);
                }
            } else {
                // 严重时钟回拨，抛出异常
                throw new RuntimeException(
                        String.format("Clock moved backwards. Refusing to generate ID for %d milliseconds", offset));
            }
        }

        // 同一毫秒内生成
        if (lastTimestamp == timestamp) {
            sequence = (sequence + 1) & MAX_SEQUENCE;
            // 序列号用完，等待下一毫秒
            if (sequence == 0) {
                timestamp = tilNextMillis(lastTimestamp);
            }
        }
        // 新毫秒重置序列号
        else {
            sequence = 0L;
        }

        lastTimestamp = timestamp;

        // 组合ID
        return ((timestamp - EPOCH) << TIMESTAMP_SHIFT)
                | (workerId << WORKER_ID_SHIFT)
                | sequence;
    }

    /**
     * 获取当前时间（毫秒）
     */
    private long currentTime() {
        return System.currentTimeMillis();
    }

    /**
     * 等待直到下一毫秒
     */
    private long tilNextMillis(long lastTimestamp) {
        long timestamp = currentTime();
        while (timestamp <= lastTimestamp) {
            timestamp = currentTime();
        }
        return timestamp;
    }

    /**
     * 从ID解析时间戳
     */
    public static long parseTimestamp(long id) {
        return (id >> TIMESTAMP_SHIFT) + EPOCH;
    }

    /**
     * 从ID解析工作机器ID
     */
    public static long parseWorkerId(long id) {
        return (id >> WORKER_ID_SHIFT) & MAX_WORKER_ID;
    }

    /**
     * 从ID解析序列号
     */
    public static long parseSequence(long id) {
        return id & MAX_SEQUENCE;
    }

在优惠券业务中，我们需要向客户端返回订单号，这个订单号在sql中作为订单表的主键，处于安全性等考虑我们不希望订单号在sql表中简单的自增。在redis中以字符串为key的键值对，key中的字符串具有不同的哈希值，且其值是自增的，我们可以利用这一特性制造全局唯一id。

基于redis生成id组成如下：

image-20250721175456776

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
@Component
public class RedisIdWorker {
    /**
     * 开始时间戳
     */
    private static final long BEGIN_TIMESTAMP = 1640995200L;
    /**
     * 序列号的位数
     */
    private static final int COUNT_BITS = 32;

    private StringRedisTemplate stringRedisTemplate;

    public RedisIdWorker(StringRedisTemplate stringRedisTemplate) {
        this.stringRedisTemplate = stringRedisTemplate;
    }

    public long nextId(String keyPrefix) {
        // 1.生成时间戳
        LocalDateTime now = LocalDateTime.now();
        long nowSecond = now.toEpochSecond(ZoneOffset.UTC);
        long timestamp = nowSecond - BEGIN_TIMESTAMP;

        // 2.生成序列号

        // 2.1.获取当前日期，精确到天
        String date = now.format(DateTimeFormatter.ofPattern("yyyyMMdd"));

        // 2.2.自增长
        long count = stringRedisTemplate.opsForValue().increment("icr:" + keyPrefix + ":" + date);

        // 3.拼接并返回
        return timestamp << COUNT_BITS | count;
    }
}

秒杀下单

在秒杀下单这一场景下我们后端需要处理许多高并发的业务问题，如超卖问题、一人一单问题等。

主要逻辑流程图：

image-20250721190304587

基础代码：

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
@Override
@Transactional
public Result seckillVoucher(Long voucherId) {
    // 1.查询优惠券
    SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
    // 2.判断秒杀是否开始
    if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
        // 尚未开始
        return Result.fail("秒杀尚未开始");
    }
    // 3.判断秒杀是否结束
    if (voucher.getBeginTime().isBefore(LocalDateTime.now())) {
        // 尚未开始
        return Result.fail("秒杀已经结束");
    }
    // 4.判断库存是否充足
    if (voucher.getStock() < 1) {
        // 库存不足
        return Result.fail("库存不足");
    }
    // 5.扣减库存
    boolean success = seckillVoucherService.update()
            .setSql("stock = stock - 1")
            .eq("voucher_id", voucherId).update();
    if(!success){
        // 扣减失败
        return Result.fail("库存不足");
    }
    // 6.创建订单
    VoucherOrder voucherOrder = new VoucherOrder();
    // 6.1订单id
    long orderId = redisIdWorker.nextId("order");
    voucherOrder.setId(orderId);
    // 6.2用户id
    Long userId = UserHolder.getUser().getId();
    voucherOrder.setUserId(userId);
    // 6.3代金券id
    voucherOrder.setVoucherId(voucherId);
    save(voucherOrder);
    // 7.返回订单
    return Result.ok(orderId);
}

超卖问题

在发放优惠券的时候，如果出现多个用户同时进入抢票程序，且由于当时数据库仍有余票未更新，这会导致卖出的票与预想的要多，这在生产环境是十分危险的，为解决并发问题，我们可以利用悲观锁与乐观锁两种思路。

  • 悲观锁：悲观锁认为随时都会有其他线程抢占资源，当前线程获取锁后，其他线程将无法对正常运行。
  • 乐观锁：乐观锁认为不会有其他线程抢占资源，当前线程与其他线程不会互斥，他们将正常访问资源，但是在更新资源前后都会进行一次比较，若前后的结果不一致，则认为有其他线程更改了资源，当前线程会认为该数据不可用。

基于乐观锁的实现：

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
@Override
@Transactional
public Result seckillVoucher(Long voucherId) {
    // 1.查询优惠券
    SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
    // 2.判断秒杀是否开始
    if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
        // 尚未开始
        return Result.fail("秒杀尚未开始");
    }
    // 3.判断秒杀是否结束
    if (voucher.getBeginTime().isBefore(LocalDateTime.now())) {
        // 尚未开始
        return Result.fail("秒杀已经结束");
    }
    // 4.判断库存是否充足
    if (voucher.getStock() < 1) {
        // 库存不足
        return Result.fail("库存不足");
    }
    // 5.扣减库存
    boolean success = seckillVoucherService.update()
            .setSql("stock = stock - 1")
            .eq("voucher_id", voucherId)
    //乐观锁解决超卖问题，没错只需要增加这个语句就可以实现在更新数据前对数据进行比较，只不过这里比较的是库存是否仍有剩余，通常我们会在更新前获取一次该数据，然后在更新时再次对该数据进行获取，与之前的数据进行比较。不过这种方法将导致大量请求被拒绝，因为在该线程从获取到更新这段时间，有很大可能其他线程已经对该数据进行了修改，性能较差，所以可以优化为对当前剩余的库存进行判断。
            .gt("stock", 0).update();
    if(!success){
        // 扣减失败
        return Result.fail("库存不足");
    }
    // 6.创建订单
    VoucherOrder voucherOrder = new VoucherOrder();
    // 6.1订单id
    long orderId = redisIdWorker.nextId("order");
    voucherOrder.setId(orderId);
    // 6.2用户id
    Long userId = UserHolder.getUser().getId();
    voucherOrder.setUserId(userId);
    // 6.3代金券id
    voucherOrder.setVoucherId(voucherId);
    save(voucherOrder);
    // 7.返回订单
    return Result.ok(orderId);
}

一人一单问题

我们希望优惠券的策略是一人一票，不允许购买多次。实现的方式很简单，我们只需要在对数据库更新前根据用户id访问数据库，如果没有查到记录则认为该用户是第一次购买优惠券，反之则拒绝。

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
@Override
@Transactional
public Result seckillVoucher(Long voucherId) {
    // 1.查询优惠券
    SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
    // 2.判断秒杀是否开始
    if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
        // 尚未开始
        return Result.fail("秒杀尚未开始");
    }
    // 3.判断秒杀是否结束
    if (voucher.getBeginTime().isBefore(LocalDateTime.now())) {
        // 尚未开始
        return Result.fail("秒杀已经结束");
    }
    // 4.判断库存是否充足
    if (voucher.getStock() < 1) {
        // 库存不足
        return Result.fail("库存不足");
    }
    // 5.一人一单
    Long userId = UserHolder.getUser().getId();
    int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
    if(count > 0){
        return Result.fail("一个账号只能购买一次优惠券");
    }
    // 6.扣减库存
    boolean success = seckillVoucherService.update()
            .setSql("stock = stock - 1")
            .eq("voucher_id", voucherId)
            .gt("stock", 0).update();
    if(!success){
        // 扣减失败
        return Result.fail("库存不足");
    }
    // 7.创建订单
    VoucherOrder voucherOrder = new VoucherOrder();
    // 7.1订单id
    long orderId = redisIdWorker.nextId("order");
    voucherOrder.setId(orderId);
    // 7.2用户id
    voucherOrder.setUserId(userId);
    // 7.3代金券id
    voucherOrder.setVoucherId(voucherId);
    save(voucherOrder);
    // 8.返回订单
    return Result.ok(orderId);
}

其实不难想到上述代码存在的并发问题，如果有人利用数据库录入该账户信息这段时间，多次请求优惠券，这些请求也将被视为合法，因为数据库不能查询到该用户的信息。我们同样利用锁机制解决这个问题，大致思想就是在进行写数据时加一把锁，其他线程的请求将被拒绝，这样就可以放心的更新数据库。

基于悲观锁(synchronized)实现：

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
@Override
public Result seckillVoucher(Long voucherId) {
    // 1.查询优惠券
    SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
    // 2.判断秒杀是否开始
    if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
        // 尚未开始
        return Result.fail("秒杀尚未开始");
    }
    // 3.判断秒杀是否结束
    if (voucher.getBeginTime().isBefore(LocalDateTime.now())) {
        // 尚未开始
        return Result.fail("秒杀已经结束");
    }
    // 4.判断库存是否充足
    if (voucher.getStock() < 1) {
        // 库存不足
        return Result.fail("库存不足");
    }
    // 这里根据用户id加锁
    Long userId = UserHolder.getUser().getId();
    // 锁整个函数是为了保证事务提交后，锁才被释放
    synchronized (userId.toString().intern()){
        //获取对象的代理对象
        IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
        return createSeckillVoucher(voucherId);
    }
}

@Transactional
public Result createSeckillVoucher(Long voucherId){
    // 5.扣减库存
    boolean success = seckillVoucherService.update()
            .setSql("stock = stock - 1")
            .eq("voucher_id", voucherId)
            .gt("stock", 0).update();
    if(!success){
        // 扣减失败
        return Result.fail("库存不足");
    }
    // 6.创建订单
    VoucherOrder voucherOrder = new VoucherOrder();
    // 6.1订单id
    long orderId = redisIdWorker.nextId("order");
    voucherOrder.setId(orderId);
    // 6.2用户id
    Long userId = UserHolder.getUser().getId();
    voucherOrder.setUserId(userId);
    // 6.3代金券id
    voucherOrder.setVoucherId(voucherId);
    save(voucherOrder);
    // 7.返回订单
    return Result.ok(orderId);
}

集群服务器

其实刚刚我们所有场景都是在单机模式下进行的，现阶段的企业开发一般会部署多台服务器来处理请求，但是我们的锁只能在本机中可见，其他服务器是看不到的，也就是说，当我们操作数据库时，其他请求可能通过其他服务器去访问数据库，这样一来又会出现并发问题。我们需要借助分布式锁解决这个问题，分布式锁有很多实现方式，我们采用redis实现。其实这里又涉及许多并发问题，详细可以去b站上看是怎么一步步完善分布式锁的。

实战篇-09.分布式锁-基本原理和不同实现方式对比_哔哩哔哩_bilibili

Lua 基本语法 | 菜鸟教程

基于redis的分布式锁实现：

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
@Override
public Result seckillVoucher(Long voucherId) {
    // 1.查询优惠券
    SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
    // 2.判断秒杀是否开始
    if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
        // 尚未开始
        return Result.fail("秒杀尚未开始");
    }
    // 3.判断秒杀是否结束
    if (voucher.getBeginTime().isBefore(LocalDateTime.now())) {
        // 尚未开始
        return Result.fail("秒杀已经结束");
    }
    // 4.判断库存是否充足
    if (voucher.getStock() < 1) {
        // 库存不足
        return Result.fail("库存不足");
    }
    Long userId = UserHolder.getUser().getId();
    // 创建锁对象
    SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);
    // 获取锁
    boolean isLock = lock.tryLock(1200);
    if(!isLock){
        // 获取锁失败，返回错误或重试
        return Result.fail("不允许重复下单");
    }
    try {
        // 执行业务
        IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
        return createSeckillVoucher(voucherId);
    }finally {
        // 释放锁
        lock.unlock();
    }
}
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
public class SimpleRedisLock implements ILock {

    private String name;
    private StringRedisTemplate stringRedisTemplate;

    public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {
        this.name = name;
        this.stringRedisTemplate = stringRedisTemplate;
    }

    private static final String KEY_PREFIX = "lock:";
    private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";
    private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;
    static {
        UNLOCK_SCRIPT = new DefaultRedisScript<>();
        UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));
        UNLOCK_SCRIPT.setResultType(Long.class);
    }

    @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() {
        // 调用lua脚本
        stringRedisTemplate.execute(
                UNLOCK_SCRIPT,
                Collections.singletonList(KEY_PREFIX + name),
                ID_PREFIX + Thread.currentThread().getId());
    }

到这里我们的锁就是一个勉强可以用的分布式锁了，当然他仍然存在许多问题，感兴趣的可以继续学习基于redisson实现分布式锁。

实战篇-18.分布式锁-Redisson快速入门_哔哩哔哩_bilibili