java并发编程基础(五)-新类库JUC包的构件

2023年9月5日 281次阅读 来源: JUC

java并发编程基础(五)-类库JUC包的构件

本博文为学习《thinking in java》一书中第21章“并发”时的小笔记，并使用这些构件结合简单的小场景写一些样例代码，这是由于原书上的代码太长了

CountDownLatch

  1. 作用：同步一个或者多个任务，强制它们等待由其他任务执行的一组操作完成

  2. 使用：

    • 向CountdownLatch对象设置一个初始计数值，任何在该对象中调用await()的方法都将阻塞，直到这个计数值为0
    • 其他任务在结束其工作时，可以在该对象中调用countDown()来减小这个值
    • CountDownLatch被设计只会触发一次，计数值不可以被重置，而CyclicBarrier是可以重置计数器的版本
    • 调用countDown()方法不会产生阻塞，只有对await()的调用才会阻塞，直到计数值到达0
    • 典型用法：将一个程序分为n个互相独立的可解决任务，并且创建值为0的CountDownLatch，当每一个任务完成时，都会在这个锁存器上调用countDown()。等待问题被解决的任务在这个锁存器上调用await()，将自己拦住直至锁存器计数完毕。

  3. 样例代码，“三思而后行”

    public class ThinkTwiceDemo {

    static class Think implements Runnable{

        private CountDownLatch latch;

        public Think(CountDownLatch latch) {
            this.latch = latch;
        }

        @Override
        public void run() {
            System.out.println("thinking...");
            latch.countDown();
        }
    }

    static class Action implements Runnable {

        private CountDownLatch latch;

        public Action(CountDownLatch latch) {
            this.latch = latch;
        }

        @Override
        public void run() {
            try {
                latch.await();
                System.out.println("now I can do something");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        int count = 3; // 三思
        CountDownLatch latch = new CountDownLatch(count);
        Think think = new Think(latch);
        Action action = new Action(latch);
        ExecutorService service = new ThreadPoolExecutor(2, 4, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2));

        service.execute(action);
        TimeUnit.SECONDS.sleep(1); // 故意拖一段时间，说明Action确实被阻塞了
        for (int i = 0; i < count; i++) {
            service.execute(think);
        }

        service.shutdown();
    }
}

CyclicBarrier

  1. 使用场景：创建一组任务，它们并行执行工作，然后在进行下一个步骤之前等待，直至所有任务都完成。它使得所有的并行任务都将在栅栏处列队，因此可以一致地向前移动。

  2. CyclicBarrier.await()就有些想CountDownLatch的countDown()，计数器都会减去1，只是CyclicBarrier可以重复使用。

  3. 样例：模拟两个玩家的掷骰子游戏，一轮游戏中只有两个人都掷出骰子，才能进入下一轮游戏。

    public class ComputerDiceGame {

    /** * 掷骰子 */
    static class Dice implements Runnable {

        private CyclicBarrier barrier;

        private int count; // 线程独有

        private String player;

        private static Random random = new Random(); // 线程安全的

        public Dice(CyclicBarrier barrier, int count, String player) {
            this.barrier = barrier;
            this.count = count;
            this.player = player;
        }

        @Override
        public void run() {
            try {
                for (int i = 1; i <= count; i++) {
                    System.out.println("Round" + i + " : " + player + " makes " + (random.nextInt(6) + 1));
                    barrier.await();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
        int round = 5; // 5轮
        int nPlayers = 2;
        CyclicBarrier barrier = new CyclicBarrier(nPlayers); // 2个玩家
        Dice dice1 = new Dice(barrier, round, "player1");
        Dice dice2 = new Dice(barrier, round, "player2");

        ExecutorService service = new ThreadPoolExecutor(2, 2, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2));
        service.execute(dice1);
        service.execute(dice2);
        service.shutdown();
    }
}

// 运行结果
Round1 : player2 makes 5
Round1 : player1 makes 1
Round2 : player2 makes 3
Round2 : player1 makes 1
Round3 : player1 makes 4
Round3 : player2 makes 1
Round4 : player2 makes 4
Round4 : player1 makes 6
Round5 : player1 makes 1
Round5 : player2 makes 6

  4. 写的时候又再想了一下，如果现在需要加上一个裁判，判断一轮游戏中哪一个player赢了，应该怎么办呢？这个时候，需要用到之前我的一篇博文的Condition来解决问题，现在的游戏流程是这样：

    • 在一轮游戏中，两个玩家掷出骰子
    • 只有在两个玩家掷出了之后，裁判才能进行判断
    • 只有在裁判判断产生结果之后，才能够进入下一轮游戏

  5. 现在代码变成

    public class ComputerDiceGame {

    /** * 游戏相关信息，实际上这个类命名不太好，不改了。。 */
    static class GameProperties {

        public int round = 5; // 游戏轮数

        public int nPlayers = 2; // 游戏玩家个数

        public int counter = 0; // counter记录一轮游戏中当前掷出的骰子数量，读写需要同步

        public CyclicBarrier barrier = new CyclicBarrier(nPlayers + 1); // 加一是因为

        private Lock lock = new ReentrantLock();

        public Condition condition = lock.newCondition();

        public int[] result = new int[nPlayers]; // 记录一轮游戏的结果，由于不同线程读写的是不同位置，所以不需要同步

        public void judgeCheck() { // 裁判主动检查玩家是否都已经掷出骰子
            lock.lock();
            try {
                while (counter != nPlayers) { // 玩家还没全部都掷出骰子
                    condition.await();
                }
                counter = 0; // 重置
            } catch (InterruptedException ex) {

            } finally {
                lock.unlock();
            }
        }

        public void incrementAndCheck() { // 玩家掷出骰子，并检查自己是否最后一个掷出的，如果是需要唤醒裁判
            lock.lock();
            try {
                counter++;
                if (counter == nPlayers) { // 自己是本轮最后一个掷出骰子的人，需要叫醒裁判
                    condition.signal();
                }
            } finally {
                lock.unlock();
            }

        }
    }

    static class Judge implements Runnable{

        private GameProperties properties;

        public Judge(GameProperties properties) {
            this.properties = properties;
        }

        @Override
        public void run() {
            try {
                for (int r = 1; r <= properties.round; r++) {
                    properties.judgeCheck();

                    // 判别哪个赢了
                    int pos = 0;
                    for (int i = 1; i < properties.nPlayers; i++) {
                        if (properties.result[i] > properties.result[pos]) {
                            pos = i;
                        }
                    }
                    System.out.println("Player" + (pos+1) + " wins in round " + r);

                    properties.barrier.await();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    }
    
    static class Dice implements Runnable {

        private int id;

        private static Random random = new Random(); // 线程安全的

        private GameProperties properties;

        public Dice(int id, GameProperties properties) {
            this.id = id;
            this.properties = properties;
        }

        @Override
        public void run() {
            try {
                for (int r = 1; r <= properties.round; r++) {
                    int point = random.nextInt(6) + 1;
                    properties.result[id] = point;
                    System.out.println("Round" + r + " : player" + (id+1) + " makes " + point);

                    // 更新轮数
                    properties.incrementAndCheck();

                    properties.barrier.await();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
        GameProperties properties = new GameProperties();
        Dice dice1 = new Dice(0, properties);
        Dice dice2 = new Dice(1, properties);
        Judge judge = new Judge(properties);

        ExecutorService service = new ThreadPoolExecutor(3, 3, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2));
        service.execute(dice1);
        service.execute(dice2);
        service.execute(judge);
        service.shutdown();
    }
}

// 运行结果(点数一样player1赢)
Round1 : player1 makes 3
Round1 : player2 makes 3
Player1 wins in round 1
Round2 : player1 makes 3
Round2 : player2 makes 3
Player1 wins in round 2
Round3 : player1 makes 6
Round3 : player2 makes 6
Player1 wins in round 3
Round4 : player1 makes 5
Round4 : player2 makes 6
Player2 wins in round 4
Round5 : player1 makes 4
Round5 : player2 makes 6
Player2 wins in round 5

DelayQueue

  1. 无界的BlockingQueue，用于放置实现了Delayed接口的对象，其中的对象只能在其到期时才能从队列中取走

  2. 队列是有序的，即从队头对象的延迟到期时间最长，如果没有任何延迟到期，那么就不会有任何头元素，并且poll返回null(故不可以将null放入队列中)

  3. 要想构建延迟任务，需要实现Runnable和Delayed接口，Delayed接口的两个方法

    • compareTo：实现自己比较两个Delayed对象的方法，关系到从DelayedQueue中的优先级问题
    • getDelayed：得到Delayed对象现在的延迟时间，可以理解为还有多久可以干活了

  4. 样例代码

    public class SleeperDemo {

    static class Sleeper implements Runnable, Delayed {

        private long millisecond; // 拖延时间

        private long trigger; // 需要干活的时间

        private int id; // 标识

        public Sleeper(long millisecond, int id) {
            this.millisecond = millisecond;
            this.id = id;
            trigger = System.currentTimeMillis() + millisecond;
        }

        @Override
        public int compareTo(Delayed o) {
            Sleeper that = (Sleeper) o;
            if (trigger > that.trigger) {
                return 1;
            } else if (trigger < that.trigger) {
                return -1;
            } else {
                return 0;
            }
        }

        @Override
        public void run() {
            System.out.println("sleeper" + id + " is working now!");
        }

        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(trigger - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }
    }

    static class DelayedTaskConsumer implements Runnable {

        private DelayQueue<Sleeper> queue;

        public DelayedTaskConsumer(DelayQueue<Sleeper> queue) {
            this.queue = queue;
        }

        @Override
        public void run() {
            try {
                while (!Thread.interrupted()) {
                    queue.take().run(); // 直接在当前线程中运行
                }
            } catch (InterruptedException ex) {

            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        DelayQueue<Sleeper> queue = new DelayQueue<>();

        int nSleepers = 5;
        for (int i = nSleepers; i >= 1; i--) {
            queue.add(new Sleeper(i * 1000, i));
        }

        ExecutorService service = Executors.newFixedThreadPool(1);
        service.execute(new DelayedTaskConsumer(queue));

        // 保证任务都执行完就关闭
        TimeUnit.SECONDS.sleep(6);
        service.shutdownNow();
    }

PriorityBlockingQueue

  1. 带有阻塞操作的优先队列，例子就不举了

ScheduledThreadPoolExecutor

  1. 相比于线程池ThreadPoolExecutor，ScheduledThreadPoolExecutor可以进行任务的调度
    • schedule()：可以设置一段时间后调度任务
    • scheduleAtFixRate()：可以设置每隔特定的时间重复执行任务
  2. 也没有什么特殊，只是多了调度，例子也不举了

Semaphore

  1. 正常的锁(concurrents.locks或synchronized)在任何时候都只允许一个任务访问一个资源

  2. 计数信号量semaphore允许n个任务同时访问这个资源

  3. 创建Sempaphore时，需要指定有多少个信号量，也可以指定信号量的分发是否公平(按照任务FIFO的请求顺序)，例如一个构造函数

    • public Semaphore(int permits, boolean fair)

  4. 信号量操作

    • acquire()：信号量-1，请求资源前需要调用该方法
    • release()：信号量+1，归还资源后需要调用该方法

  5. 明显，当信号量为1时，Semaphore的作用就和普通的Lock作用类似

  6. 样例代码：书本代码，一个对象池，管理有限的对象

    public class Pool<T> {

    private int size;

    private List<T> items = new ArrayList<>(); // 资源存放处

    private volatile boolean[] checkOut; // 记录资源是否已经被借出

    private Semaphore available;

    /** * 由于擦除，需要传入class参数 */
    public Pool(Class<T> classObject, int size){
        this.size = size;
        checkOut = new boolean[size];
        available = new Semaphore(size, true);

        // load pool with objects that can be checked out
        for (int i = 0; i < size; i++){
            try {
                items.add(classObject.getDeclaredConstructor().newInstance());
            } catch (InstantiationException e) {
                e.printStackTrace();
            } catch (IllegalAccessException e) {
                e.printStackTrace();
            } catch (InvocationTargetException e) {
                e.printStackTrace();
            } catch (NoSuchMethodException e) {
                e.printStackTrace();
            }
        }
    }

    public T checkOut() throws InterruptedException {
        available.acquire(); // 信号量-1
        return getItem();
    }

    public void checkIn(T x){
        if (releaseItem(x))
            available.release(); // 信号量+1
    }

    /** * 获取资源 */
    private synchronized T getItem() {
        for (int i = 0; i < size; i++){
            if (!checkOut[i]){
                checkOut[i] = true; // 成功借入
                return items.get(i);
            }
        }
        return null; // 资源全部被借出
    }

    /** * 释放资源 */
    private synchronized boolean releaseItem(T item) {
        int index = items.indexOf(item);
        if (index == -1) { // 该资源根本就不是这里的
            return false; 
        } 
        if (checkOut[index]){ // 成功归还
            checkOut[index] = false;
            return true;
        }
        return false;
    }
}

    如果对象的创建过程或者销毁过程十分昂贵，而且对象也会被多次使用，那么使用上述的对象池应该十分理想，这个思想应该和线程池，连接池相通的。

Exchanger

  1. 书本描述：两个任务之间交换对象的栅栏，当这些任务进入栅栏时，它们各自拥有一个对象，当它们离开时，他们都拥有之前由对象持有的对象

  2. 简单来说，就是任务A持有对象A，任务B持有对象B(A和B需要是同样的类型，这个看Exchanger.exchange()方法便可知道)，两个任务共同分享对象Exchanger，只要双方调用了exchange()方法，持有的对象就会交换，即最后任务A持有对象B，任务B持有对象A

  3. 样例代码，还是按照书本上的：

    class ExchangeProducer<T> implements Runnable{

    private Generator<T> generator;

    private Exchanger<List<T>> exchanger;

    private List<T> holder;

    public ExchangeProducer(Generator<T> generator, Exchanger<List<T>> exchanger, List<T> holder) {
        this.generator = generator;
        this.exchanger = exchanger;
        this.holder = holder;
    }

    @Override
    public void run() {
        try {
            while (!Thread.interrupted()){
                for (int i = 0; i < ExchangeDemo.size; i++)
                    holder.add(generator.next()); // 保存生成器生成的对象

                holder = exchanger.exchange(holder); // 进入栅栏，将刚才生产满的列表拿给消费者，换来空的列表
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

class ExchangerConsumer<T> implements Runnable{

    private Exchanger<List<T>> exchanger;

    private List<T> holder;

    private volatile T value;

    public ExchangerConsumer(Exchanger<List<T>> exchanger, List<T> holder) {
        this.exchanger = exchanger;
        this.holder = holder;
    }

    @Override
    public void run() {
        try {
            while (!Thread.interrupted()){
                holder = exchanger.exchange(holder); // 进入栅栏，拿到生产者满的list
                for (T x : holder){
                    value = x; // 取出
                    holder.remove(x); // 在列表中删除 (书本注释: ok for CopyOnWriteArrayList)
                }
            }
        } catch (InterruptedException e) {
        }
        System.out.println("Final value : " + value);
    }
}

public class ExchangeDemo {

    static int size = 10;

    static int delay = 5; // 秒

    public static void main(String[] args) throws InterruptedException {
        ExecutorService service = Executors.newCachedThreadPool();
        Exchanger<List<Fat>> exchanger = new Exchanger<>();
        List<Fat> producerList = new CopyOnWriteArrayList<>();
        List<Fat> consumerList = new CopyOnWriteArrayList<>();
        Generator<Fat> generator = new BasicGenerator(Fat.class);
        service.execute(new ExchangeProducer<>(generator, exchanger, producerList));
        service.execute(new ExchangerConsumer<>(exchanger, consumerList));
        TimeUnit.SECONDS.sleep(delay);
        service.shutdownNow();
    }
}

    值得一提的是，该代码互换的是CopyOnWriteArrayList，这个家伙要细说也需要不少笔墨，这里简单提一下：CopyOnWriteArrayList允许在列表遍历时调用remove()，不会抛出ConcurrentModificationException异常，它一种“写时复制”的容器。明显，如果多个线程只读不写，我们也没必要做并发控制，当写操作发生时，CopyOnWriteArrayList不会直接往当前容器添加，而是先将当前容器进行Copy，复制出一个新的容器，然后新的容器里添加元素，添加完元素之后，再将原容器的引用指向新的容器。这样做的好处是我们可以对CopyOnWrite容器进行并发的读，而不需要加锁，因为当前容器不会添加任何元素。当然，这也导致一些读操作在短时间内看不到更新的内容。在大量读少量写的情况下这应该是不错的选择。当然容器的copy也是一份开销，如果容器很大，这个开销可能也是需要考虑的。CopyOnWrite容器也是一种读写分离的思想，读和写不同的容器。等我更加深入这种写时复制容器再单独做成笔记。

    原文作者：JUC
    原文地址: https://blog.csdn.net/qq_37993487/article/details/86561596
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。
点赞