Java并发辅助类的使用

Posted 2021-01-13 吴庆龙的技术轮子

目录

• 概述
• CountdownLatch
  • 构造方法
  • 重要方法
  • 使用示例
• CyclicBarrier
  • 构造方法
  • 使用示例
• Semaphore
  • 构造方法
  • 重要方法
  • 使用示例
• 总结

概述

本文主要对Java中的关于并发的类的使用写一遍Demo.

具体涉及到的类有:

1. CountdownLatch
2. CyclicBarrier
3. Semaphore

CountdownLatch

CountdownLatch类位于java.util.concurrent包下, 利用它可以实现类似计数器的功能, 比如有一个任务C, 它需要等到任务A和任务B执行完成之后才能执行. 此时就可以利用CountdownLatch进行实现.

构造方法

public CountDownLatch(int count) {
    // 省略...
}

count一般指线程的数量.

重要方法

// 调用await()方法的线程会挂起, 它会等到count=0时才继续执行
public void await() throws InterruptedException {
    // 省略...
}

// 与await()方法类似, 只不过是等待一段时间之后不管count值是多少, 都要继续向下执行
public boolean await(long timeout, TimeUnit unit)
    throws InterruptedException {
    // 省略...
}

// 将count值减1
public void countDown() {
    // 省略...
}

使用示例

import java.util.Random;
import java.util.concurrent.CountDownLatch;

public class CountdownLatchDemo {

    public static void main(String[] args) {
        new CountdownLatchDemo().start();
    }

    private void start() {

        // 定义线程的数量
        int num = 5;
        CountDownLatch latch = new CountDownLatch(num);
        for (int i = 0; i < num; i++) {
            new Thread(new RunImpl(latch), "线程"+i).start();
        }

        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("所有线程写入完毕，继续处理其他任务...");

    }

    class RunImpl implements Runnable {

        private CountDownLatch latch;

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

        @Override
        public void run() {
            try {
                System.out.println(Thread.currentThread().getName()+"正在写入数据...");
                Thread.sleep(new Random().nextInt(2000)); //以睡眠来模拟写入数据操作

                System.out.println(Thread.currentThread().getName()+"写入数据完毕，等待其他线程写入完毕");

                // 完成任务, 计数器减1
                latch.countDown();
            } catch (Exception e){
                e.printStackTrace();
            }

        }
    }

}

输出结果如下:

线程0正在写入数据...
线程1正在写入数据...
线程2正在写入数据...
线程3正在写入数据...
线程4正在写入数据...
线程1写入数据完毕，等待其他线程写入完毕
线程4写入数据完毕，等待其他线程写入完毕
线程3写入数据完毕，等待其他线程写入完毕
线程2写入数据完毕，等待其他线程写入完毕
线程0写入数据完毕，等待其他线程写入完毕
所有线程写入完毕，继续处理其他任务...

初始化时CountDownLatch的count为5, 随后启动了5个线程, 然后调用了latch.await(), 挂起main线程, 等待5个线程执行完毕. 每个线程执行完毕时调用了latch.countDown(); 让count减1, 等到count为0时main线程继续执行.

CyclicBarrier

与CountdownLatch类似, 只不过它可以进行重用, 而且还可以支持任务执行完之后随即选择一个线程来执行构造方法中传入的线程任务, 还有一点不同是, 主线程main不会堵塞.

构造方法

public CyclicBarrier(int parties) {
    // 省略...
}

public CyclicBarrier(int parties, Runnable barrierAction) {
    // 省略...
}

parties是指让多少个线程或者任务等待至barrier状态, 参数barrierAction是当所有线程都到达barrier状态时, 用最后一个到达barrier的线程执行barrierAction.

使用示例

import java.util.Random;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierDemo {

    public static void main(String[] args) {
        new CyclicBarrierDemo().start();
    }

    private void start() {

        int num = 5;
        CyclicBarrier barrier = new CyclicBarrier(num, () -> System.out.println("最后调用了线程: " + Thread.currentThread().getName()));

        for (int i = 0; i < num; i++) {
            new Thread(new RunImpl(barrier), "线程"+i).start();
        }

        // 等待上面线程执行完后再次执行
        // try {
        //     Thread.sleep(10000);
        // } catch (InterruptedException e) {
        //     e.printStackTrace();
        // }
        // for (int i = 0; i < num; i++) {
            // new Thread(new RunImpl(barrier), "线程"+i).start();
        // }

        System.out.println("所有线程写入完毕，继续处理其他任务...");

    }

    class RunImpl implements Runnable {

        private CyclicBarrier cyclicBarrier;

        public RunImpl(CyclicBarrier cyclicBarrier) {
            this.cyclicBarrier = cyclicBarrier;
        }

        @Override
        public void run() {
            try {
                System.out.println(Thread.currentThread().getName()+"正在写入数据...");
                Thread.sleep(new Random().nextInt(2000)); //以睡眠来模拟写入数据操作

                System.out.println(Thread.currentThread().getName()+"写入数据完毕，等待其他线程写入完毕");
                cyclicBarrier.await();

                System.out.println(Thread.currentThread().getName()+"await()完成.");
            } catch (Exception e){
                e.printStackTrace();
            }
        }
    }

}

输出结果如下:

线程0正在写入数据...
线程2正在写入数据...
所有线程写入完毕，继续处理其他任务...
线程1正在写入数据...
线程3正在写入数据...
线程4正在写入数据...
线程1写入数据完毕，等待其他线程写入完毕
线程4写入数据完毕，等待其他线程写入完毕
线程2写入数据完毕，等待其他线程写入完毕
线程3写入数据完毕，等待其他线程写入完毕
线程0写入数据完毕，等待其他线程写入完毕
最后调用了线程: 线程0
线程0await()完成.
线程1await()完成.
线程2await()完成.
线程4await()完成.
线程3await()完成.

发现, main方法并不会等待所有线程执行完毕之后执行.

这里没有使用countDown()进行计数器减1, 然后使用await()方法等待计数器变为0, 而是使用await()方法, 并且在await()方法中进行计数器等于0的判断. 并且, 在所有线程到达barrier时, 用最后一个到达的线程去执行barrierAction.

Semaphore

Semaphore可以翻译为信号量, 它可以控制同时并发的线程数量, 通过acquire()方法获取许可, 如果没有就等待, 而release()释放一个许可, 注意: 在释放前必须先获取许可.

构造方法

public Semaphore(int permits) {
    sync = new NonfairSync(permits);
}
public Semaphore(int permits, boolean fair) {
    sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

permits是允许同时并发的线程数量, 而fair是控制是否允许等待时间越长的线程优先获取许可(公平与不公平).

重要方法

public void acquire() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}
public void release() {
    sync.releaseShared(1);
}

acquire()方法获取许可, release()方法释放许可.

使用示例

import java.util.Random;
import java.util.concurrent.Semaphore;

public class SemaphoreDemo {

    public static void main(String[] args) {
        new SemaphoreDemo().start();
    }

    private void start() {
        int num = 8; // 工人数
        Semaphore semaphore = new Semaphore(5); // 机器数目

        for (int i = 0; i < num; i++) {
            new Thread(new RunImpl(i, semaphore)).start();
        }

    }

    class RunImpl implements Runnable {

        private int num;
        private Semaphore semaphore;
        public RunImpl(int num, Semaphore semaphore){
            this.num = num;
            this.semaphore = semaphore;
        }

        @Override
        public void run() {
            try {
                semaphore.acquire();
                System.out.println("工人" + this.num + "占用一个机器在生产...");
                Thread.sleep(new Random().nextInt(2000)); // 以睡眠来模拟写入数据操作

                System.out.println("工人" + this.num + "释放出机器");
                semaphore.release();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}

输出结果如下:

工人0占用一个机器在生产...
工人3占用一个机器在生产...
工人2占用一个机器在生产...
工人1占用一个机器在生产...
工人4占用一个机器在生产...
工人4释放出机器
工人5占用一个机器在生产...
工人1释放出机器
工人6占用一个机器在生产...
工人6释放出机器
工人7占用一个机器在生产...
工人5释放出机器
工人0释放出机器
工人3释放出机器
工人2释放出机器
工人7释放出机器

这里控制并发线程数量为5个, 工人多了只能等待其它工人释放机器, 再去公平/不公平竞争去使用机器.

总结

1. CountDownLatch一般用于线程A等待若干个线程执行完成任务之后, 它才执行. 不可以重用.
2. CyclicBarrier一般用于一组线程相互等待至barrier时, 先用最后一个到达的线程执行barrierAction, 再同时执行await()之后的代码. 可以重用.
3. Semaphore其实和锁有点类似, 用于控制对某个资源的并发控制.