面试官：MySQL如何实现分布式读写锁？

1、先看个业务场景

对 X 资源，可以执行 2 种操作：W 操作、R 操作，2 种操作需要满足下面条件

(1)、执行操作的机器分布式在不同的节点中，也就是分布式的

(2)、W 操作是独享的，也就是说同一时刻只允许有一个操作者对 X 执行 W 操作

(3)、R 操作是共享的，也就是说同时可以有多个执行者对 X 资源执行 R 操作

(4)、W 操作和 R 操作是互斥的，什么意思呢？也就是说 W 操作和 R 操作不能同时存在

通俗点说：

如果当前 W 操作正在执行，此时有 R 操作请求过来，那么这个 R 请求只能等待或者执行失败

如果前有 R 操作正在执行，此时有 W 操作请求过来，那么这个 W 请求只能等待或者执行失败。

这种业务场景如果是单台虚拟机，在 java 中可以使用 ReadWriteLock 读写锁就可以实现了，但是今天我们要讨论的是操作者不在同一个 jvm 中，而是分布式在不同的节点，服务中。

大家可能在思考，哪里有这样的业务场景？

我之前做过 p2p，这里给大家举个 p2p 中的例子。

可能大家对 p2p 不了解，这里先介绍一下 p2p 的业务。

比如小明需要 10 万买车，但是手头上没钱，此时可以在 p2p 平台上申请一个 10 万的借款，然后 p2p 平台会发布一个借款项目，开始募集资金。

其他网民可以去投资这个项目，每个月借款人会进行还款，投资人会拿到收益。

当投资人每次投资的时候，会产生一份债权，可以把债权理解为借款人欠你钱的一个凭证。

如果投资人急着用钱，但是此时投资还未到期，此时你可以发起债权转让，将你的债权卖给给其他人，这样你就可以及时拿到本金了。

这里面涉及到 2 个关键的业务：借款人执行还款、投资人发起债权转让。

借款人还款：借款人执行还款的时候，会将资金发到投资人账户中，涉及到投资人账户资金的变动，还有债权信息的变化等，整个还款过程涉及到调用银行系统，过程比较复杂，耗时相对比较长。

债权转让：投资人发起债权转让，也涉及到债权的编号和投资人账户的资金的变动。

由于这 2 个业务都会操作债权记录和投资人账户资金，为了保证资金的正确性，降低系统的复杂度，我们是这么做的，让这 2 种业务互斥

• 某笔借款执行还款的过程中，那么这笔借款关联的所有债权记录不允许发起转让
• 如果某笔借款记录当前没有在还款处理中，那么这笔借款记录关联的债权都可以同时发起债权转让

开头提到的 X、W、R 三个对象，和我们这个业务场景对标一下，如下

2、解决问题的思路

mysql 大家都用过，mysql 中同时对一笔记录发起 update 操作的时候，mysql 会确保整个操作会排队执行，内部是互斥锁实现的，从而可以确保在并发修改数据的时候，数据的正确性，执行 update 的时候，会返回被更新的行数，这里我们就利用 mysql 这个特性来实现读写锁的功能。

2.1、创建读写锁表

在业务库创建一个锁表，如下：

create table t_read_write_lock(
    resource_id varchar(50) primary key not null comment '互斥资源id',
    w_count int not null default 0 comment '目前执行中的W操作数量' ,
    r_count int not null default 0 comment '目前执行中的R操作数量',
    version bigint not null default 0 comment '版本号，每次执行update的时候+1'
);

这里主要关注 3 个字段：

1、resource_id：互斥资源 id，比如上面的借款记录 id

2、w_count：当前执行 W 操作的数量

3、r_count：当前执行 R 操作的数量

下面来看 W 操作和 R 操作的实现。

2.2、W 操作过程

1、通过resource_id去t_read_write_lock查询，如果不存在，则插入一条记录，这里由于resource_id是主键，所以对于同一个resource_id只会有一个插入成功，这里用 $lock_record表示t_read_write_lock记录
2、判断lock_record.w_count ==0 && lock_record.r_count==0，如果为true继续向下，否则返回false，业务终止
3、获取锁，过程如下
    {
        3.1、开启事务
        3.2、int count = (update t_read_write_lock set w_count=1 where r_count = 0)
        3.3、提交事务;
    }
4、如果3.2的count==1，继续向下执行，否则终止业务
5、执行业务操作
6、释放锁，过程如下
    {
        6.1、开启事务
        6.2、update t_read_write_lock set w_count=0 where w_count = 1
        6.3、提交事务
    }

整个过程有个问题，不知道大家发现没有，如果执行到 5 之后，系统挂了，会出现什么情况？

业务执行完毕了，但是 w 锁却没有释放，这种后果就是死锁了，以后 r 操作就没法执行了。

我们来看看，如何改进？

需要添加一下上锁日志表，每次上锁成功，则记录一条日志，表结构如下

create table t_lock_log(
    id bigint primary key auto_increment comment '主键，自动增长'
    resource_id varchar(50) primary key not null comment '互斥资源id',
    lock_type smallint default 0 comment '锁类型,0:W锁，1:R锁',
    status smallint default 0 comment '状态，0：获取锁成功，1：业务执行完毕，2：锁被释放'，
    create_time bigint default 0 comment '记录创建时间',
    version bigint not null default 0 comment '版本号，每次执行update的时候+1'
);

如何使用呢？

下面看 W 过程的改进

1、通过resource_id去t_read_write_lock查询，如果不存在，则插入一条记录，这里由于resource_id是主键，所以对于同一个resource_id只会有一个插入成功，这里用 $lock_record表示t_read_write_lock记录
2、判断lock_record.w_count ==0 && lock_record.r_count==0，如果为true继续向下，否则返回false，业务终止
3、获取锁，过程如下
    {
        3.1、开启事务
        3.2、int count = (update t_read_write_lock set w_count=1 where r_count = 0)
        3.3、如果count==1，则插入一条上锁日志，锁类型是0，状态是0：insert t_lock_log (resource_id,lock_type,status,create_time) values (#{resource_id},0,0,'当前时间');
        3.4、提交事务;
    }
4、如果3.2的count==1，继续向下执行，否则终止业务
5、执行业务操作，业务操作过程如下
    {
        5.1、业务库开启事务
        5.2、执行业务
        5.3、更新锁日志记录的状态为1，条件中必须带上status=0：int updateLogCount = (update t_lock_log set status=1 where id=#{日志记录id} and status = 0)
        5.4、if(updateLogCount==1){
                5.5、提交事务
            }else{
                5.6、回滚事务【走到这里说明更新锁日志记录失败了，说明t_lock_log的status被其他地方改掉了，被防止死锁的job修改了】
            }
        }
6、释放锁，过程如下
    {
        6.1、开启事务
        6.2、释放锁：update t_read_write_lock set w_count=0 where w_count = 1 and resource_id = #{resource_id}
        6.3、更新锁日志状态为2：update t_lock_log set status=2 where id = #{日志记录id}
        6.4、提交事务
 }

2.3、死锁的处理

上面这个是正常流程，如果第 3 步执行完了，也就是上锁 W 锁成功，但是执行到第 6 步之前，系统挂了，此时 W 锁没有释放，会出现死锁。

此时我们需要一个 job，通过这个 job 来释放长时间还未释放的锁，比如过了 10 分钟，锁还未被释放的，job 的逻辑如下

1、获取10分钟之前锁未释放的锁日志列表：select * from t_lock_log where status in (0,1) and create_time+10分钟<=当前时
间的;
2、轮询获取的日志列表，释放锁，操作如下
    {
        2.1、开启事务
        2.2、if(t_lock_log.lock_type==0){
                //lock_type为0表示是W锁，下面准备释放W锁
                //先将日志状态更新为2，注意条件中带上version作为条件，这里使用到了乐观锁，可以确保并发修改时只有一个count的值为1
                int count = (update t_lock_log set status=2 where id = #{日志记录id} and version = #{日志记录.version})
            if(count==1){
                //将w_count置为0
                update t_read_write_lock set w_count=0 where w_count = 1 and resource_id = #{resource_id}
            }
        }else{
            //准备释放R锁
            //先将日志状态置为2
            int count = (update t_lock_log set status=2 where id = #{日志记录id} and version = #{日志记录.version})
            if(count==1){
            //将r_count置为r_count-1,注意条件中带上r_count - 1>=0
            update t_read_write_lock set r_count=r_count-1 where r_count - 1>=0 and resource_id = #{resource_id}
        }
    }
    2.3、提交事务
}

2.4、R 锁的过程

1、通过resource_id去t_read_write_lock查询，如果不存在，则插入一条记录，这里由于resource_id是主键，所以对于同一个resource_id只会有一个插入成功，这里用 $lock_record表示t_read_write_lock记录
2、判断lock_record.w_count ==0，如果为true继续向下，否则返回false，业务终止
3、获取锁，过程如下
    {
        3.1、开启事务
        3.2、int count = (update t_read_write_lock set r_count=r_count+1 where w_count = 0)
        3.3、如果count==1，则插入一条上锁日志，锁类型是1【表示R锁】，状态是0：insert t_lock_log (resource_id,lock_type,status,create_time) values (#{resource_id},1,0,'当前时间');
        3.4、提交事务;
    }
4、如果3.2的count==1，继续向下执行，否则终止业务
5、执行业务操作，业务操作过程如下
    {
        5.1、业务库开启事务
        5.2、执行业务
        5.3、更新锁日志记录的状态为1，条件中必须带上status=0：int updateLogCount = (update t_lock_log set status=1 where id=#{日志记录id} and status = 0)
        5.4、if(updateLogCount==1){
            5.5、提交事务
        }else{
            5.6、回滚事务【走到这里说明更新锁日志记录失败了，说明t_lock_log的status被其他地方改掉了，被防止死锁的job修改了】
        }
    }
6、释放锁，过程如下
    {
        6.1、开启事务
        6.2、释放锁：update t_read_write_lock set r_count=r_count-1 where r_count - 1 >= 0 and resource_id = #{resource_id}
        6.3、更新锁日志状态为2：update t_lock_log set status=2 where id = #{日志记录id}
        6.4、提交事务
    }

3、总结

本文主要介绍了如何使用 mysql 来实现读写锁，如何防止死锁，重点就是 2 张表，锁表和日志表，2 个表配合一个 job，就把问题解决了。

大家可以将上面代码转换为程序，结合 spring 的 aop 可以实现一个通用的 db 读写锁，有兴趣的可以试试，有问题欢加我微信 itsoku 交流。

4、Spring 系列

1. Spring 系列第 1 篇：为何要学 spring？
2. Spring 系列第 2 篇：控制反转（IoC）与依赖注入（DI）
3. Spring 系列第 3 篇：Spring 容器基本使用及原理
4. Spring 系列第 4 篇：xml 中 bean 定义详解(-)
5. Spring 系列第 5 篇：创建 bean 实例这些方式你们都知道？
6. Spring 系列第 6 篇：玩转 bean scope，避免跳坑里！
7. Spring 系列第 7 篇：依赖注入之手动注入
8. Spring 系列第 8 篇：自动注入（autowire）详解，高手在于坚持
9. Spring 系列第 9 篇：depend-on 到底是干什么的？
10. Spring 系列第 10 篇：primary 可以解决什么问题？
11. Spring 系列第 11 篇：bean 中的 autowire-candidate 又是干什么的？
12. Spring 系列第 12 篇：lazy-init：bean 延迟初始化
13. Spring 系列第 13 篇：使用继承简化 bean 配置(abstract & parent)
14. Spring 系列第 14 篇：lookup-method 和 replaced-method 比较陌生，怎么玩的？
15. Spring 系列第 15 篇：代理详解（Java 动态代理&cglib 代理）？
16. Spring 系列第 16 篇：深入理解 java 注解及 spring 对注解的增强（预备知识）
17. Spring 系列第 17 篇：@Configration 和@Bean 注解详解(bean 批量注册)
18. Spring 系列第 18 篇：@ComponentScan、@ComponentScans 详解(bean 批量注册)
19. Spring 系列第 18 篇：@import 详解(bean 批量注册)
20. Spring 系列第 20 篇：@Conditional 通过条件来控制 bean 的注册
21. Spring 系列第 21 篇：注解实现依赖注入（@Autowired、@Resource、@Primary、@Qulifier）
22. Spring 系列第 22 篇：@Scope、@DependsOn、@ImportResource、@Lazy 详解
23. Spring 系列第 23 篇：Bean 生命周期详解
24. Spring 系列第 24 篇：父子容器详解
25. Spring 系列第 25 篇：@Value【用法、数据来源、动态刷新】
26. Spring 系列第 26 篇：国际化详解
27. Spring 系列第 27 篇：spring 事件机制详解
28. Spring 系列第 28 篇：Bean 循环依赖详解
29. Spring 系列第 29 篇：BeanFactory 扩展（BeanFactoryPostProcessor、BeanDefinitionRegistryPostProcessor）
30. Spring 系列第 30 篇：jdk 动态代理和 cglib 代理
31. Spring 系列第 31 篇：aop 概念详解
32. Spring 系列第 32 篇：AOP 核心源码、原理详解
33. Spring 系列第 33 篇：ProxyFactoryBean 创建 AOP 代理
34. Spring 系列第 34 篇：@Aspect 中@Pointcut 12 种用法
35. Spring 系列第 35 篇：@Aspect 中 5 中通知详解
36. Spring 系列第 36 篇：@EnableAspectJAutoProxy、@Aspect 中通知顺序详解
37. Spring 系列第 37 篇：@EnableAsync & @Async 实现方法异步调用
38. Spring 系列第 38 篇：@Scheduled & @EnableScheduling 定时器详解
39. Spring 系列第 39 篇：强大的 Spel 表达式
40. Spring 系列第 40 篇：缓存使用（@EnableCaching、@Cacheable、@CachePut、@CacheEvict、@Caching、@CacheConfig）
41. Spring 系列第 41 篇：@EnableCaching 集成 redis 缓存
42. Spring 系列第 42 篇：玩转 JdbcTemplate
43. Spring 系列第 43 篇：spring 中编程式事务怎么用的？
44. Spring 系列第 44 篇：详解 spring 声明式事务(@Transactional)
45. Spring 系列第 45 篇：带你吃透 Spring 事务 7 种传播行为
46. Spring 系列第 46 篇：Spring 如何管理多数据源事务？
47. Spring 系列第 47 篇：spring 编程式事务源码解析
48. Spring 系列第 48 篇：@Transaction 事务源码解析
49. Spring 系列第 49 篇：通过 Spring 事务实现 MQ 中的事务消息
50. Spring 系列第 50 篇：spring 事务拦截器顺序如何控制？
51. Spring 系列第 51 篇：导致 Spring 事务失效常见的几种情况
52. Spring 系列第 52 篇：Spring 实现数据库读写分离
53. Spring 系列第 53 篇：Spring 集成 MyBatis
54. Spring 系列第 54 篇：集成 junit
55. Spring 系列第 55 篇：spring 上下文生命周期
56. Spring 系列第 56 篇：面试官：循环依赖不用三级缓存可以么？

5、更多好文章

1. Java 高并发系列（共 34 篇）
2. MySql 高手系列（共 27 篇）
3. Maven 高手系列（共 10 篇）
4. Mybatis 系列（共 12 篇）
5. 聊聊 db 和缓存一致性常见的实现方式
6. 接口幂等性这么重要，它是什么？怎么实现？
7. 泛型，有点难度，会让很多人懵逼，那是因为你没有看这篇文章！