分布式锁之数据库实现

电玩女神 2022-12-07 15:08 131阅读 0赞

# 分布式锁之数据库实现 #

## 什么是分布式锁 ##

在单实例单进程的系统中，当有多个线程同时修改某个共享变量时，为了保证线程安全，就需要对变量或者代码做同步处理，这种同步操作在java中可以使用synchronized、JUC包下的显式锁、cas+volatile来实现。

而目前大部分系统都是分布式部署的，使用synchronized等手动只能保证单个进程内的线程安全，多个进程多个实例下的线程安全就需要分布式锁来实现。

目前主流的分布式锁解决方案有四种：

*  基于数据库实现（悲观+乐观）
 *  基于Redis实现
 *  基于ZooKeeper实现
 *  基于Etcd实现

## 数据库之悲观锁实现分布式锁 ##

建表sql：

CREATE TABLE `t_order` (
      `id` int(11) NOT NULL AUTO_INCREMENT,
      `amount` int(11) NOT NULL,
      `status` int(11) NOT NULL,
      `version` int(11) NOT NULL,
      PRIMARY KEY (`id`)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8
    
    INSERT INTO `t_order` (`amount`, `status`, `version`) VALUES ('100', '1', '1');

先借助sql来模拟分布锁的实现：

<table> 
 <thead> 
  <tr> 
   <th>步骤</th> 
   <th>SessionA</th> 
   <th>SessionB</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>1</td> 
   <td>begin;</td> 
   <td>begin;</td> 
  </tr> 
  <tr> 
   <td>2</td> 
   <td>select * from t_order where id=1 for update;</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>3</td> 
   <td></td> 
   <td>select * from t_order where id=1 for update; – 阻塞</td> 
  </tr> 
  <tr> 
   <td>4</td> 
   <td>update t_order set status=2 where id=1 and status=1;</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>5</td> 
   <td>commit;</td> 
   <td>返回查询结果</td> 
  </tr> 
  <tr> 
   <td>6</td> 
   <td></td> 
   <td>update t_order set status=2 where id=1 and status=1; – 状态变了未更新成功</td> 
  </tr> 
  <tr> 
   <td>7</td> 
   <td></td> 
   <td>commit;</td> 
  </tr> 
 </tbody> 
</table>

说明：

1.  客户端A和客户端B同时执行前面两行sql，客户端A返回数据，而客户端B阻塞等待获取行锁。
2.  客户端A执行后面两行sql，提交事务，客户端B获得行锁，立刻返回数据。
3.  客户端B执行后面两行sql，提交事务，释放行锁。

> 注意`for update`语句一定要走主键索引，否则没走索引会锁住整个表，走了其他索引会产生间隙锁，可能会锁住多条记录。

Java代码实现：

package com.morris.distribute.lock.database.exclusive;
    
    import com.morris.distribute.entity.Order;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jdbc.core.JdbcTemplate;
    import org.springframework.stereotype.Service;
    import org.springframework.transaction.annotation.Transactional;
    
    import java.util.concurrent.TimeUnit;
    
    @Service
    @Slf4j
    public class OrderService { 
    
        @Autowired
        private JdbcTemplate jdbcTemplate;
    
        /** * 数据库分布式锁之悲观锁 * * @param id */
        @Transactional
        public void updateStatus(int id) { 
            log.info("updateStatus begin, {}", id);
    
            Integer status = jdbcTemplate.queryForObject("select status from t_order where id=? for update", new Object[]{ id}, Integer.class);
    
            if (Order.ORDER_STATUS_NOT_PAY == status) { 
    
                try { 
                    // 模拟耗时操作
                    TimeUnit.SECONDS.sleep(3);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
    
                int update = jdbcTemplate.update("update t_order set status=? where id=? and status=1", new Object[]{ 2, id, Order.ORDER_STATUS_NOT_PAY});
    
                if (update > 0) { 
                    log.info("updateStatus success, {}", id);
                } else { 
                    log.info("updateStatus failed, {}", id);
                }
            } else { 
                log.info("updateStatus status already updated, ignore this request, {}", id);
            }
            log.info("updateStatus end, {}", id);
        }
    }

注意开启事务`@Transactional`。

使用多个线程模拟竞争锁：

package com.morris.distribute.lock.database.exclusive;
    
    import com.morris.distribute.config.JdbcConfig;
    import org.springframework.context.annotation.AnnotationConfigApplicationContext;
    
    import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    import java.util.stream.IntStream;
    
    public class Demo { 
    
        public static void main(String[] args) { 
            AnnotationConfigApplicationContext applicationContext = new AnnotationConfigApplicationContext();
            applicationContext.register(JdbcConfig.class);
            applicationContext.register(OrderService.class);
            applicationContext.refresh();
    
            CyclicBarrier cyclicBarrier = new CyclicBarrier(3);
            IntStream.rangeClosed(1, 3).forEach((i) -> new Thread(() -> { 
                OrderService orderService = applicationContext.getBean(OrderService.class);
                try { 
                    cyclicBarrier.await();
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                } catch (BrokenBarrierException e) { 
                    e.printStackTrace();
                }
                orderService.updateStatus(1);
            }, "t" + i).start());
        }
    }

运行结果如下：

2020-09-16 14:16:53,248  INFO [t2] (OrderService.java:26) - updateStatus begin, 1
    2020-09-16 14:16:53,248  INFO [t1] (OrderService.java:26) - updateStatus begin, 1
    2020-09-16 14:16:53,248  INFO [t3] (OrderService.java:26) - updateStatus begin, 1
    2020-09-16 14:16:56,289  INFO [t2] (OrderService.java:42) - updateStatus success, 1
    2020-09-16 14:16:56,289  INFO [t2] (OrderService.java:49) - updateStatus end, 1
    2020-09-16 14:16:56,290  INFO [t3] (OrderService.java:47) - updateStatus status already updated, ignore this request, 1
    2020-09-16 14:16:56,290  INFO [t3] (OrderService.java:49) - updateStatus end, 1
    2020-09-16 14:16:56,291  INFO [t1] (OrderService.java:47) - updateStatus status already updated, ignore this request, 1
    2020-09-16 14:16:56,291  INFO [t1] (OrderService.java:49) - updateStatus end, 1

从运行结果可以看出，同一时间只有一个线程持有了锁。

## 数据库之乐观锁实现分布式锁 ##

乐观锁每次通过版本号来判断记录是否被更新过。

package com.morris.distribute.lock.database.share;
    
    import com.morris.distribute.entity.Order;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jdbc.core.JdbcTemplate;
    import org.springframework.stereotype.Service;
    
    import java.util.concurrent.TimeUnit;
    
    @Service
    @Slf4j
    public class OrderService { 
    
        @Autowired
        private JdbcTemplate jdbcTemplate;
    
        /** * 数据库分布式锁之乐观锁 * * @param id */
        public void updateStatus(int id) { 
    
            log.info("updateStatus begin, {}", id);
            for (;;) {  // 自旋，有可能对订单做其他操作，导致version变了，所以需要自旋
                Order order = jdbcTemplate.queryForObject("select status, version from t_order where id=?",
                        new Object[]{ id}, (rs, row) -> { 
                            Order o = new Order();
                            o.setStatus(rs.getInt(1));
                            o.setVersion(rs.getInt(2));
                            return o;
                        });
    
                if (Order.ORDER_STATUS_NOT_PAY == order.getStatus()) { 
    
                    try { 
                        TimeUnit.SECONDS.sleep(3);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
    
                    int update = jdbcTemplate.update("update t_order set status=?,version=? where id=? and version=? and status=?",
                            new Object[]{ Order.ORDER_STATUS_PAY_SUCCESS, order.getVersion() + 1, id, order.getVersion(), Order.ORDER_STATUS_NOT_PAY});
    
                    if (update > 0) { 
                        log.info("updateStatus success, {}", id);
                        break;
                    } else { 
                        log.info("updateStatus failed, {}", id);
                    }
                } else { 
                    log.info("updateStatus status already updated, ignore this request, {}", id);
                    break;
                }
            }
            log.info("updateStatus end, {}", id);
        }
    
    }

运行结果如下：

2020-09-16 14:21:08,934  INFO [t3] (OrderService.java:25) - updateStatus begin, 1
    2020-09-16 14:21:08,934  INFO [t2] (OrderService.java:25) - updateStatus begin, 1
    2020-09-16 14:21:08,934  INFO [t1] (OrderService.java:25) - updateStatus begin, 1
    2020-09-16 14:21:12,110  INFO [t1] (OrderService.java:50) - updateStatus failed, 1
    2020-09-16 14:21:12,110  INFO [t2] (OrderService.java:50) - updateStatus failed, 1
    2020-09-16 14:21:12,111  INFO [t3] (OrderService.java:47) - updateStatus success, 1
    2020-09-16 14:21:12,111  INFO [t3] (OrderService.java:57) - updateStatus end, 1
    2020-09-16 14:21:12,117  INFO [t2] (OrderService.java:53) - updateStatus status already updated, ignore this request, 1
    2020-09-16 14:21:12,117  INFO [t1] (OrderService.java:53) - updateStatus status already updated, ignore this request, 1
    2020-09-16 14:21:12,117  INFO [t1] (OrderService.java:57) - updateStatus end, 1
    2020-09-16 14:21:12,117  INFO [t2] (OrderService.java:57) - updateStatus end, 1

## 总结 ##

悲观锁：会锁住整行记录，导致对数据的其他业务操作也无法进行，效率低，如果sql没写好，可能会产生间隙锁，锁住多条记录，甚至锁住全表。

乐观锁：每个表都需要增加与业务无关的version字段。

优点：直接基于数据库实现，实现简单。

缺点：IO开销大，连接数有限，无法满足高并发的需求。