分布式锁的场景
秒杀场景案例
对于商品秒杀的场景,我们需要防止库存超卖或者重复扣款等并发问题,我们通常需要使用分布式锁,来解决共享资源竞争导致数据不一致的问题。
以手机秒杀的场景为例子,在抢购的过程中通常我们有三个步骤:
- 扣掉对应商品的库存;2. 创建商品的订单;3. 用户支付。
对于这样的场景我们就可以采用分布式锁的来解决,比如我们在用户进入秒杀 “下单“ 链接的过程中,我们可以对商品库存进行加锁,然后完成扣库存和其他操作,操作完成后。释放锁,让下一个用户继续进入保证库存的安全性;也可以减少因为秒杀失败,导致 DB 回滚的次数。整个流程如下图所示:
注:对于锁的粒度要根据具体的场景和需求来权衡。
三种分布式锁
对于 Zookeeper 的分布式锁实现,主要是利用 Zookeeper 的两个特征来实现:
- Zookeeper 的一个节点不能被重复创建
- Zookeeper 的 Watcher 监听机制
非公平锁
对于非公平锁,我们在加锁的过程如下图所示。
优点和缺点
其实上面的实现有优点也有缺点:
- 优点:
实现比较简单,有通知机制,能提供较快的响应,有点类似
ReentrantLock
的思想,对于节点删除失败的场景由Session
超时保证节点能够删除掉。 - 缺点: 重量级,同时在大量锁的情况下会有 “惊群” 的问题。
“惊群” 就是在一个节点删除的时候,大量对这个节点的删除动作有订阅 Watcher
的线程会进行回调,这对Zk集群是十分不利的。所以需要避免这种现象的发生。
解决“惊群”:
为了解决“惊群“问题,我们需要放弃订阅一个节点的策略,那么怎么做呢?
- 我们将锁抽象成目录,多个线程在此目录下创建瞬时的顺序节点,因为 Zookeeper 会为我们保证节点的顺序性,所以可以利用节点的顺序进行锁的判断。
- 首先创建顺序节点,然后获取当前目录下最小的节点,判断最小节点是不是当前节点,如果是那么获取锁成功,如果不是那么获取锁失败。
- 获取锁失败的节点获取当前节点上一个顺序节点,对此节点注册监听,当节点删除的时候通知当前节点。
- 当unlock的时候删除节点之后会通知下一个节点。
公平锁
基于非公平锁的缺点,我们可以通过一下的方案来规避。
优点和缺点
优点: 如上借助于临时顺序节点,可以避免同时多个节点的并发竞争锁,缓解了服务端压力。
缺点: 对于读写场景来说,无法解决一致性的问题,如果读的时候也去获取锁的话,这样会导致性能下降,对于这样的问题,我们可以通过读写锁来实现如类似 jdk 中的 ReadWriteLock
读写锁实现
对于读写锁的特点:读写锁在如果多个线程都是在读的时候,是可以并发读的,就是一个无锁的状态,如果有写锁正在操作的时候,那么读锁需要等待写锁。在加写锁的时候,由于前面的读锁都是并发,所以需要监听最后一个读锁完成后执行写锁。步骤如下:
- read 请求, 如果前面是读锁,可以直接读取,不需要监听。如果前面是一个或者多个写锁那么只需要监听最后一个写锁。
- write 请求,只需要对前面的节点监听。
Watcher
机制和互斥锁一样。
分布式锁实战
本文源码中使用环境:JDK 1.8
、Zookeeper 3.6.x
Curator 组件实现
POM 依赖
代码语言:javascript复制<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-framework</artifactId>
<version>2.13.0</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>2.13.0</version>
</dependency>
互斥锁运用
由于 Zookeeper 非公平锁的 “惊群” 效应,非公平锁在 Zookeeper 中其实并不是最好的选择。下面是一个模拟秒杀的例子来使用 Zookeeper 分布式锁。
代码语言:javascript复制
代码语言:javascript复制public class MutexTest {
static ExecutorService executor = Executors.newFixedThreadPool(8);
static AtomicInteger stock = new AtomicInteger(3);
public static void main(String[] args) throws InterruptedException {
CuratorFramework client = getZkClient();
String key = "/lock/lockId_111/111";
final InterProcessMutex mutex = new InterProcessMutex(client, key);
for (int i = 0; i < 99; i ) {
executor.submit(() -> {
if (stock.get() < 0) {
System.err.println("库存不足, 直接返回");
return;
}
try {
boolean acquire = mutex.acquire(200, TimeUnit.MILLISECONDS);
if (acquire) {
int s = stock.decrementAndGet();
if (s < 0) {
System.err.println("进入秒杀,库存不足");
} else {
System.out.println("购买成功, 剩余库存: " s);
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (mutex.isAcquiredInThisProcess())
mutex.release();
} catch (Exception e) {
e.printStackTrace();
}
}
});
}
while (true) {
if (executor.isTerminated()) {
executor.shutdown();
System.out.println("秒杀完毕剩余库存为:" stock.get());
}
TimeUnit.MILLISECONDS.sleep(100);
}
}
private static CuratorFramework getZkClient() {
String zkServerAddress = "127.0.0.1:2181";
ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3, 5000);
CuratorFramework zkClient = CuratorFrameworkFactory.builder()
.connectString(zkServerAddress)
.sessionTimeoutMs(5000)
.connectionTimeoutMs(5000)
.retryPolicy(retryPolicy)
.build();
zkClient.start();
return zkClient;
}
}
代码语言:javascript复制
读写锁运用
读写锁可以用来保证缓存双写的强一致性的,因为读写锁在多线程读的时候是无锁的, 只有在前面有写锁的时候才会等待写锁完成后访问数据。
代码语言:javascript复制
代码语言:javascript复制public class ReadWriteLockTest {
static ExecutorService executor = Executors.newFixedThreadPool(8);
static AtomicInteger stock = new AtomicInteger(3);
static InterProcessMutex readLock;
static InterProcessMutex writeLock;
public static void main(String[] args) throws InterruptedException {
CuratorFramework client = getZkClient();
String key = "/lock/lockId_111/1111";
InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(client, key);
readLock = readWriteLock.readLock();
writeLock = readWriteLock.writeLock();
for (int i = 0; i < 16; i ) {
executor.submit(() -> {
try {
boolean read = readLock.acquire(2000, TimeUnit.MILLISECONDS);
if (read) {
int num = stock.get();
System.out.println("读取库存,当前库存为: " num);
if (num < 0) {
System.err.println("库存不足, 直接返回");
return;
}
}
} catch (Exception e) {
e.printStackTrace();
}finally {
if (readLock.isAcquiredInThisProcess()) {
try {
readLock.release();
} catch (Exception e) {
e.printStackTrace();
}
}
}
try {
boolean acquire = writeLock.acquire(2000, TimeUnit.MILLISECONDS);
if (acquire) {
int s = stock.get();
if (s <= 0) {
System.err.println("进入秒杀,库存不足");
} else {
s = stock.decrementAndGet();
System.out.println("购买成功, 剩余库存: " s);
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (writeLock.isAcquiredInThisProcess())
writeLock.release();
} catch (Exception e) {
e.printStackTrace();
}
}
});
}
while (true) {
if (executor.isTerminated()) {
executor.shutdown();
System.out.println("秒杀完毕剩余库存为:" stock.get());
}
TimeUnit.MILLISECONDS.sleep(100);
}
}
private static CuratorFramework getZkClient() {
String zkServerAddress = "127.0.0.1:2181";
ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3, 5000);
CuratorFramework zkClient = CuratorFrameworkFactory.builder()
.connectString(zkServerAddress)
.sessionTimeoutMs(5000)
.connectionTimeoutMs(5000)
.retryPolicy(retryPolicy)
.build();
zkClient.start();
return zkClient;
}
}
代码语言:javascript复制
打印结果如下,一开始会有 8 个输出结果为 读取库存,当前库存为: 3
然后在写锁中回去顺序的扣减少库存。
代码语言:javascript复制读取库存,当前库存为: 3
读取库存,当前库存为: 3
读取库存,当前库存为: 3
读取库存,当前库存为: 3
读取库存,当前库存为: 3
读取库存,当前库存为: 3
读取库存,当前库存为: 3
读取库存,当前库存为: 3
购买成功, 剩余库存: 2
购买成功, 剩余库存: 1
购买成功, 剩余库存: 0
进入秒杀,库存不足
进入秒杀,库存不足
进入秒杀,库存不足
进入秒杀,库存不足
进入秒杀,库存不足
读取库存,当前库存为: 0
读取库存,当前库存为: 0
读取库存,当前库存为: 0
读取库存,当前库存为: 0
读取库存,当前库存为: 0
读取库存,当前库存为: 0
读取库存,当前库存为: 0
读取库存,当前库存为: 0
进入秒杀,库存不足
进入秒杀,库存不足
进入秒杀,库存不足
进入秒杀,库存不足
进入秒杀,库存不足
进入秒杀,库存不足
进入秒杀,库存不足
进入秒杀,库存不足
代码语言:javascript复制
分布式锁的选择
咱们最常用的就是 Redis 的分布式锁和 Zookeeper 的分布式锁,在性能方面 Redis 的每秒钟 TPS 可以上轻松上万。在大规模的高并发场景我推荐使用 Redis 分布式锁来作为推荐的技术方案。如果对并发要求不是特别高的场景可以使用 Zookeeper 分布式来处理。
参考资料
- https://www.cnblogs.com/leeego-123/p/12162220.html
- http://curator.apache.org/
- https://blog.csdn.net/hosaos/article/details/89521537