面试官：MySQL如何实现分布式读写锁？
以下文章来源于路人甲Java ，作者路人甲java
工作10年的前阿里P7，所有文章以系列的方式呈现，带领大家成为java高手，目前已出：java高并发系列、mysql高手系列、Maven高手系列、mybatis系列、spring系列，正在连载springcloud系列，欢迎关注！
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 三个对象，和我们这个业务场景对标一下，如下
| X 表示资源 | W 操作 | R 操作 |
|---|---|---|
| 标的 id | 还款操作 | 债权转让 |
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 读写锁，有兴趣的可以试试
推荐阅读：
欢迎关注微信公众号：互联网全栈架构，收取更多有价值的信息。