十、curator recipes之信号量InterProcessSemaphoreV2

时间:2022-08-16 07:14:29

简介

跟Java并信号量没有什么不同,curator实现的信号量也是基于令牌桶算法,当一个线程要执行的时候就去桶里面获取令牌,如果有足够的令牌那么我就执行如果没有那么我就阻塞,当线程执行完毕也要将令牌放回桶里。

官方文档:http://curator.apache.org/curator-recipes/shared-semaphore.html

javaDoc:http://curator.apache.org/apidocs/org/apache/curator/framework/recipes/locks/InterProcessSemaphoreV2.html

代码示例

以下示例中,我们设置了信号量为1,如果其中一个线程取走了,那么下一个线程将阻塞直接信号量被返回到桶里面。

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2;
import org.apache.curator.framework.recipes.locks.Lease;
import org.apache.curator.retry.ExponentialBackoffRetry; public class InterProcessSemaphoreDemo {
private static CuratorFramework client = CuratorFrameworkFactory.newClient("localhost:2181", new ExponentialBackoffRetry(3000, 2));
private static String path = "/semaphore/0001";
static {
client.start();
} public static void main(String[] args) throws InterruptedException {
startThread0();
startThread1();
Thread.sleep(50000);
client.close();
} private static void startThread1() {
new Thread(() -> {
InterProcessSemaphoreV2 semaphoreV2 = new InterProcessSemaphoreV2(client, path, 1);
Lease lease = null;
try {
System.out.println("thread0 acquiring");
lease = semaphoreV2.acquire();
System.out.println("thread0 acquired and sleeping");
Thread.sleep(3000);
} catch (Exception e) {
e.printStackTrace();
} finally {
semaphoreV2.returnLease(lease);
System.out.println("thread0 return lease");
}
}).start();
} private static void startThread0() {
new Thread(() -> {
InterProcessSemaphoreV2 semaphoreV2 = new InterProcessSemaphoreV2(client, path, 1);
Lease lease = null;
try {
System.out.println("thread1 acquiring");
lease = semaphoreV2.acquire();
System.out.println("thread1 acquired and sleeping");
Thread.sleep(3000);
} catch (Exception e) {
e.printStackTrace();
} finally {
semaphoreV2.returnLease(lease);
System.out.println("thread1 return lease");
}
}).start();
}
}