多线程-CountDownLatch,CyclicBarrier,Semaphore,Exchanger,Phaser

Posted 小路不懂2

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了多线程-CountDownLatch,CyclicBarrier,Semaphore,Exchanger,Phaser相关的知识,希望对你有一定的参考价值。

CountDownLatch 
一个同步辅助类,在完成一组正在其他线程中执行的操作之前,它允许一个或多个线程一直等待。用给定的计数初始化CountDownLatch。调用countDown()计数减一,当计数到达零之前await()方法会一直阻塞,计数无法被重置。

public class CountDownLatch {
    private final Sync sync;
    public CountDownLatch(int count);
    public void countDown() {
        sync.releaseShared(1);
    }
    public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }
    public boolean await(long timeout, TimeUnit unit) throws InterruptedException {
        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    }
}

CountDownLatch中主要有countDown()和await()方法。 
countDown()递减计数,如果计数达到零,则是否所有等待的线程。 
1. 如果当前计数大于零,则计数减一; 
2. 如果减一之后计数为零,则重新调度所有等待该计数为零的线程; 
3. 如果计数已经为零,则不发生任何操作; 
await()使当前线程在计数为零之前一直阻塞,除非线程被中断或超出指定的等待时间; 
如果计数为零,则立刻返回true 
在进入此方法时,当前线程已经设置了中断状态或在等待时被中断,则抛出InterruptedException异常,并且清除当前线程的中断状态。如果超出了指定等待时间,则返回false,如果该时间小于等于零,则此方法根本不会等待。

package org.github.lujiango;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class Test16 {

    public static void main(String[] args) throws InterruptedException {
        final CountDownLatch begin = new CountDownLatch(1);
        final CountDownLatch end = new CountDownLatch(10);
        final ExecutorService exec = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 10; i++) {
            final int no = i + 1;
            Runnable run = new Runnable() {
                @Override
                public void run() {
                    try {
                        begin.await();
                        TimeUnit.MILLISECONDS.sleep((long) (Math.random() * 10000));
                        System.out.println("No." + no + " arrived");
                    } catch (Exception e) {

                    } finally {
                        end.countDown();
                    }
                }
            };
            exec.submit(run);
        }

        System.out.println("Game start");
        begin.countDown();
        end.await();
        System.out.println("Game over");
        exec.shutdown();
    }

}

CyclicBarrier 
一个同步辅助类,它允许一组线程互相等待,直到到达某个公共屏障点。在涉及一组固定大小的线程的程序中,这些线程必须不时的互相等待。

package org.github.lujiango;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class Test16 {

    public static void main(String[] args) throws InterruptedException, BrokenBarrierException {
        final CyclicBarrier end = new CyclicBarrier(10);
        final ExecutorService exec = Executors.newFixedThreadPool(10);
        System.out.println("Game start");
        for (int i = 0; i < 10; i++) {
            final int no = i + 1;
            Runnable run = new Runnable() {
                @Override
                public void run() {
                    try {
                        end.await();
                        TimeUnit.MILLISECONDS.sleep((long) (Math.random() * 10000));
                        System.out.println("No." + no + " arrived");
                    } catch (Exception e) {

                    } finally {
                    }
                }
            };
            exec.submit(run);
        }
        System.out.println("Game over");
        exec.shutdown();

    }

}

需要所有的子任务都完成时,才执行主任务,这个时候可以选择使用CyclicBarrier。

Semaphore 
一个计数信号量,信号量维护了一个许可集,在许可可用之前会阻塞每一个acquire(),然后获取该许可。每个release()释放许可,从而可能释放一个正在阻塞的获取者。 
Semaphore只对可用许可的号码进行计数,并采取相应的行动,拿到信号的线程可以进入代码,否则就等待。

package org.github.lujiango;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

public class Test17 {

    public static void main(String[] args) {
        ExecutorService exec = Executors.newCachedThreadPool();
        final Semaphore semp = new Semaphore(5);
        for (int i = 0; i < 20; i++) {
            final int no = i;
            Runnable run = new Runnable() {

                @Override
                public void run() {
                    try {
                        semp.acquire();
                        System.out.println("Accessing: " + no);
                        TimeUnit.MILLISECONDS.sleep((long) (Math.random() * 10000));
                    } catch (Exception e) {

                    } finally {
                        semp.release();
                    }
                }
            };
            exec.submit(run);
        }
        exec.shutdown();
    }

}

Exchanger 
Exchanger可以在两个线程之间交换数据,只能在两个线程,不支持更多的线程之间互换数据。 
当线程A调用Exchanger对象的exchage()方法后,会阻塞;直到B线程也调用exchange()方法,然后线程以安全的方式交换数据,之后A和B线程继续执行。

package org.github.lujiango;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Exchanger;

public class Test18 {

    public static void main(String[] args) {
        Exchanger<List<Integer>> ex = new Exchanger<List<Integer>>();
        new A(ex).start();
        new B(ex).start();
    }

}

class A extends Thread {
    List<Integer> list = new ArrayList<Integer>();
    Exchanger<List<Integer>> ex;

    public A(Exchanger<List<Integer>> ex) {
        this.ex = ex;
    }

    @Override
    public void run() {
        Random random = new Random();
        for (int i = 0; i < 10; i++) {
            list.clear();
            list.add(random.nextInt(10));
            list.add(random.nextInt(10));
            list.add(random.nextInt(10));
            try {
                list = ex.exchange(list);
            } catch (Exception e) {

            }
        }
    }
}

class B extends Thread {
    List<Integer> list = new ArrayList<Integer>();
    Exchanger<List<Integer>> ex;

    public B(Exchanger<List<Integer>> ex) {
        this.ex = ex;
    }

    @Override
    public void run() {
        for (int i = 0; i < 10; i++) {
            try {
                list = ex.exchange(list);
            } catch (Exception e) {

            }
            System.out.println(list);
        }
    }
}

Phaser 
Phaser是一个灵活的线程同步工具,它包含了CountDownLatch和CyclicBarrier的相关功能。 
CountDownLatch的countDown()和await()可以通过Phaser的arrive()和awaitAdvance(int n)代替 
而CyclicBarrier的await可以使用Phaser的arriveAndAwaitAdvance()方法代替 
用Phaser代替CountDownLatch:

package org.github.lujiango;

import java.util.concurrent.Phaser;
import java.util.concurrent.TimeUnit;

public class Test19 {

    public static void main(String[] args) throws InterruptedException {
        final Phaser latch = new Phaser(10);
        for (int i = 1; i <= 10; i++) {
            final int id = i;
            Thread t = new Thread(new Runnable() {

                @Override
                public void run() {
                    try {
                        TimeUnit.SECONDS.sleep((long) (Math.random() * 10));
                        System.out.println("thread: " + id + " is running");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        latch.arrive();
                    }
                }
            });
            t.start();
        }
        latch.awaitAdvance(latch.getPhase());
        System.out.println("all thread has run");

    }

}
package org.github.lujiango;

import java.util.concurrent.Phaser;
import java.util.concurrent.TimeUnit;

public class Test19 {

    public static void main(String[] args) throws InterruptedException {
        final Phaser latch = new Phaser(10);
        for (int i = 1; i <= 10; i++) {
            final int id = i;
            Thread t = new Thread(new Runnable() {

                @Override
                public void run() {
                    try {
                        TimeUnit.SECONDS.sleep((long) (Math.random() * 10));
                        latch.arriveAndAwaitAdvance(); // 所有线程都执行到这里,才会继续执行,否则全部阻塞
                        System.out.println("thread: " + id + " is running");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        latch.arrive();
                    }
                }
            });
            t.start();
        }
    }

}

  

 

以上是关于多线程-CountDownLatch,CyclicBarrier,Semaphore,Exchanger,Phaser的主要内容,如果未能解决你的问题,请参考以下文章

java多线程-CountDownLatch

Java并发多线程编程——CountDownLatch

多线程(七同步计数器-CountDownLatch)

多线程(十二AQS原理-CountDownLatch基于AQS的共享实现)

java多线程-CountDownLatch

JAVA多线程之CountDownLatch