基于数据库实现分布式锁

前言

随着分布式架构的广泛应用，基于分布式环境下产生的并发问题也越来越多，如在分布式环境下确保并发时的数据一致性问题成为很多开发人员亟待解决的问题

解决方案

分布式环境下，通常解决并发时数据一致性问题的方案主要是通过分布式锁进行解决。一把来说，应用部署在单机上，通过简单的JDK锁即可保证并发环境的数据安全性，但是一旦跨越JVM进程进行分布式部署时，JDK锁就无能无能为力了

既然是为了保证数据库数据的一致性，抛开性能，从实现方案上来说，就有很多种，比如可以利用数据库本身的锁,mysql的行锁，redis实现的分布式锁，zookeeper实现的分布式锁等，下面就这3种锁的实现做一一的说明，为后续的工作中的应用提供一个思路

环境准备

为演示方便，建议提前搭一个基于springboot的工程，我这里的演示demo使用springboot+mybatisplus的结构，下面贴出关键配置

1、pom依赖

<parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.2.1.RELEASE</version><relativePath/> <!-- lookup parent from repository --></parent><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding><java.version>1.8</java.version><mysql-connector-java.version>8.0.11</mysql-connector-java.version><commons-lang3.version>3.7</commons-lang3.version><fastjson.version>1.2.47</fastjson.version><mybatis-plus-boot-starter.version>3.3.0</mybatis-plus-boot-starter.version><mybatis-plus-generator.version>3.3.0</mybatis-plus-generator.version><druid.version>1.1.14</druid.version><lombok.version>1.18.0</lombok.version><redis.version>2.0.7.RELEASE</redis.version><swagger.version>2.9.2</swagger.version><swagger-bootstrap-ui.version>1.9.6</swagger-bootstrap-ui.version><guava.version>23.0</guava.version><easyexcel.version>2.1.6</easyexcel.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId><version>2.2.1.RELEASE</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><!--mysql依赖--><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>${mysql-connector-java.version}</version></dependency><!--阿里巴巴fastjosn依赖--><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>${fastjson.version}</version></dependency><!--阿里巴巴数据库连接池依赖--><!-- Druid --><dependency><groupId>com.alibaba</groupId><artifactId>druid-spring-boot-starter</artifactId><version>${druid.version}</version></dependency><dependency><groupId>org.mybatis.spring.boot</groupId><artifactId>mybatis-spring-boot-starter</artifactId><version>2.1.1</version></dependency><!-- MyBatis增强工具--><dependency><groupId>com.baomidou</groupId><artifactId>mybatis-plus-boot-starter</artifactId><version>${mybatis-plus-boot-starter.version}</version></dependency><dependency><groupId>com.baomidou</groupId><artifactId>mybatis-plus-generator</artifactId><version>${mybatis-plus-generator.version}</version></dependency><!-- lombok --><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>${lombok.version}</version></dependency><!-- redis --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId><version>${redis.version}</version></dependency><!--swagger-ui--><dependency><groupId>io.springfox</groupId><artifactId>springfox-swagger2</artifactId><version>${swagger.version}</version></dependency><dependency><groupId>io.springfox</groupId><artifactId>springfox-swagger-ui</artifactId><version>${swagger.version}</version></dependency><dependency><groupId>com.github.xiaoymin</groupId><artifactId>swagger-bootstrap-ui</artifactId><version>${swagger-bootstrap-ui.version}</version></dependency><!-- guava--><dependency><groupId>com.google.guava</groupId><artifactId>guava</artifactId><version>${guava.version}</version></dependency><!-- 阿里easyexcel--><dependency><groupId>com.alibaba</groupId><artifactId>easyexcel</artifactId><version>2.2.3</version></dependency><!-- https://mvnrepository.com/artifact/org.redisson/redisson --><dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.13.1</version></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build>

2、yml配置

server:port: 7748#数据库连接配置
spring:datasource:driver-class-name: com.mysql.cj.jdbc.Driverurl: jdbc:mysql://16.15.39.176:3306/test?autoReconnect=true&useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8&useSSL=falseusername: rootpassword: root#mybatisplus的配置
mybatis-plus:mapper-locations: classpath*:mapper/*.xmlglobal-config:db-column-underline: true  #开启驼峰转换db-config:id-type: uuidfield-strategy: not_nullrefresh: trueconfiguration:map-underscore-to-camel-case: truelog-impl: org.apache.ibatis.logging.stdout.StdOutImpl #打印sql语句便于调试

3、mybatisplus的配置类

@Configuration
@Slf4j
@MapperScan(basePackages = {"com.congge.mapper",})
public class MyBatisConfig {/*** 分页插件配置* @return*/@Beanpublic PaginationInterceptor paginationInterceptor() {return new PaginationInterceptor();}}

mybatis默认持久化字段配置

@Configuration
public class MetaObjectHandlerConfig implements MetaObjectHandler {@Overridepublic void insertFill(MetaObject metaObject) {Date currentDate = new Date();setFieldValByName("createDate",currentDate,metaObject);setFieldValByName("createBy","admin",metaObject);setFieldValByName("delFlag",0,metaObject);}@Overridepublic void updateFill(MetaObject metaObject) {Date currentDate = new Date();setFieldValByName("updateDate",currentDate,metaObject);setFieldValByName("updateBy","admin",metaObject);setFieldValByName("delFlag",0,metaObject);}
}

4、实体类(其他的参考即可)

@Data
@TableName("t_order")
public class Order extends BasePlusEntity<Order> implements Serializable {private String id;@TableField("order_status")private Integer orderStatus;@TableField("receiver_name")private String receiverName;@TableField("receiver_phone")private String receiverPhone;@TableField("order_amount")private Double orderAmount;@Overrideprotected Serializable pkVal() {return this.id;}}


import com.alibaba.fastjson.annotation.JSONField;
import com.baomidou.mybatisplus.annotation.FieldFill;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableLogic;
import com.baomidou.mybatisplus.extension.activerecord.Model;
import com.fasterxml.jackson.annotation.JsonFormat;
import com.fasterxml.jackson.annotation.JsonIgnore;
import lombok.Data;import java.io.Serializable;
import java.util.Date;/*** 共有属性继承类*/
@Data
public class BasePlusEntity<T extends Model> extends Model implements Serializable {private static final long serialVersionUID = 1L;protected String id;protected String remarks;@TableField(value = "create_date", fill = FieldFill.INSERT)@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")protected Date createDate;@TableField(value = "update_date", fill = FieldFill.UPDATE)@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")protected Date updateDate;@TableField(value = "create_by", fill = FieldFill.INSERT)protected String createBy;@TableField(value = "update_by", fill = FieldFill.UPDATE)protected String updateBy;@JSONField(serialize = false)@TableField(value = "del_flag", fill = FieldFill.INSERT_UPDATE)@TableLogic@JsonIgnoreprotected Integer delFlag;public BasePlusEntity(String id) {this();this.id = id;}public BasePlusEntity() {}}

以上的基础准备工作就到此结束，其余的包结构想必看到这儿的小伙伴们都很熟悉了，就是服务接口，实现类和mapper了

业务描述

生成一笔订单，对商品进行扣库存操作

涉及到的表

CREATE TABLE `t_product` (`id` varchar(128) NOT NULL COMMENT 'id',`product_name` varchar(255) DEFAULT '' COMMENT '商品名',`price` decimal(2,0) NOT NULL COMMENT '商品价格',`count` int(12) NOT NULL COMMENT '商品库存数',`product_desc` varchar(512) DEFAULT NULL COMMENT '排序',`remarks` varchar(512) DEFAULT '' COMMENT '备注',`update_date` datetime DEFAULT NULL COMMENT '更新时间',`update_by` varchar(64) DEFAULT '' COMMENT '更新人',`create_date` datetime DEFAULT NULL COMMENT '创建时间',`create_by` varchar(64) DEFAULT '' COMMENT '创建人',`del_flag` char(1) NOT NULL DEFAULT '0' COMMENT '删除标志 0正常 1删除',PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8 ROW_FORMAT=DYNAMIC COMMENT='用户表';

CREATE TABLE `t_order` (`id` varchar(128) NOT NULL COMMENT 'id',`order_status` int(2) NOT NULL DEFAULT '0' COMMENT '订单状态',`receiver_name` varchar(24) DEFAULT NULL COMMENT '收货人',`receiver_phone` varchar(32) DEFAULT '收货人电话',`order_amount` decimal(2,0) DEFAULT NULL COMMENT '订单总金额',`remarks` varchar(512) DEFAULT '' COMMENT '备注',`update_date` datetime DEFAULT NULL COMMENT '更新时间',`update_by` varchar(64) DEFAULT '' COMMENT '更新人',`create_date` datetime DEFAULT NULL COMMENT '创建时间',`create_by` varchar(64) DEFAULT '' COMMENT '创建人',`del_flag` char(1) NOT NULL DEFAULT '0' COMMENT '删除标志 0正常 1删除',PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8 ROW_FORMAT=DYNAMIC COMMENT='订单表';

CREATE TABLE `t_order_item` (`id` varchar(128) NOT NULL COMMENT 'id',`order_id` varchar(128) NOT NULL COMMENT '订单ID',`product_id` int(12) NOT NULL COMMENT '商品ID',`purchase_price` decimal(2,0) NOT NULL COMMENT '购买金额',`purchase_num` int(12) NOT NULL COMMENT '购买数量',`remarks` varchar(512) DEFAULT '' COMMENT '备注',`update_date` datetime DEFAULT NULL COMMENT '更新时间',`update_by` varchar(64) DEFAULT '' COMMENT '更新人',`create_date` datetime DEFAULT NULL COMMENT '创建时间',`create_by` varchar(64) DEFAULT '' COMMENT '创建人',`del_flag` char(1) NOT NULL DEFAULT '0' COMMENT '删除标志 0正常 1删除',PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8 ROW_FORMAT=DYNAMIC COMMENT='订单详情表';

以上简化了业务员，最终的效果就是，订单表新增一条数据，订单详情表增加一条数据，这里初始化一条商品数据

提供一个产生订单的接口

 @GetMapping("/order/create")public ResponseResult createOrder() {logger.info("进入方法");return orderService.createOrder();}

业务实现类

@Service
@Slf4j
public class OrderServiceImpl implements OrderService {@Autowiredprivate OrderMapper orderMapper;@Autowiredprivate ProductMapper productMapper;@Autowiredprivate OrderItemMapper orderItemMapper;@Autowiredprivate PlatformTransactionManager platformTransactionManager;@Autowiredprivate TransactionDefinition transactionDefinition;private String productId = "0001";private int purchaseProductNum = 1;@Overridepublic ResponseResult getOrderById(String id) {Order order = orderMapper.selectById(id);return ResponseResult.success(order, 200);}private String insertOrder(Product product) {Order order = new Order();String orderId = UUID.randomUUID().toString().replaceAll("-", "").substring(0, 16);order.setId(orderId);order.setOrderAmount(product.getPrice() * purchaseProductNum);order.setOrderStatus(1);order.setReceiverName("zhangsan");order.setReceiverPhone("13323412345");orderMapper.insert(order);return orderId;}private void insertOrderItem(Product product, String orderId) {OrderItem orderItem = new OrderItem();orderItem.setOrderId(orderId);orderItem.setProductId(productId);orderItem.setPurchasePrice(product.getPrice());orderItem.setPurchaseNum(purchaseProductNum);orderItemMapper.insert(orderItem);}@Override@Transactional(rollbackFor = Exception.class)public synchronized ResponseResult createOrder() {Product product = productMapper.selectById(productId);if(product==null){throw new BusinessException("购买的商品不存在");}Integer currCount = product.getCount();if(purchaseProductNum > currCount){log.info("购买的商品库存数量不够了，购买的数量是:{},实际库存数是:{}",purchaseProductNum,product.getCount());throw new BusinessException("购买的商品库存数量不够了");}Integer leftCount = currCount-purchaseProductNum;product.setCount(leftCount);//更新商品的库存productMapper.updateById(product);//订单表和订单详情表各自插入一条数据String orderId = insertOrder(product);insertOrderItem(product, orderId);return ResponseResult.success(200,"订单创建成功");}}

我们知道，单进程时，可以通过synchronized关键字或者lock进行并发控制

提供一个并发测试类，以供测试使用


@RunWith(SpringRunner.class)
@SpringBootTest
public class OrderTest {@Autowiredprivate OrderService orderService;@Testpublic void testConcurrenOrder() throws InterruptedException {CountDownLatch cdl = new CountDownLatch(5);CyclicBarrier cyclicBarrier = new CyclicBarrier(5);ExecutorService service = Executors.newFixedThreadPool(5);for(int i=0;i<5;i++){service.execute(()->{try {cyclicBarrier.await();orderService.createOrder();System.out.println("产生订单");} catch (InterruptedException e) {e.printStackTrace();} catch (BrokenBarrierException e) {e.printStackTrace();}finally {cdl.countDown();}});}cdl.await();service.shutdown();}}

但是如果这是分布式的环境，synchronized关键字或者lock也不好使了，因为他们都只能锁住当前进程的这个方法

方案1，利用mysql数据库的行锁进行并发控制

多个进程或者多个线程访问共同的数据库资源时，利用mysql的行锁机制实现并发控制，即 "select … for update "语句进行控制，在此例中，其关键性的查询语句为:

 <select id="selectProductById" resultType="com.congge.entity.Product">select * from t_product where `id` = #{productId} for update;</select>

为了放大效果，我们将上面创建订单的逻辑改造如下：

 @Override@Transactional(rollbackFor = Exception.class)public synchronized ResponseResult createOrder() {log.info("准备创建订单...");Product product = productMapper.selectProductById(productId);try {Thread.sleep(60000);} catch (InterruptedException e) {e.printStackTrace();}if(product==null){throw new BusinessException("购买的商品不存在");}Integer currCount = product.getCount();if(purchaseProductNum > currCount){log.info("购买的商品库存数量不够了，购买的数量是:{},实际库存数是:{}",purchaseProductNum,product.getCount());throw new BusinessException("购买的商品库存数量不够了");}Integer leftCount = currCount-purchaseProductNum;product.setCount(leftCount);//更新商品的库存productMapper.updateById(product);//订单表和订单详情表各自插入一条数据String orderId = insertOrder(product);insertOrderItem(product, orderId);return ResponseResult.success(200,"订单创建成功");}

测试：

我们将当前的服务在复制一份出来，使用不同的端口进行区分，idea中的配置比较简单：

以上的准备工作完成之后，启动两个服务，同时调用两个创建订单的接口：

http://localhost:7749/order/create
http://localhost:7748/order/create

观察两个服务的控制台的打印输出结果：

多测试几次，还会出现下面的结果，但是都在预料之中

同时我们去观察数据库的相关表，产品表的数量为0，订单表和订单明细表各自产生一条数据

通过上面的分析和产生的数据基本上可以总结如下，通过mysql的查询行锁语句，达到了并发时的控制，保证了数据的安全性，在实际生产中，比较典型的场景就是超卖问题

本篇到此结束，最后感谢观看！(需要源码的同学可以私信我)