黑马点评项目总结(中)

三、优惠券秒杀

3.1 全局ID生成器

3.1.1 原因

受限于单表的存储能力，分表会导致重复ID，同时要保证ID的安全性，无法使用数据库的自增字段

3.1.2 生成方式

当系统是分布式系统时，也需要满足全局ID唯一，因此采用Redis生成全局唯一ID。为保证安全性，需要将redis自增的数值拼接一些其他内容，具体生成策略如下：

第一位为符号位
接下来的31位用时间戳填满，以秒为单位
最后32位是自增序列号，每秒内可支持2^32^个ID

3.1.3 其他全局唯一ID生成策略

UUID
snowflake算法 (算法在附录中)
数据库自增

3.2 实现秒杀下单

3.2.1 业务流程

3.2.2 存在的问题

在当前的实现中，会产生多卖问题，这属于多线程安全问题。为解决此问题，需要给当前业务加锁。

乐观锁 vs 悲观锁

乐观锁：

认为线程安全问题不一定会发生，因此不会加锁，只会在更新数据时检查数据有没有被其他线程修改。

悲观锁：

认为线程安全问题一定会发生，因此在代码执行前先获取锁，保证线程串行执行。

3.2.3 解决方案及代码实现

本项目采用乐观锁解决超卖问题，乐观锁一般有两种实现方式：

版本号法：每次更新数据时查询数据库中的版本号和先前查询的版本号是否一致，不一致证明数据已被其他线程修改过；若一致，则更新数据和版本号

CAS(compare and switch)法：利用数据库中的一个字段代替版本号
以下代码是乐观锁的CAS代码实现：

boolean success = seckillVoucherService.update()
                .setSql("stock = stock - 1")
                .eq("voucher_id", voucherId)
                .gt("stock", 0)
                .update();

3.3 实现一人一单

// 5. 一户一单，判断用户是否已经购买过
        Long userId = UserHolder.getUser().getId();
        int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
        if (count > 0) {
            return Result.fail("用户已经购买过一次！");
        }

3.3.1 出现的问题

由于对于秒杀券订单的增加在上述代码后面实现，因此可能会出现一个ID第一个抢购秒杀券的订单还未写入到数据库中，此时同ID的另一个线程执行上述代码，导致查询到的结果与实际不符，最终一个用户ID购买了多个秒杀券。

3.3.2 解决方案

由于乐观锁只能针对更新操作，本操作是添加操作，因此使用悲观锁解决。锁的对象是每个用户ID，范围是从判断用户是否下过单直到添加订单到数据库。

3.3.3 实现过程中出现的问题

项目中将加锁的部分抽出形成一个方法B，在原方法A中调用了方法B。由于方法B中对数据库的两个表进行的修改，因此添加了事务，由于锁需要在事务提交后进行释放，因此锁是加到方法A中。

// 方法A
public Result seckillVoucher(Long voucherId) {
    Long userId = UserHolder.getUser().intern();
    synchronized(userId.toString().intern()) {
      return getResult(voucherId);
    }
}

// 方法B
@Transactional
public Result getResult(Long voucherId) {
    // 包含了判断用户是否下过单以及对两个数据库的修改
}

在上述代码实现中，会导致方法B的事务失效。原因如下：在spring中，@Transactional注解是基于动态代理实现的，事务的实现由代理对象管理。在上面的代码中，直接使用方法名称调用方法B，实际是this.getResult()，没有通过代理对象调用，因此方法B的事务不会生效。

3.3.4 对该问题的解决

若想解决事务失效的问题，首先需要获取到本类的代理对象，随后通过代理对象调用方法B。具体代码实现如下：

// 方法A
    public Result seckillVoucher(Long voucherId) {
        Long userId = UserHolder.getUser().intern();
        synchronized(userId.toString().intern()) {
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
            return proxy.getResult(voucherId);
        }
    }

    // 方法B
    @Transactional
    public Result getResult(Long voucherId) {
        // 包含了判断用户是否下过单以及对两个数据库的修改
    }

四、分布式锁

4.1 服务器集群模式下的线程安全问题

4.1.1 问题描述

由于每一个jvm都有自己的锁监视器，因此当后端服务器是集群模式的情况时，同一时刻同一用户不同线程访问不同的后端服务器，都会获得锁，故针对“一人一单”的锁就会失效。

4.1.2 使用redis实现分布式锁

使用redis的setnx命令即可实现分布式锁，下面是一个使用redis实现分布式锁的样例。

public class SimpleRedisLock implements ILock{

    private StringRedisTemplate stringRedisTemplate;
    private String key;

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

    private static final String KEY_PREFIX = "lock:";
    private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "_";

    @Override
    public boolean tryLock(long timeoutSec) {
        String value = ID_PREFIX +  Thread.currentThread().getId();
        Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + key, value, timeoutSec, TimeUnit.MINUTES);
        return Boolean.TRUE.equals(flag);
    }

    @Override
    public void unlock() {
        String myLock = ID_PREFIX + Thread.currentThread().getId();

        String redisLock = stringRedisTemplate.opsForValue().get(KEY_PREFIX + key);

        if (myLock.equals(redisLock)) {
            stringRedisTemplate.delete(KEY_PREFIX + key);
        }
    }
}

在该分布式锁的实现过程中，需要注意一下几个问题：

设置锁的超时施放时间，可以防止因线程拿到锁后迟迟不释放锁导致出现死锁的问题。
使用随机前缀 + 线程ID作为锁的value，原因有两个：
- 当线程需要释放锁时，首先获取锁的value，与自己的线程ID作比较，可以判断自己的锁是否已经释放，防止误释放其他线程锁的情况
- 由于后端是集群模式，因此线程ID可能会有重复，因此需要随机前缀保证不同后端服务器的不同线程保存锁的value值不重复

4.1.3 实现过程中的问题及解决

由于在释放锁时，判断锁的value和自己现成的id是否相等与释放锁的操作不是原子性的，因此可能会出现线程安全问题
- 使用lua脚本可以实现对redis操作的原子性，代码实现如下：

if (redis.call('get', KEYS[1]) == ARGV[1]) then
    return redis.call('del', KEYS[1])
end
return 0
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 void unlock() {
        stringRedisTemplate.execute(UNLOCK_SCRIPT, Collections.singletonList(KEY_PREFIX + key),
                ID_PREFIX + Thread.currentThread().getId());
    }

4.2 redisson工具

4.2.1 书写的redis分布式锁的缺点

不可重入：同一个线程无法多次获取同一把锁
不可重试：获取锁失败就会返回false，没有重试机会
超时释放：线程任务未完成也会释放锁，存在安全隐患
主从一致性：redis集群可能会导致锁被多个线程获取，当然这种可能性极低

4.2.2 redisson的使用

redisson提供了一系列分布式的java常用对象，其中包括各种分布式锁的实现，下面是使用redisson的步骤：

引入redisson依赖

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.13.6</version>
</dependency>

配置redisson客户端

@Configuration
public class RedisConfig {
    @Bean
    public RedissonClient redissonClient () {
        Config config = new Config();
        config.useSingleServer().setAddress("redis://192.168.223.101:6379");
        return Redisson.create(config);
    }
}

使用redissonClient调用redisson方法

4.2.3 redisson解决上述四个问题的原理

可重入：
- redisson使用redis的hash数据解决来存储一把锁，当一个线程获取锁成功时，会将hash的value + 1，hash结构中的value代表该锁被获取的个数。

可重试：
- 使用redis的消息订阅以及信号量的机制实现。当获取锁失败，线程会在自己设置的等待时间内等待其他线程释放锁，若该时间段内其他现车给释放了锁，则会使用消息队列的机制发送给当前线程一条消息，该线程就可以再次尝试获取锁；否则将会返回false。
不会超时释放
- 当调用获取锁的方法时，如果给leaseTime赋值，那么会开启Watchdog机制，该机制可以无限期给当前线程任务重置超时时间直到锁释放

主从一致性分布式锁
- 当redis是主从集群模式时，需要使用multiLock，该方法需要获取所有独立的redis节点上的锁，才算获取锁成功