使用Jedis和Sentinel完成秒杀功能

正文

需要写一个秒杀功能，需要解决的是高并发、商品超卖、数据正确性、限流等问题

解决并发与分布式锁保证数据正确性

import redis.clients.jedis.Jedis;
import java.util.concurrent.TimeUnit;

public class SecKillWithDistributedLock {

    private final static String LOCK_KEY = "sec:kill:lock";
    private final static int TIMEOUT = 3 * 1000; // 过期时间
    private final static int SLEEP_TIME = 500; // 重试休眠时间

    public static void main(String[] args) {
        Jedis jedis = new Jedis("localhost", 6379);
		String productId="123";
        // 模拟商品数量
        jedis.set("product_num", "100");

        // 设置分布式锁
        boolean lock = false;
        try {
            // 获取锁
            while (!lock) {
                lock = tryGetDistributedLock(jedis, LOCK_KEY+productId, TIMEOUT);
                if (!lock) {
                    System.out.println("获取锁失败，等待" + SLEEP_TIME / 1000 + "秒重新获取！");
                    TimeUnit.MILLISECONDS.sleep(SLEEP_TIME);
                }
            }

            // 进行秒杀操作
            String productNumStr = jedis.get("product_num");
            if (productNumStr != null && Integer.parseInt(productNumStr) > 0) {
                jedis.decr("product_num");
                System.out.println("秒杀成功！");
            } else {
                System.out.println("秒杀失败！");
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 释放锁
            releaseDistributedLock(jedis, LOCK_KEY);
        }
    }

    /**
     * 尝试获取分布式锁
     * @param jedis Redis客户端
     * @param lockKey 锁的key值
     * @param timeout 超时时间（毫秒）
     * @return 获取锁成功返回true，否则返回false
     */
    public static boolean tryGetDistributedLock(Jedis jedis, String lockKey, int timeout) {
        long currentTime = System.currentTimeMillis();
        while (true) {
            // 设置锁的过期时间
            String result = jedis.set(lockKey, String.valueOf(currentTime + timeout), "NX", "PX", timeout);
            if ("OK".equals(result)) {
                return true;
            }
            // 判断锁是否已经过期
            String lockValue = jedis.get(lockKey);
            if (lockValue != null && Long.parseLong(lockValue) < System.currentTimeMillis()) {
                // 解锁(防止当前线程解了别人的锁)
                jedis.del(lockKey);
            }

            // 给一点休眠时间，避免出现死循环
            try {
                TimeUnit.MILLISECONDS.sleep(SLEEP_TIME);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            // 超时退出
            if (System.currentTimeMillis() - currentTime > timeout - SLEEP_TIME) {
                return false;
            }
        }
    }

    /**
     * 释放分布式锁
     * @param jedis Redis客户端
     * @param lockKey 锁的key值
     */
    public static void releaseDistributedLock(Jedis jedis, String lockKey) {
        String lockValue = jedis.get(lockKey);
        if (lockValue != null && Long.parseLong(lockValue) > System.currentTimeMillis()) {
            jedis.del(lockKey);
        }
    }
}
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

这份代码结合了秒杀代码和Jedis的分布式锁代码，首先尝试获取分布式锁，成功之后进行秒杀操作，完成操作之后释放锁。在获取锁失败时会等待一段时间重新获取锁，以避免重复提交请求或出现死循环等问题。

使用Sentinel对接口进行限流

1、引入Sentinel依赖

2、初始化Sentinel

在SpringBoot项目中，你需要创建一个配置类来初始化Sentinel，代码如下：

@Configuration
public class SentinelConfig {

    /**
     * 初始化Sentinel
     */
    @PostConstruct
    public void init() {
        // 注册Sentinel的ServletDispatcher
        WebCallbackManager.setRequestOriginParser(new RequestOriginParser() {
            @Override
            public String parseOrigin(HttpServletRequest request) {
                // 获取请求来源
                return request.getHeader("origin");
            }
        });
        // 启动Sentinel
        InitFunc.init();
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

其中，Sentinel提供了一个WebCallbackManager，在初始化时可以设置RequestOriginParser来获取请求来源，传入的请求来源可以是IP地址、域名或其他信息。此外，还需要调用InitFunc.init()方法来启动Sentinel。

3、编写限流规则

在Sentinel中，你需要定义限流规则。可以使用注解@SentinelResource将方法进行限流，也可以通过编码的方式来实现。以下是通过编码方式来定义限流规则的示例：

public class FlowRuleInit {
    public static void initFlowRules() {
        List rules = new ArrayList<>();
        // 定义限流规则
        FlowRule rule = new FlowRule();
        rule.setResource("your-resource-name");
        rule.setCount(10);
        rule.setGrade(RuleConstant.FLOW_GRADE_QPS);
        rule.setLimitApp("default");
        rules.add(rule);
        // 注册限流规则
        FlowRuleManager.register2Property(rules);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

您也可以选择在Spring Boot应用启动时，自动调用该类的初始化方法进行限流规则的注册，具体实现可以将该方法加上@PostConstruct或者使用CommandLineRunner等方式。例如下面的示例代码可以让FlowRuleInit类的initFlowRules()方法在应用启动时自动执行：

@Component
public class SentinelInit implements CommandLineRunner {

    @Override
    public void run(String... args) throws Exception {
        // 调用 FlowRuleInit.initFlowRules() 方法进行限流规则初始化
        FlowRuleInit.initFlowRules();
    }
}
1
2
3
4
5
6
7
8
9

在上述代码中，定义了一个资源名为“testResourceName”的限流规则，设置限流等级为QPS，即每秒最多处理的请求数量为10。

4、使用Sentinel限流

在你的业务代码中，使用@SentinelResource注解来标记需要进行限流的方法，示例如下：

@Service
public class ProductService {

    /**
     * 查询商品详情
     */
    @SentinelResource(value = "productDetail", blockHandler = "handleBlock")
    public ProductVO getProductDetail(String productId) {
        // 查询商品详情逻辑
        // ...
    }

    /**
     * 限流处理
     */
    public ProductVO handleBlock(String productId, BlockException e) {
        // 限流处理逻辑
        // ...
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

在上述代码中，@SentinelResource注解的value属性值表示资源名称，指定了需要对该方法进行限流。另外，还可以通过blockHandler属性来指定限流处理方法。如果被限流，就会调用该方法进行处理。

5、加入乐观锁保证数据正确性

productId：商品的主键
count：每次减掉的库存

public boolean lockStock(String productId, int count) {
    // 查询当前库存
    Stock stock = stockDao.getStockById(productId);
    if (stock == null) {
        return false; // 库存不存在
    }
    
    // 获取当前库存version值
    int oldVersion = stock.getVersion();
    
    // 计算新的库存数量和version值
    int newCount = stock.getCount() - count;
    int newVersion = oldVersion + 1;
    
    // 更新库存，并设置新的version值
    int updateCount = stockDao.updateStock(productId, newCount, oldVersion, newVersion);
    if (updateCount <= 0) {
        return false; // 更新失败，库存version值已经被其他线程修改
    }
    
    return true;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22