分布式锁问题_演示问题

通过idea创建两个服务

启动Nginx服务

下载Nginx windows服务，官网nginx: download

当然我这里提供了：

我们打开nginx的conf目录，然后打开配置文件nginx.conf进行配置：

upstream test{

server localhost:9090 ;

server localhost:9091 ;

}

server {

listen 80;

server_name localhost;

#charset koi8-r;

#access_log logs/host.access.log main;

location / {

proxy_pass http://test;

}

当然我已经帮你配置好了，含义就是当我请求80端口服务的时候nginx帮我代理上面的两个服务，负载均衡类型为轮询

启动nginx即可

然后我们把数据库的库存修改为2，其他表的数据清空，一会我要进行测试

启动9090和9091服务

打开Jmeter进行测试，把端口进行修改成80

启动测试，它有可能会发生问题

你会发现数据库创建了3个订单，说明出现了并发问题【其实就是分布式下锁出现的问题】

分布式锁解决方案

分布式的CAP理论告诉我们“任何一个分布式系统都无法同时满足一致性（Consistency）、可用性（Availability）和分区容错性（Partition tolerance），最多只能同时满足两项。”所以，很多系统在设计之初就要对这三者做出取舍。在互联网领域的绝大多数的场景中，都需要牺牲强一致性来换取系统的高可用性，系统往往只需要保证“最终一致性”，只要这个最终时间是在用户可以接受的范围内即可。

分布式锁实现方案

基于数据库实现的分布式锁

基于数据库实现分布式锁主要是利用数据库的唯一索引来实现，唯一索引天然具有排他性，这刚好符合我们对锁的要求：同一时刻只能允许一个竞争者获取锁。【我们这里会通过两种操作来实现1悲观锁，2乐观锁】

基于 Redis 实现的分布式锁

使用Redis来实现分布式锁的效率最高，加锁速度最快，因为Redis几乎都是纯内存操作，而基于数据库的方案和基于Zookeeper的方案都会涉及到磁盘文件IO，效率相对低下。一般使用Redis来实现分布式锁都是利用Redis的SETNX key value这个命令，只有当key不存在时才会执行成功，如果key已经存在则命令执行失败。

基于 Zookeeper 实现的分布式锁

Zookeeper一般用作配置中心，其实现分布式锁的原理和Redis类似，我们在Zookeeper中创建临时顺序节点，利用节点不能重复创建的特性来保证排他性。

分布式锁解决方案_数据库悲观锁实现的分布式锁

什么是悲观锁

顾名思义，就是比较悲观的锁，总是假设最坏的情况，每次去拿数据的时候都认为别人会修改，所以每次在拿数据的时候都会上锁，这样别人想拿这个数据就会阻塞直到它拿到锁。

通过for upate进行上锁操作

在Mysql中开启2个命令行界面

现第二个客户端什么都没查询出来，因为第一个客户端查询数据的使用因为for update 语句持有了1001这条数据的锁，如果我不提交事物【commit命令】，客户端二是不能查询出来数据的这就是我们所说的悲观锁

如果我么commit

我们客户端一commit后释放了锁，那么客户端二才能查询出来数据

下面我们在程序中通过悲观锁来解决问题

修改mapper包下的接口ProductMapper，创建查询方法：

package com.ss.demo.mapper;

import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.ss.demo.domain.Product;
import org.apache.ibatis.annotations.Param;

/**
* 
* Mapper 接口
* 
*/
public interface ProductMapper extends BaseMapper<Product> {
Product findById(@Param("id") Integer id);
}

修改映射文件ProductMapper.xml:

<?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="com.ss.demo.mapper.ProductMapper">

 
 <resultMap id="BaseResultMap" type="com.ss.demo.domain.Product">
 <id column="id" property="id" />
 <result column="product_name" property="productName" />
 <result column="price" property="price" />
 <result column="count" property="count" />
 <result column="product_desc" property="productDesc" />
 <result column="version" property="version" />
 </resultMap>

 
 <select id="findById" resultType="com.ss.demo.domain.Product">
 SELECT * FROM product WHERE id = #{id} FOR UPDATE
 </select>

</mapper>

修改service在接口ITOrderService中添加悲观锁的方法

package com.ss.demo.service;

import com.baomidou.mybatisplus.extension.service.IService;
import com.ss.demo.domain.TOrder;

/**
* 
* 服务类
* 
*/
public interface ITOrderService extends IService<TOrder> {
 /**
 * 创建订单方法
 * @param productId
 * @param count
 * @return
 */
 String createOrder(Integer productId, Integer count);

 /**
 * 使用悲观锁进行实现
 * @param productId
 * @param count
 * @return
 */
 String createOrderPessimisticlock(Integer productId, Integer count);
}

实现类：TOrderServiceImpl

/**
* 悲观锁操作
* 创建订单操作
* @param productId
* @param count
* @return
*/

@Transactional
@Override
public String createOrderPessimisticlock(Integer productId, Integer count) {
    //根据商品id获取商品信息
    Product product = productMapper.findById(productId);
    if(product == null) {
        throw new RuntimeException("购买商品不存在");
    }
    log.info(Thread.currentThread().getName() +"库存数量" + product.getCount());
    //校验库存
    if(count > product.getCount()) {
        throw new RuntimeException("库存不足");
    }
    //更新库存
    Integer iCount = product.getCount() - count;
    product.setCount(iCount);
    //更新操作
    productMapper.updateById(product);
    //创建订单操作
    TOrder order = new TOrder();
    order.setOrderStatus(1);
    order.setReceiverName("张三");
    order.setReceiverMobile("12345678765");
    //设置订单价格【商品单价*商品数量】
    order.setOrderAmount(product.getPrice().multiply(new BigDecimal(count)));
    orderMapper.insert(order); //插入订单操作
    //创建订单商品表的操作
    OrderItem orderItem = new OrderItem();
    orderItem.setOrderId(order.getId());     //订单Id
    orderItem.setProduceId(product.getId()); //商品Id
    orderItem.setPurchasePrice(product.getPrice()); //购买价格
    orderItem.setPurchaseNum(count);   //购买数量
    orderItemMapper.insert(orderItem);
    return order.getId();
}

修改controller：TorderController

package com.ff.test.controller;
import com.ff.test.service.ITOrderService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.zip.CheckedOutputStream;

/**
* 
* 前端控制器
* 
*
* @author laozhang
* @since 2023-05-12
*/
@RestController
@RequestMapping("/order")
public class TOrderController {

 @Autowired
 private ITOrderService orderService;

 @PostMapping("/order")
 public String createOrder(Integer productId, Integer count) {
 //String orderId = orderService.createOrder(productId, count);
 String orderId = orderService.createOrderPessimisticlock(productId, count);
 return orderId;
 }

}

数据库商品表恢复2条数据，其他数据删除

启动9090,9091服务，并使用Jmeter进行测试：

解决问题

分布式锁解决方案_数据库乐观锁实现的分布式锁

乐观锁试用于读多的应用类型，可以提高吞吐量

什么是乐观锁

总是假设最好的情况，每次去拿数据的时候都认为别人不会修改，所以不会上锁，但是在更新的时候会判断一下在此期间别人有没有去更新这个数据，可以使用版本号机制和CAS算法实现。

乐观锁实现方式

取出记录时，获取当前 version

更新时，带上这个 version 执行

更新时， set version = newVersion where version = oldVersion

如果 version 不对，就更新失败

编写乐观锁更新语句

修改mapper包中的接口ProductMapper

package com.ss.demo.mapper;

import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.ss.demo.domain.Product;
import org.apache.ibatis.annotations.Param;

/**
* 
* Mapper 接口
* 
*
* @author laozhang
* @since 2023-04-04
*/
public interface ProductMapper extends BaseMapper<Product> {
 Product findById(@Param("id") Integer id);

 /**
 * 乐观锁操作
 * @param id
 * @param count
 * @return
 */
 int updateProductVersion(Integer id, Integer count, Integer version);
}

修改映射文件ProductMapper.xml:

<?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="com.ss.demo.mapper.ProductMapper">

 
 <resultMap id="BaseResultMap" type="com.ss.demo.domain.Product">
 <id column="id" property="id" />
 <result column="product_name" property="productName" />
 <result column="price" property="price" />
 <result column="count" property="count" />
 <result column="product_desc" property="productDesc" />
 <result column="version" property="version" />
 </resultMap>

 
 <select id="findById" resultType="com.ss.demo.domain.Product">
 SELECT * FROM product WHERE id = #{id} FOR UPDATE
 </select>

 
 <update id="updateProductVersion" parameterType="int" >
 UPDATE product SET count = count - #{count}, version = version + 1 WHERE id = #{id} AND count > 0 AND version = #{version}
 </update>
</mapper>

在项目中的service中进行操作修改接口ITOrderService

/**
* 乐观锁
* @param productId
* @param count
* @return
*/
String createOrderOptmisticlock(Integer productId, Integer count);

修改实现类ITOrderServiceImpl：

/**
* 乐观锁
* @param productId
* @param count
* @return
*/
@Transactional
@Override
public String createOrderOptmisticlock(Integer productId, Integer count) {
 int retryCount = 0; //重试次数
 int update = 0; //更新的结果

 //根据商品id获取商品信息
 Product product = productMapper.selectById(productId); //如果我们使用乐观锁就得换成普通查询
 if(product == null) {
 throw new RuntimeException("购买商品不存在");
 }
 log.info(Thread.currentThread().getName() +"库存数量" + product.getCount());
 //校验库存
 if(count > product.getCount()) {
 throw new RuntimeException("库存不足");
 }

 /**
 * 乐观锁更新库存
 * 更新失败，说明其他线程已经修改过数据，本地扣减库存失败了，可以进行重试
 * 最多重试3次
 */
 while(retryCount < 3 && update == 0) { //如果符合这这个条件我们重试3次
 update = this.reduceCount(productId, count);
 retryCount++;
 }
 if(update == 0) {
 throw new RuntimeException("库存不足");
 }

 //创建订单操作
 TOrder order = new TOrder();
 order.setOrderStatus(1);
 order.setReceiverName("张三");
 order.setReceiverMobile("12345678765");
 //设置订单价格【商品单价*商品数量】
 order.setOrderAmount(product.getPrice().multiply(new BigDecimal(count)));
 orderMapper.insert(order); //插入订单操作
 //创建订单商品表的操作
 OrderItem orderItem = new OrderItem();
 orderItem.setOrderId(order.getId()); //订单Id
 orderItem.setProduceId(product.getId()); //商品Id
 orderItem.setPurchasePrice(product.getPrice()); //购买价格
 orderItem.setPurchaseNum(count); //购买数量
 orderItemMapper.insert(orderItem);
 return order.getId();
}

/**
* 更新操作,减库存操作
* mysql的默认隔离级别为可重复读，导致在同一个事物里面查询3次商品
* 得到的数据始终是相同的，所以我们提供reduceCount方法，每次操作是都会提供一个新的事物，来去做扣减库存操作
* @param id
* @param count
* @return
*/
@Transactional
public int reduceCount(int id, int count) {
 int result = 0;
 //查询商品操作
 Product product = productMapper.selectById(id);
 //判断库存
 if(product.getCount() >= count) {
 //扣减库存我们这里使用乐观锁操作
 result = productMapper.updateProductVersion(product.getId(), count, product.getVersion());
 }
 return result;
}

修改controller：TOrderController

package com.ss.demo.controller;
import com.ss.demo.service.ITOrderService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.AutoConfigureOrder;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
* 
* 前端控制器
* 
*/
@RestController
@RequestMapping("/order")
public class TOrderController {
 @Autowired
 private ITOrderService orderService;

 @PostMapping("/create")
 public String createOrder(Integer productId, Integer count) {
 //return orderService.createOrder(productId, count);
 return orderService.createOrderOptmisticlock(productId,count);
 }
}