死磕 java同步系列之mysql分布式锁

忘是亡心i 2023-08-17 17:42 127阅读 0赞

## 问题 ##

（1）什么是分布式锁？

（2）为什么需要分布式锁？

（3）mysql如何实现分布式锁？

（4）mysql分布式锁的优点和缺点？

## 简介 ##

随着并发量的不断增加，单机的服务迟早要向多节点或者微服务进化，这时候原来单机模式下使用的synchronized或者ReentrantLock将不再适用，我们迫切地需要一种分布式环境下保证线程安全的解决方案，今天我们一起来学习一下mysql分布式锁如何实现分布式线程安全。

## 基础知识 ##

mysql中提供了两个函数——`get_lock('key', timeout)`和`release_lock('key')`——来实现分布式锁，可以根据`key`来加锁，这是一个字符串，可以设置超时时间（单位：秒），当调用`release_lock('key')`或者`客户端断线`的时候释放锁。

它们的使用方法如下：

mysql> select get_lock('user_1', 10);
        -> 1
    mysql> select release_lock('user_1');
        -> 1

`get_lock('user_1', 10)`如果10秒之内获取到锁则返回1，否则返回0；

`release_lock('user_1')`如果该锁是当前客户端持有的则返回1，如果该锁被其它客户端持有着则返回0，如果该锁没有被任何客户端持有则返回null；

## 多客户端案例 ##

为了便于举例【本篇文章由“彤哥读源码”原创，请支持原创，谢谢！】，这里的超时时间全部设置为0，也就是立即返回。

<table> 
 <thead> 
  <tr> 
   <th align="center">时刻</th> 
   <th align="center">客户端A</th> 
   <th align="center">客户端B</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="center">1</td> 
   <td align="center">get_lock(‘user_1’, 0) -&gt; 1</td> 
   <td align="center">-</td> 
  </tr> 
  <tr> 
   <td align="center">2</td> 
   <td align="center">-</td> 
   <td align="center">get_lock(‘user_1’, 0) -&gt; 0</td> 
  </tr> 
  <tr> 
   <td align="center">3</td> 
   <td align="center">-</td> 
   <td align="center">release_lock(‘user_1’, 0) -&gt; 0</td> 
  </tr> 
  <tr> 
   <td align="center">4</td> 
   <td align="center">release_lock(‘user_1’, 0) -&gt; 1</td> 
   <td align="center">-</td> 
  </tr> 
  <tr> 
   <td align="center">5</td> 
   <td align="center">release_lock(‘user_2’, 0) -&gt; null</td> 
   <td align="center">-</td> 
  </tr> 
  <tr> 
   <td align="center">6</td> 
   <td align="center">-</td> 
   <td align="center">get_lock(‘user_1’, 0) -&gt; 1</td> 
  </tr> 
  <tr> 
   <td align="center">7</td> 
   <td align="center">-</td> 
   <td align="center">release_lock(‘user_1’, 0) -&gt; 1</td> 
  </tr> 
 </tbody> 
</table>

## Java实现 ##

为了方便快速实现，这里使用 springboot2.1 mybatis 实现，并且省略spring的配置，只列举主要的几个类。

### 定义Locker接口 ###

接口中只有一个方法，入参1为加锁的key，入参2为执行的命令。

public interface Locker { 
        void lock(String key, Runnable command);
    }

### mysql分布式锁实现 ###

mysql的实现中要注意以下两点：

（1）加锁、释放锁必须在同一个session（同一个客户端）中，所以这里不能使用Mapper接口的方式调用，因为Mapper接口有可能会导致不在同一个session。

（2）可重入性是通过ThreadLocal保证的；

@Slf4j
    @Component
    public class MysqlLocker implements Locker { 
    
        private static final ThreadLocal<SqlSessionWrapper> localSession = new ThreadLocal<>();
    
        @Autowired
        private SqlSessionFactory sqlSessionFactory;
    
        @Override
        public void lock(String key, Runnable command) { 
            // 加锁、释放锁必须使用同一个session
            SqlSessionWrapper sqlSessionWrapper = localSession.get();
            if (sqlSessionWrapper == null) { 
                // 第一次获取锁
                localSession.set(new SqlSessionWrapper(sqlSessionFactory.openSession()));
            }
            try { 
                // 【本篇文章由“彤哥读源码”原创，请支持原创，谢谢！】
                // -1表示没获取到锁一直等待
                if (getLock(key, -1)) { 
                    command.run();
                }
            } catch (Exception e) { 
                log.error("lock error", e);
            } finally { 
                releaseLock(key);
            }
        }
    
        private boolean getLock(String key, long timeout) { 
            Map<String, Object> param = new HashMap<>();
            param.put("key", key);
            param.put("timeout", timeout);
            SqlSessionWrapper sqlSessionWrapper = localSession.get();
            Integer result = sqlSessionWrapper.sqlSession.selectOne("LockerMapper.getLock", param);
            if (result != null && result.intValue() == 1) { 
                // 获取到了锁，state加1
                sqlSessionWrapper.state  ;
                return true;
            }
            return false;
        }
    
        private boolean releaseLock(String key) { 
            SqlSessionWrapper sqlSessionWrapper = localSession.get();
            Integer result = sqlSessionWrapper.sqlSession.selectOne("LockerMapper.releaseLock", key);
            if (result != null && result.intValue() == 1) { 
                // 释放锁成功，state减1
                sqlSessionWrapper.state--;
                // 当state减为0的时候说明当前线程获取的锁全部释放了，则关闭session并从ThreadLocal中移除
                if (sqlSessionWrapper.state == 0) { 
                    sqlSessionWrapper.sqlSession.close();
                    localSession.remove();
                }
                return true;
            }
            return false;
        }
    
        private static class SqlSessionWrapper { 
            int state;
            SqlSession sqlSession;
    
            public SqlSessionWrapper(SqlSession sqlSession) { 
                this.state = 0;
                this.sqlSession = sqlSession;
            }
        }
    }

### LockerMapper.xml ###

定义get\_lock()、release\_lock()的语句。

<?xml version="1.0" encoding="UTF-8"?>
    <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
    
    <mapper namespace="LockerMapper">
        <select id="getLock" resultType="integer">
            select get_lock(#{key}, #{timeout});
        </select>
    
        <select id="releaseLock" resultType="integer">
            select release_lock(#{key})
        </select>
    </mapper>

### 测试类 ###

这里启动1000个线程，每个线程打印一句话并睡眠2秒钟。

@RunWith(SpringRunner.class)
    @SpringBootTest(classes = Application.class)
    public class MysqlLockerTest { 
    
        @Autowired
        private Locker locker;
    
        @Test
        public void testMysqlLocker() throws IOException { 
            for (int i = 0; i < 1000; i  ) { 
                // 多节点测试
                try { 
                    Thread.sleep(2000);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
                new Thread(()->{ 
                    locker.lock("lock", ()-> { 
                        // 可重入性测试
                        locker.lock("lock", ()-> { 
                            System.out.println(String.format("time: %d, threadName: %s", System.currentTimeMillis(), Thread.currentThread().getName()));
                            try { 
                                Thread.sleep(2000);
                            } catch (InterruptedException e) { 
                                e.printStackTrace();
                            }
                        });
                    });
                }, "Thread-" i).start();
            }
    
            System.in.read();
        }
    }

### 运行结果 ###

查看运行结果发现每隔2秒打印一个线程的信息，说明这个锁是有效的，至于分布式环境下面的验证也很简单，起多个MysqlLockerTest实例即可。

time: 1568715905952, threadName: Thread-3
    time: 1568715907955, threadName: Thread-4
    time: 1568715909966, threadName: Thread-8
    time: 1568715911967, threadName: Thread-0
    time: 1568715913969, threadName: Thread-1
    time: 1568715915972, threadName: Thread-9
    time: 1568715917975, threadName: Thread-6
    time: 1568715919997, threadName: Thread-5
    time: 1568715921999, threadName: Thread-7
    time: 1568715924001, threadName: Thread-2

## 总结 ##

（1）分布式环境下需要使用分布式锁，单机的锁将无法保证线程安全；

（2）mysql分布式锁是基于`get_lock('key', timeout)`和`release_lock('key')`两个函数实现的；

（3）mysql分布式锁是可重入锁；

## 彩蛋 ##

**使用mysql分布式锁需要注意些什么呢？**

答：必须保证多个服务节点使用的是同一个mysql库【本篇文章由“彤哥读源码”原创，请支持原创，谢谢！】。

**mysql分布式锁具有哪些优点？**

答：1）方便快捷，因为基本每个服务都会连接数据库，但是不是每个服务都会使用redis或者zookeeper；

2）如果客户端断线了会自动释放锁，不会造成锁一直被占用；

3）mysql分布式锁是可重入锁，对于旧代码的改造成本低；

**mysql分布式锁具有哪些缺点？**