JUC 并发编程--04 常用的辅助类CountDownLatch , CyclicBarrier , Semaphore , 读写锁 , 阻塞队列,CompletableFuture(异步回调)

CountDownLatch 相当于一个减法计数器, 构造方法指定一个数字,比如6, 一个线程执行一次,这个数字减1, 当变为0 的时候, await()方法,才开始往下执行,,

看这个例子

CyclicBarrier 的用法, 字面意思:循环栅栏,

这是构造方法, 第一个参数parties 是线程数量, 第二个参数是barrierAction: 最后一个线程执行完毕之后,要做的操作 ,

重要的方法: await(), 表示这个方法的调用线程已经执行完毕,到达了栅栏, BrokenBarrierException 表示栅栏已破坏原因可能是其中一个线程 await() 时被中断或者超时

基本使用: 一个线程组的所有线程都执行完毕之后, 再往下执行, 再比如一个线程组的计算,最终把计算结果合并

Semaphore 的用法信号量: 控制访问某个资源的并发数量, acquire()这个方法用来获取一个许可, 如果没有获取到就等待, release()方法用来释放这个许可, 实际应用:比如某个共享文件最大多少人访问控制

这里模拟, 业务逻辑,共10个线程,每次只能3个线程并发执行,

结果为:

读写锁 demo

public class ReentrantReadWriteLockDemo {

    public static void main(String[] args) {

        MyCashMap cash = new MyCashMap();

        //写数据,一个一个写, 写缓存的时候要保证原子性

        for (int i = 1; i <= 10; i++) {

            int temp = i;

            new Thread(()->{

                cash.putObject(Thread.currentThread().getName(),temp);

            },"线程"+String.valueOf(i)).start();

        }

        //读数据,都可以读

        for (int i = 0; i < 10; i++) {

            new Thread(()->{

                cash.getObject(Thread.currentThread().getName());

            },"线程"+String.valueOf(i)).start();

        }

    }

}

//自定义缓存

class MyCashMap{

    private Map<String,Object> map = new HashMap<>();

    private ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();

    //加入缓存,  写数据加写锁, 只能一个线程调用

    public void putObject(String key ,Object value){

        readWriteLock.writeLock().lock();

        try {

            System.out.println(key + "---" + value);

            map.put(key,value);

        } catch (Exception e) {

            e.printStackTrace();

        } finally {

            readWriteLock.writeLock().unlock();

        }

    }

    //取数据, 加读锁,都可以读取

    public void getObject(String key){

        readWriteLock.readLock().lock();

        System.out.println(map.get(key));

        readWriteLock.readLock().unlock();

    }

}

阻塞队列 BlockingQueue, 使用场景: 多线程并发处理, 线程池

异步回调类 CompletableFuture

public class test2 {

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(()->{

            try {

                TimeUnit.SECONDS.sleep(2);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            System.out.println(Thread.currentThread().getName()+"runAsync--> void");

        });

        System.out.println(completableFuture.get());//null  异步执行,无返回

        //这里是 异步执行,有结果回调

        CompletableFuture<String> completableFuture1 = CompletableFuture.supplyAsync(()->{

            try {

                TimeUnit.SECONDS.sleep(2);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            System.out.println(Thread.currentThread().getName() + "--supplyAsync-->String");

            return "异步执行结果: 123123";

        });

        System.out.println(completableFuture1.get());

    }

    /**

     * 另一个demo

     * @throws ExecutionException

     * @throws InterruptedException

     */

    @Test

    public void test01() throws ExecutionException, InterruptedException {

        C c = new C();

        String result = c.ask("1+1=?");

        System.out.println(result);

    }

}

class C {

    public String ask(String question) throws ExecutionException, InterruptedException {

        System.out.println("C 收到了一个问题, 交给D解决,C 要出去玩");

        //这里使用异步回调, 有返回

        CompletableFuture<Integer> completableFuture = doQuestion(question);

        System.out.println("C出去玩了");

        //获取D 的执行结果为

        return "D的执行结果为:"+completableFuture.get();

    }

    private CompletableFuture<Integer> doQuestion(String question) {

        CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(()->{

            try {

                TimeUnit.SECONDS.sleep(3);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            System.out.println(Thread.currentThread().getName() + "--supplyAsync-->Integer");

            return 2;

        });

        return completableFuture;

    }

}

https://www.jianshu.com/p/6bac52527ca4

这里使用的CompletableFuture.supplyAsync(runnable,executor),如果没有指定executor,会默认使用ForkJoinPool这个线程池,但是这个线程池不是自定义的,不确定内部的各个参数详情,比如阻塞队列长度,ForkJoinPool是维护了一个无限长的队列来存放任务,如果长度很大很大,阻塞队列永远装不满,反而会出现OOM的风险,所以这里可以自定义线程池

关于forkJoinPool

https://blog.csdn.net/wojiao228925661/article/details/89505575

巴特西

JUC 并发编程--04 常用的辅助类CountDownLatch , CyclicBarrier , Semaphore , 读写锁 , 阻塞队列,CompletableFuture(异步回调)

CyclicBarrier 的用法, 字面意思:循环栅栏,

Semaphore 的用法信号量: 控制访问某个资源的并发数量, acquire()这个方法用来获取一个许可, 如果没有获取到就等待, release()方法用来释放这个许可, 实际应用:比如某个共享文件最大多少人访问控制

读写锁 demo

阻塞队列 BlockingQueue, 使用场景: 多线程并发处理, 线程池

异步回调类 CompletableFuture

最新文章

热门文章

巴特西

JUC 并发编程--04 常用的辅助类CountDownLatch , CyclicBarrier , Semaphore , 读写锁 , 阻塞队列,CompletableFuture(异步回调)

CyclicBarrier 的用法, 字面意思:循环栅栏,

Semaphore 的用法 信号量: 控制访问某个资源的并发数量, acquire()这个方法用来获取一个许可, 如果没有获取到就等待, release()方法用来释放这个许可, 实际应用:比如某个共享文件最大多少人访问控制

读写锁 demo

阻塞队列 BlockingQueue, 使用场景: 多线程并发处理, 线程池

异步回调类 CompletableFuture

最新文章

热门文章

Semaphore 的用法信号量: 控制访问某个资源的并发数量, acquire()这个方法用来获取一个许可, 如果没有获取到就等待, release()方法用来释放这个许可, 实际应用:比如某个共享文件最大多少人访问控制