Barrier of cursor application scenario

curator A distributed barrier is implemented for distributed scenarios ：barrier. We can use... In distributed systems barrier To block the process , Know that a condition is triggered . In fact, with Java Multithreading barrier It's the same .

for example ： When two processes are performing tasks ,A Called B,A Need to wait B Notification after completion .

Distributed Barrier It's such a class ： It blocks waiting processes on all nodes , Know that one is satisfied , And then all the nodes go on .
For example, in horse racing , When the horses come to the starting line . ex , All the horses were racing out .

One 、 fence Barrier

1、DistributedBarrier Class description

DistributedBarrier Class implements the function of a fence . Its constructor is as follows ：

/**
 * @param client client
 * @param barrierPath path to use as the barrier
 */
public DistributedBarrier(CuratorFramework client, String barrierPath)

DistributedBarrier In the constructor barrierPath The parameter is used to determine a fence , as long as barrierPath Parameters are the same ( The path is the same ) It's the same fence . In general, the use of fences is as follows ：

 1. The dominant client Set up a fence 
    2. Other clients will call waitOnBarrier() Wait for the fence to be removed , The program handles thread blocking 
    3. The dominant client Remove the fence , Other client processors will continue to run at the same time .

DistributedBarrier The main methods of class are as follows ：

setBarrier() -  Set up fences 
waitOnBarrier() -  Wait for the fence to be removed 
removeBarrier() -  Remove the fence 

As follows ：
First of all, you need to set up a fence , It will block the threads waiting on it :

setBarrier();

Then the blocked thread needs to call “ Method waiting for release conditions :

public void waitOnBarrier()

When the conditions are met , Remove the fence , All waiting threads will continue to execute ：

removeBarrier();

exception handling ：DistributedBarrier Will monitor the connection status , When the connection is broken waitOnBarrier() The method throws an exception .

2、 Code example ：

public class CuratorBarrier2 {

    /** zookeeper Address  */
    static final String CONNECT_ADDR = "172.16.158.11:2181,"
            + "172.16.158.12:2181,"
            + "172.16.158.13:2181";
    /** session Timeout time  */
    static final int SESSION_OUTTIME = 5000;//ms

    static DistributedBarrier barrier = null;

    public static void main(String[] args) throws Exception {



        for(int i = 0; i < 5; i++){
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
                        CuratorFramework cf = CuratorFrameworkFactory.builder()
                                .connectString(CONNECT_ADDR)
                                .sessionTimeoutMs(SESSION_OUTTIME)
                                .retryPolicy(retryPolicy)
                                .build();
                        cf.start();
                        barrier = new DistributedBarrier(cf, "/super");
                        System.out.println(Thread.currentThread().getName() + " Set up barrier!");
                        barrier.setBarrier();	// Set up 
                        barrier.waitOnBarrier();	// wait for 
                        System.out.println("--------- Start the program ----------");
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            },"t" + i).start();
        }

        Thread.sleep(5000);
        barrier.removeBarrier();	// Release 


    }
}

This example creates DistributedBarrier object To set fences and remove fences .
We created 5 Threads , Here it is Barrier Wait on . Finally, all threads continue to execute after the fence is removed .
If you start without fences , All threads will not block .

Two 、 Double fences Double Barrier

 Double fences allow clients to synchronize at the beginning and end of the calculation . When enough processes are added to the double fence , The process begins to calculate , When the calculation is complete , Get out of the fence . Double Palisades are DistributedDoubleBarrier

1、 DistributedDoubleBarrier Class description

DistributedDoubleBarrier Class implements the function of double fence . Its constructor is as follows ：

// client - the client
// barrierPath - path to use
// memberQty - the number of members in the barrier
public DistributedDoubleBarrier(CuratorFramework client, String barrierPath, int memberQty)

memberQty Is the number of members , When enter Method is called , Member blocked , Until all members call enter. When leave Method is called , It also blocks the calling thread , Know that all members have called leave.
It's like the 100 meter race , Start the gun , All the athletes started running , When all the athletes run across the finish line , The game is just over .

 Be careful ： Parameters memberQty The value of is just a threshold , Instead of a limit . When the number of waiting fences is greater than or equal to this value, the fence will open ！

And the fence (DistributedBarrier) equally , Double fenced barrierPath Parameters are also used to determine whether the fence is the same , The use of double fences is as follows ：
1. Create double fences on the same path from multiple clients (DistributedDoubleBarrier), And then call enter() Method , Wait for the number of fences to reach memberQty You can enter the fence when you're in .
2. The number of fences reaches memberQty, Multiple clients stop blocking at the same time and continue to run , Until the execution of leave() Method , wait for memberQty A number of fences are blocked at the same time leave() In the method .
3.memberQty A number of fences are blocked at the same time leave() In the method , Multiple clients leave() Method to stop blocking , Continue operation .

2、DistributedDoubleBarrier The main methods of class are as follows ：

enter()、enter(long maxWait, TimeUnit unit) - Waiting to enter the fence at the same time
leave()、leave(long maxWait, TimeUnit unit) - Waiting to leave the fence at the same time
exception handling ：DistributedDoubleBarrier Will monitor the connection status , When the connection is broken enter() and leave The method throws an exception .

public class CuratorBarrier1 {

    /** zookeeper Address  */
    static final String CONNECT_ADDR = "172.16.158.11:2181,"
            + "172.16.158.12:2181,"
            + "172.16.158.13:2181";
    /** session Timeout time  */
    static final int SESSION_OUTTIME = 5000;//ms

    public static void main(String[] args) throws Exception {



        for(int i = 0; i < 5; i++){
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
                        CuratorFramework cf = CuratorFrameworkFactory.builder()
                                .connectString(CONNECT_ADDR)
                                .retryPolicy(retryPolicy)
                                .build();
                        cf.start();

                        DistributedDoubleBarrier barrier = new DistributedDoubleBarrier(cf, "/super", 5);
                        Thread.sleep(1000 * (new Random()).nextInt(3));
                        System.out.println(Thread.currentThread().getName() + " Ready to ");
                        barrier.enter();
                        System.out.println(" Start running at the same time ...");
                        Thread.sleep(1000 * (new Random()).nextInt(3));
                        System.out.println(Thread.currentThread().getName() + " Operation completed ");
                        barrier.leave();
                        System.out.println(" Quit running at the same time ...");


                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            },"t" + i).start();
        }
    }
}

Be careful ： The number of double fences created is ：(QTY + 2), The parameters for creating a double fence are ：new DistributedDoubleBarrier(client, PATH, QTY), When the number of waiting fences is greater than or equal to this value (QTY) The fence will open ！