前言：

随着笔者的颜值不断提高，用户量的日益增长，传统的单机方案已经不能满足产品的需求。笔者在网上寻遍方案，发现均为人云亦云，一份以毫秒为精度的轮询分布式锁被转发转载上万次。然，该方案没法满足笔者性能要求。故此，笔者研发 ELock 插件，并发布本文章。

其实集群也好，分布式服务也好。当我们不能保证团队成员的整体素质，那么在某些业务上，分布式锁自然没法避免。

公认开发原则

能不使用分布式锁的，尽可能不使用

举个例子

一个品交易，需要检查库存、检查余额、扣库存、扣款、生成订单。可能很多人觉得，在分布式环境下一定要分布式锁才能安全。

致此，笔者提供一种简单的方案：

订单处理{ if(库存不足){ return 库存不足; } if(余额不足){ return 余额不足; } 事务管理(rollbackFor = Exception.class){ //扣库存 int changeLine = 执行语句(update 商品表 set 库存=库存-购买数量 where 库存>购买数量 and 商品 ID = ?); if(changeLine != 1){ return 库存不足; } double 扣款金额= 商品价格 x 购买数量; //扣款 changeLine = 执行语句(update 用户余额表 set 余额=余额-扣款金额 where 余额 > 扣款金额 and 扣款金额 > 0 and 用户 ID = ?); if(changeLine != 1){ throw CustomRuntimeException("余额不足"); } //生成订单 changeLine = 执行语句( insert into 订单表 set ......); if(changeLine < 1){ throw CustomRuntimeException("订单生成失败"); } } } 

我们仔细来分析一下如上的整个逻辑

1、当一个业务进入逻辑体，先检查余额和库存，不满足条件则返回错误（可阻挡非并发情况下的大部分业务流入事物）

2、进入事物后，先扣取库存，当扣取失败，直接返回错误

3、扣取库存后，则进行扣款，当扣款失败，则抛出异常(由于在业务体走到这里，已经扣取了库存，本处不能 return，需抛出异常，让事物回滚)

4、扣款成功后，则生成订单，当订单生成失败，则抛出异常(理由同第三点)

特别注意：语句中，通过 where 来进行余额不足和库存不足的条件判断。通过执行语句返回的影响行数，来判断是否扣取成功。 在以上流程中，我们发现，即便不使用分布式锁，也无并发问题。

===========================================================

分布式锁

友情提示：切勿觉得笔者以上理论是拆自己的台，笔者作为互联网技术人，希望各位技术人能够将产品质量做到最好，少加班，多回家陪陪家人

ELock 介绍

 ELock 是笔者闲暇之余写的一套分布式锁插件，代码非常精简、并且以非轮询阻塞的方式进行加锁控制。适用于面向用户的互联网产品，目前用在一套用户量为 7 位数的直播系统中。 源码地址： https://gitee.com/coodyer/Coody-Framework/tree/original/coody-elock 

Maven 引用代码(可关注更新情况)：

<dependency> <groupId>org.coody.framework</groupId> <artifactId>coody-elock</artifactId> <!--更新于 2019-01-18 12:23:00 --> <version>alpha-1.2.2</version> </dependency> 

初始化 JedisPool

//直接传入连接池初始化(注：无密码请传 null) ELockCache.initJedisPool(JediPool); //传入 ip、端口、密码、超时时间初始化 ELockCache.initJedisPool(host, port, secretKey, timeOut); //传入 ip、端口、密码、超时时间、配置器初始化 ELockCache.initJedisPool(host, port, secretKey, timeOut, jedisPoolConfig); 

加锁

ELocker.lock(key, expireSecond); 

释放锁

ELocker.unLock(key); 

注意： 加锁代码(ELocker.lock(key, expireSecond);)。需 try{}catch{}包围，并在 finally 释放锁(ELocker.unLock(key);)

try { ELocker.lock(key, 100); for (int i = 0; i < 10; i++) { System.out.println(Thread.currentThread().getId() + ">>" + i); Thread.sleep(100l); } } catch (InterruptedException e) { e.printStackTrace(); } finally { ELocker.unLock(key); } 

测试代码

import java.util.ArrayList; import java.util.List; import org.coody.framework.elock.ELocker; import org.coody.framework.elock.redis.ELockCache; /** * 分布式锁测试 * [@author]( https://my.oschina.net/arthor) Coody * * 2018 年 12 月 14 日 * * [@blog]( https://my.oschina.net/wangboxi) 54sb.org */ public class ELockTest { //要加锁的 key static String key = "TESTLOCK_1"; static { //初始化 jedis 连接 ELockCache.initJedisPool("127.0.0.1", 16379, "123456", 10000); } public static void main(String[] args) { List<Thread> threads = new ArrayList<Thread>(); for (int i = 0; i < 10; i++) { Thread thread = new Thread(new Runnable() { [@Override]( https://my.oschina.net/u/1162528) public void run() { test(); } }); threads.add(thread); } //启动十个线程 for (Thread thread : threads) { thread.start(); } } //要锁的方法 private static void test() { try { ELocker.lock(key, 100); for (int i = 0; i < 10; i++) { System.out.println(Thread.currentThread().getId() + ">>" + i); Thread.sleep(100l); } } catch (InterruptedException e) { e.printStackTrace(); } finally { ELocker.unLock(key); } } } 

执行效果：

===========================================================

这套锁在 Spring 下的花样玩法

1、配置分布式锁中使用的缓存

 <bean id="eLockCache" class="org.coody.framework.elock.redis.ELockCache" lazy-init="false"> <property name="jedisPool" ref="jedisPool" /> </bean> 

2、配置分布式锁切面

<!-- 配置切面的 bean --> <bean id="eLockInterceptor" class="org.coody.framework.elock.aspect.ELockAspect"></bean> <!-- 配置 AOP --> <aop:config> <!-- 配置切面表达式 --> <aop:pointcut expression="@annotation(org.coody.framework.elock.annotation.ELock)" id="eLockPointcut" /> <!-- 配置切面和通知 order：越小优先级越高 --> <aop:aspect id="logAspect" ref="eLockInterceptor"> <aop:around method="rdLockForAspectj" pointcut-ref="eLockPointcut" /> </aop:aspect> </aop:config> 

致此，分布式锁配置完成，开始进入我们的花样玩法。

NO1. 使用注解添加分布式锁：

 @ELock(name = "USER_MODIFY_LOCK", fields = "userId", waitTime = 20) public void delUser(String userId) { userDao.delUser(userId); } 

在 ELock 注解中，name 代表 key 名字，field 代表拼接的字段。

当 fields 所有字段长度超过 32 时，elock 将会对 key 进行 md5 获取摘要作为缓存的 key，即 name:key。

本处 fields 支持选择对象的字段，即：方法参数名.字段值（如：userInfo.userId ）

本处 fields 支持多个字段，fields={"userInfo.userId","orderInfo.orderId"}

当不指定 key 时，elock 将会根据包名、类名、方法名和方法参数生成 key

当不指定 fields 时，elock 不会拼接任何多余参数，则该方法变成全局同步方法

如图：

NO2. 使用锁执行器添加分布式锁

public void delUser(String userId) throws InterruptedException { String key="USER_MODIFY_LOCK"+userId; Integer code=new AbstractLockAble(key,20) { @Override public Object doService() { return userDao.delUser(userId); } }.invoke(); } 

通过 返回值=new AbstractLockAble(锁名称，超时时间){}.invoke()的方式，覆盖 doService 方法，将需要加锁的代码块放置 doService 方法里面执行。

如图：

===========================================================

如果在项目中觉得两种方式不可取，可采用上文中常规方式。

笔者曾经百度搜索 Java 分布式锁实现，发现所提供方案都如出一辙(由于没有作图工具，就随便写下流程)。

1、尝试获得锁

2、死循环轮询获得锁

3、执行业务

4、释放锁

在网上查到的方案，相信很多小朋友都知道，不知道是谁通过这种方式来做分布式锁，然后被一大堆网友转载。

这种方案是可以实现锁，但是不适用于对外的互联网产品。

重大问题地雷：当多个线程尝试获得锁，只有一个线程会执行，剩下的线程都在轮询获得锁。这里我们假设时间精度为 1ms，那就意味着每个线程每秒钟最多轮询 1000 次。然而在分布式锁中，我们需要借助中介容器去进行尝试获得锁的操作，如 redis zookeeper。故此，我们假设这个 key 有 100 个线程，第一个线程执行卡住，那么，1 个线程在执行业务，99 个线程在以每秒钟 1000 的频次对中间容器发起 ddos 攻击。故此，如上方案不适用于对外的互联网产品。

笔者的方案：

1、尝试获得锁

2、线程入列并暂停

3、执行业务

4、发送消息释放锁，并唤醒下一个线程(轮询至第 1 步)

我们知道，redis 也好，zookeeper 也好，都有消息订阅机制。当业务流入的时候，获取锁失败的线程，都进入了挂起的状态，那么此时有一个线程在执行。当这个线程执行完毕后，发送消息，这时候所有的应用程序都收到了这个消息，并尝试获得锁，以此往复，实现业务体执行权限

版权说明：

作者：Coody

版权：2014-2020 Test404 All right reserved. 版权所有

反馈邮箱： [email protected]