十curator recipes之信号量InterProcessSemaphoreV2

Posted 2021-02-03 lay2017

简介

跟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();
    }
}