十、curator recipes之信号量InterProcessSemaphoreV2

简介

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

    }

}