分布式Semaphore

  1. semaphore的定义,意义
  2. 在没有juc semaphore之前怎么实现
  3. semaphore使用
  4. 分布式semaphore实现

信号量

最早用来解决进程同步与互斥问题的机制:
包括一个称为信号量的变量及对它进行的两个原语操作(PV操作)

什么是信号量?

信号量(semaphore)的数据结构为一个值和一个指针,指针指向等待该信号量的下一个进程。信号量的值与相应资源的使用情况有关。

PV操作由P操作原语和V操作原语组成(原语是不可中断的过程)

(注,P是荷兰语的Passeren,相当于英文的pass,V是荷兰语的Verhoog,相当于英文中的incremnet)

对信号量进行操作,具体定义如下:

  • P(S):
    • ①将信号量S的值减1,即S=S-1;
    • ②如果S>=0,则该进程继续执行;否则该进程置为等待状态,排入等待队列
  • V(S):
    • ①将信号量S的值加1,即S=S+1;
    • ②如果S>0,则该进程继续执行;否则释放队列中第一个等待信号量的进程

PV操作的意义:我们用信号量及PV操作来实现进程的同步和互斥。PV操作属于进程的低级通信

使用PV操作实现进程互斥时应该注意的是:

  1. 每个程序中用户实现互斥的P、V操作必须成对出现,先做P操作,进临界区,后做V操作,出临界区。若有多个分支,要认真检查其成对性
  2. P、V操作应分别紧靠临界区的头尾部,临界区的代码应尽可能短,不能有死循环
  3. 互斥信号量的初值一般为1
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
//许可数量
private int permits = 1;

public synchronized void P() {
permits--;
if(permits < 0 ){
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

public synchronized void V(){
permits++;
if(permits <=0){
notifyAll();
}
}

J.U.C Semaphore

JUC提供了工具类之一就是Semaphore,提供了丰富的API,不再需要自己实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
// 创建具有给定的许可数和非公平的公平设置的 Semaphore。
Semaphore(int permits)
// 创建具有给定的许可数和给定的公平设置的 Semaphore。
Semaphore(int permits, boolean fair)

// 从此信号量获取一个许可,在提供一个许可前一直将线程阻塞,否则线程被中断。
void acquire()
// 从此信号量获取给定数目的许可,在提供这些许可前一直将线程阻塞,或者线程已被中断。
void acquire(int permits)
// 从此信号量中获取许可,在有可用的许可前将其阻塞。
void acquireUninterruptibly()
// 从此信号量获取给定数目的许可,在提供这些许可前一直将线程阻塞。
void acquireUninterruptibly(int permits)
// 返回此信号量中当前可用的许可数。
int availablePermits()
// 获取并返回立即可用的所有许可。
int drainPermits()
// 返回一个 collection,包含可能等待获取的线程。
protected Collection<Thread> getQueuedThreads()
// 返回正在等待获取的线程的估计数目。
int getQueueLength()
// 查询是否有线程正在等待获取。
boolean hasQueuedThreads()
// 如果此信号量的公平设置为 true,则返回 true。
boolean isFair()
// 根据指定的缩减量减小可用许可的数目。
protected void reducePermits(int reduction)
// 释放一个许可,将其返回给信号量。
void release()
// 释放给定数目的许可,将其返回到信号量。
void release(int permits)
// 返回标识此信号量的字符串,以及信号量的状态。
String toString()
// 仅在调用时此信号量存在一个可用许可,才从信号量获取许可。
boolean tryAcquire()
// 仅在调用时此信号量中有给定数目的许可时,才从此信号量中获取这些许可。
boolean tryAcquire(int permits)
// 如果在给定的等待时间内此信号量有可用的所有许可,并且当前线程未被中断,则从此信号量获取给定数目的许可。
boolean tryAcquire(int permits, long timeout, TimeUnit unit)
// 如果在给定的等待时间内,此信号量有可用的许可并且当前线程未被中断,则从此信号量获取一个许可。
boolean tryAcquire(long timeout, TimeUnit unit)

对于JUC的Semaphore源码,此篇不阐述了,另开新篇;但对分布式的Semaphore倒是可以研究下

分布式Semaphore

Redission中有对应的RSemaphore

1
2
3
4
5
6
RSemaphore semaphore = redisson.getSemaphore("semaphore");
semaphore.acquire();
//或
semaphore.acquireAsync();
semaphore.acquire(23);
semaphore.tryAcquire();

可过期信号量

1
2
3
4
5
6
RPermitExpirableSemaphore semaphore = redisson.getPermitExpirableSemaphore("mySemaphore");
String permitId = semaphore.acquire();
// 获取一个信号,有效期只有2秒钟。
String permitId = semaphore.acquire(2, TimeUnit.SECONDS);
// ...
semaphore.release(permitId);

直接上最本质的源码片段,lua脚本很简单,对信号量进行计数,acquire时,信号量减1,release时,信号量加1;主要是保证操作的原子性

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
@Override
public RFuture<Boolean> tryAcquireAsync(int permits) {
if (permits < 0) {
throw new IllegalArgumentException("Permits amount can't be negative");
}
if (permits == 0) {
return RedissonPromise.newSucceededFuture(true);
}

return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"local value = redis.call('get', KEYS[1]); " +
"if (value ~= false and tonumber(value) >= tonumber(ARGV[1])) then " +
"local val = redis.call('decrby', KEYS[1], ARGV[1]); " +
"return 1; " +
"end; " +
"return 0;",
Collections.<Object>singletonList(getName()), permits);
}

@Override
public RFuture<Void> releaseAsync(int permits) {
if (permits < 0) {
throw new IllegalArgumentException("Permits amount can't be negative");
}
if (permits == 0) {
return RedissonPromise.newSucceededFuture(null);
}

return commandExecutor.evalWriteAsync(getName(), StringCodec.INSTANCE, RedisCommands.EVAL_VOID,
"local value = redis.call('incrby', KEYS[1], ARGV[1]); " +
"redis.call('publish', KEYS[2], value); ",
Arrays.<Object>asList(getName(), getChannelName()), permits);
}

在最本质的基础上,再深入看一下还做了哪些事,能真正达到一个工业生产标准

tryAcquire()

非阻塞式,有信息量就正常获取,没有刚快速返回,就是lua本质,没有做额外的事情

acquire()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
@Override
public void acquire(int permits) throws InterruptedException {
if (tryAcquire(permits)) {
return;
}
RFuture<RedissonLockEntry> future = subscribe();
commandExecutor.syncSubscription(future);
try {
while (true) {
if (tryAcquire(permits)) {
return;
}
getEntry().getLatch().acquire(permits);
}
} finally {
unsubscribe(future);
}
}

阻塞式,相对非阻塞式就多了一些事

  • 1.先tryAcquire,看是否能获取到信号量
  • 2.订阅channel事件
  • 3.无限循环
    • 3.1.先tryAcquire(),尝试一下
    • 3.2.通过getEntry().getLatch(),也就是j.u.c.Semaphore,acquire()阻塞
  • 4.取消订阅

订阅事件内部细节,另开篇再说了,他的目的其实就是释放Semaphore

想像一下,同一个client的两个线程A,B 同时需要获取信号量,如果A成功获取,那么B将被Semaphore阻塞住了,何时退出阻塞呢?

就在线程A进行release()之后,会publish,细节可查看上面的release()中的lua脚本,当B监听到事件时,就会调用Semaphore.release(),再次进行tryAcquire()

tryAcquire(int permits, long waitTime, TimeUnit unit)

如果在给定的等待时间内此信号量有可用的所有许可,并且当前线程未被中断,则从此信号量获取给定数目的许可

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@Override
public boolean tryAcquire(int permits, long waitTime, TimeUnit unit) throws InterruptedException {
long time = unit.toMillis(waitTime);
long current = System.currentTimeMillis();

if (tryAcquire(permits)) {
return true;
}

time -= (System.currentTimeMillis() - current);
if (time <= 0) {
return false;
}

current = System.currentTimeMillis();
RFuture<RedissonLockEntry> future = subscribe();
if (!await(future, time, TimeUnit.MILLISECONDS)) {
return false;
}

try {
time -= (System.currentTimeMillis() - current);
if (time <= 0) {
return false;
}

while (true) {
current = System.currentTimeMillis();
if (tryAcquire(permits)) {
return true;
}

time -= (System.currentTimeMillis() - current);
if (time <= 0) {
return false;
}

// waiting for message
current = System.currentTimeMillis();

getEntry().getLatch().tryAcquire(permits, time, TimeUnit.MILLISECONDS);

time -= (System.currentTimeMillis() - current);
if (time <= 0) {
return false;
}
}
} finally {
unsubscribe(future);
}
// return get(tryAcquireAsync(permits, waitTime, unit));
}

其实await(future, time, TimeUnit.MILLISECONDS)是使用的CountDownLatch

如果计数到达零,则返回 true;如果在计数到达零之前超过了等待时间,则返回 false

当前是第一个请求,或者别的释放,那就再往下进入循环

CountDownLatch.await()+Semaphore.tryAcquire()配合使用

每一次等待时间后,都需要检查是否超过等待时间

为什么需要引入CountDownLatch.await()呢? 都使用Semaphore.tryAcquire()不行吗?这个需要再次深入挖掘了

总结

分布式信号量,原理很明了,主要还是通过lua保障redis操作的原子性

阅读redisson源码,发现里面的操作基本都是异步化,底层又是基于netty,大量使用了future模式,如果不知道future模式,会很绕,debug都会晕掉,所以在深入redisson之前,需要再对future模式温习一下

朱兴生 wechat
最新文章尽在微信公众号『码农戏码』