首页 > 编程语言 >Java多线程06——JUC并发包02

Java多线程06——JUC并发包02

时间：2023-02-10 13:33:56浏览次数：43  
标签：02 int CountDownLatch private 线程 new 发包 多线程 public

1 线程的同步工具类 ​​CountDownLatch​​

​​CountDownLatch​​ 同步辅助类，在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待。

  • ​​CountDownLatch​​​ 类是一个同步计数器，构造时传入int参数，该参数就是计数器的初始值，每调用一次 ​​countDown()​​​ 方法，计数器减 1，当计数器大于0时，​​await()​​ 方法会阻塞当前线程继续执行。
  • 由于调用了 ​​countDown()​​​ 方法，所以在当前计数到达零之前，​​await()​​​ 方法会一直受阻塞。之后，会释放所有处于等待的线程，​​await()​​ 方法之后的所有后续调用都将立即返回，这种现象只出现一次，计数无法被重置。一个线程或者多个，等待另外N个线程完成某个事情之后才能执行。

创建线程类

import java.util.concurrent.CountDownLatch;

public class UserThread1 extends Thread {
    private int sum1 = 0;
    private CountDownLatch cd;

    public UserThread1(CountDownLatch cd){
        this.cd = cd;
    }

    @Override
    public void run() {
        for(int i=0;i<=50;i++){
            sum1 += i;
        }
        cd.countDown();
    }

    public int getSum1(){
        return sum1;
    }
}
import java.util.concurrent.CountDownLatch;

public class UserThread2 extends Thread {
    private int sum2 = 0;
    private CountDownLatch cd;

    public UserThread2(CountDownLatch cd){
        this.cd = cd;
    }

    @Override
    public void run() {
        for(int i=51;i<=100;i++){
            sum2 += i;
        }
        cd.countDown();
    }

    public int getSum2(){
        return sum2;
    }
}

线程类中使用了 ​​countDown()​​ 方法，用于更新计数器。

创建测试类

import java.util.concurrent.CountDownLatch;

public class Test {
    public static void main(String[] args) {
        CountDownLatch cd = new CountDownLatch(2);
        UserThread1 thread1 = new UserThread1(cd);
        UserThread2 thread2 = new UserThread2(cd);
        thread1.start();
        thread2.start();

        try {
            cd.await();
            int sum = thread1.getSum1() + thread2.getSum2();
            System.out.println("1~100 的和是：" + sum);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

输出结果为：

1~100 的和是：5050

在测试类中调用了 await() 方法，用于阻塞当前线程，直到 CountDownLatch 初始化时的计数器变为0为止。

Java多线程06——JUC并发包02_java

注意：

  • CountDownLatch 的初始值，应当与所有线程当中的 countDown() 数量相等，
  • 如果上述测试类初始化为1，初始值小于 countDown() 数量，即 ​​CountDownLatch cd = new CountDownLatch(1);​​运行程序后，输出结果可能为

1~100 的和是：1275

  • 如果上述测试类初始化为3，初始值大于 countDown() 数量，即 ​​CountDownLatch cd = new CountDownLatch(3);​​运行程序后，程序将被挂起，无法退出阻塞状态。

2 线程的同步工具类 ​​CyclicBarrier​​

​​CyclicBarrier​​ 是一个同步辅助类，允许一组线程互相等待，直到到达某个公共屏障点（common barrier point）。类似于集合点。

Java多线程06——JUC并发包02_JUC_02

因为该barrier在释放等待线程后可以重用，所以称它为循环的 barrier。

public class TimeCount {
    private int count1;
    private int count2;
    private int sum;

    public int getCount1() {
        return count1;
    }

    public void setCount1(int count1) {
        this.count1 = count1;
    }

    public int getCount2() {
        return count2;
    }

    public void setCount2(int count2) {
        this.count2 = count2;
    }

    public int getSum() {
        return this.count1 + this.count2;
    }

    public void setSum(int sum) {
        this.sum = sum;
    }
}
import java.util.concurrent.CyclicBarrier;

public class UserRunn implements Runnable{
    private TimeCount tc;
    private String name;
    private CyclicBarrier cyclicBarrier;

    public UserRunn(TimeCount tc, String name, CyclicBarrier cyclicBarrier){
        this.tc = tc;
        this.name = name;
        this.cyclicBarrier = cyclicBarrier;
    }

    @Override
    public void run() {
        if(name.equals("爬虫功能")){
            try {
                Thread.sleep(4000);
                tc.setCount1(4000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        else if(name.equals("发送邮件功能")){
            try {
                Thread.sleep(2000);
                tc.setCount2(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        //阻塞点，等所有线程运行完毕，自动解锁
        try {
            cyclicBarrier.await();
            System.out.println("------------" + name + " end-----------");
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
import java.util.concurrent.CyclicBarrier;

public class Test {
    public static void main(String[] args) {
        TimeCount tc = new TimeCount();

        CyclicBarrier cyclicBarrier = new CyclicBarrier(2, new Runnable() {
            @Override
            public void run() {
                int sum = tc.getSum();
                System.out.println("功能总耗时：" + sum);
            }
        });
        new Thread(new UserRunn(tc, "爬虫功能", cyclicBarrier)).start();
        new Thread(new UserRunn(tc, "发送邮件功能", cyclicBarrier)).start();
    }
}

3 线程的同步工具类 ​​Semaphore​​

​​Semaphore​​​ 是一个计数信号量，它的本质是一个共享锁，是基于AQS实现的，通过state变量来实现共享。
通过调用 acquire 方法，对 state 值减去一，当调用 release 的时候，对 state 值加一。
当 state 变量小0时，在AQS队列中阻塞等待。

import java.util.concurrent.Semaphore;

public class Address {
    private int num;
    //通过引入Semaphore，实现停车场限流
    private Semaphore semaphore;

    public Address(int num){
        this.num = num;
        //设置最大可用的并行信号量
        this.semaphore = new Semaphore(num);
    }

    public void autoCar(){
        try {
            //加锁
            semaphore.acquire();

            System.out.println(Thread.currentThread().getName() + "进入停车场");

            Thread.sleep(3000);
            System.out.println(Thread.currentThread().getName() + "离开停车场");
            //释放锁
            semaphore.release();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
public class Car extends Thread {
    private Address address;

    public Car(Address address){
        this.address = address;
    }

    @Override
    public void run() {
        address.autoCar();
    }
}
public class Test {
    public static void main(String[] args) {
        Address address = new Address(2);

        for(int i=0;i<5;i++) {
            new Car(address).start();
        }
    }
}

4 线程的交换类 ​​Exchanger​​

Exchanger（交换者）是一个用于线程间协作的工具类，Exchanger用于进行线程间的数据交换。
它提供一个同步点，在这个同步点，两个线程可以交换彼此的数据。
这两个线程通过 exchange() 方法交换数据。

如果第一个线程先执行 exchange() 方法，它会一直等待第二个线程也执行 exchange() 方法，当两个线程都到达同步点时，这两个线程就可以交换数据，将本线程生产出来的数据传递给对方。

import java.util.concurrent.Exchanger;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Test {
    private static final Exchanger<String> exchanger = new Exchanger<>();
    private static ExecutorService executorService = Executors.newFixedThreadPool(2);

    public static void main(String[] args) {
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                String a = " A 银行转入";
                System.out.println(Thread.currentThread().getName() + a);
                try {
                    String b = exchanger.exchange(a);
                    System.out.println(Thread.currentThread().getName() + b);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        executorService.execute(new Runnable() {
            @Override
            public void run() {
                String a = " B 银行转出";
                System.out.println(Thread.currentThread().getName() + a);
                try {
                    String b = exchanger.exchange(a);
                    System.out.println(Thread.currentThread().getName() + b);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        executorService.shutdown();
    }
}

执行输出

pool-1-thread-1 A 银行转入

pool-1-thread-2 B 银行转出

pool-1-thread-2 A 银行转入

pool-1-thread-1 B 银行转出

5 线程的 ​​Fork​​​-​​Join​​ 机制

Fork/Join 框架是 JAVA7 提供的一个用于并行执行任务的框架，
是一个把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架。

分治法：

把一个规模大的问题划分为规模较小的子问题，然后分而治之，最后合并子问题的解，得到原问题的解。

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;

public class ContTask extends RecursiveTask<Integer> {
    private int start;
    private int end;

    //计算任务量的阈值
    private static final int TASKSIZE = 30;
    private static int count = 0;

    public ContTask(int start, int end){
        this.start = start;
        this.end = end;
    }

    //重写compute方法，任务执行的主要计算
    @Override
    protected Integer compute() {
        int sum = 0;

        System.out.println("开启线程进行计算" + count++);
        boolean state = (end - start) <= TASKSIZE;
        //如果小于等于任务的阈值
        if(state){
            //无需拆分任务计算
            for(int i=start;i<=end;i++){
                sum += i;
            }
        }else{
            //进行拆分任务计算
            System.out.println("这个任务需要进行拆分任务计算。。。" + Thread.currentThread().getName());
            //分割成两个任务
            int middle = (end + start) / 2;
            ContTask contTask1 = new ContTask(start, middle);
            ContTask contTask2 = new ContTask(middle+1, end);
            //开启线程计算分布式任务
            invokeAll(contTask1, contTask2);
            //阻塞，直到任务完成或取消
            Integer tasksum1 = contTask1.join();
            Integer tasksum2 = contTask2.join();
            //结果合并
            sum = tasksum1 + tasksum2;
        }

        return sum;
    }

    public static void main(String[] args) {
        //分布式计算池
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        //初始化设置任务
        ContTask contTask = new ContTask(1, 100);
        //分布式计算任务，提交任务
        ForkJoinTask forkJoinTask = forkJoinPool.submit(contTask);
        //得到最终计算结果
        try {
            System.out.println(forkJoinTask.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
}

执行结果：

开启线程进行计算0

这个任务需要进行拆分任务计算。。。ForkJoinPool-1-worker-1

开启线程进行计算1

这个任务需要进行拆分任务计算。。。ForkJoinPool-1-worker-1

开启线程进行计算2

开启线程进行计算3

开启线程进行计算4

这个任务需要进行拆分任务计算。。。ForkJoinPool-1-worker-2

开启线程进行计算5

开启线程进行计算6

5050

标签：02,int,CountDownLatch,private,线程,new,发包,多线程,public
From： https://blog.51cto.com/u_113754/6047376

相关文章

  • 2023年2月北京限行新规定！手机每周提醒限号怎么设置？
    为了更好地保障市民群众出行需求，统筹考虑到北京市交通运行情况及大气污染防治工作，减小通勤、通学出行的交通压力，自2023年2月13日（星期一）起，北京将恢复实施工作日高峰时段区......
  • js基础笔记学习102-面向对象简介1
    ......
  • NOI2022 游记
    突然发现到现在还没写这篇游记，不过总有些事情要面对，痛定思过才能更好地重新出发虽然主要原因其实是懒得更赛前没啥好说的，前一天就看了眼自己最近做的题，还顺便回忆一下学......
  • 2023.2.9【模板】快速傅里叶变换
    2023.2.9【模板】快速傅里叶变换（FFT）好多天没写博客了qwq题目描述给定一个 n 次多项式 F(x)，和一个 m 次多项式 G(x)。请求出 F(x) 和 G(x) 的卷积。朴素（正常......
  • 网工行业里的多面手，数据通信工程师2023年最新系统学习思维导图整理
    什么是数据通信工程师？数据通信工程师是从事电信网络的维护、参与和指导远端节点设备的安装调试、负责编制相关技术方案和制订维护规范的技术工程师。数据通信行业前景如何？......
  • CSP-J 2022 T2-解密
    原题目链接题目描述给定一个正整数\(k\)，有\(k\)次询问，每次给定三个正整数\(n_i,e_i,d_i\)，求两个正整数\(p_i,q_i\)，使\(n_i=p_i\timesq_i\)、\(e_i\timesd......
  • CSP-J 2022 T1-乘方
    题目描述小文同学刚刚接触了信息学竞赛，有一天她遇到了这样一个题：给定正整数\(a\)和\(b\)，求\(a^b\)的值是多少。\(a^b\)即\(b\)个\(a\)相乘的值，例如\(2^3\)即......
  • 按快捷键F3显示或者隐藏UltraSearch主窗口 2023-2-10 (快捷键F3 位于键盘主键区数字4
     按快捷键F3显示或者隐藏UltraSearch主窗口2023-2-10(快捷键F3位于键盘主键区数字4键正上方)按快捷键`显示或者隐藏TotalCommander10.52主窗口2023-2-5(快捷......
  • 如何使用 Python 编程进行多线程并发？
    当单线程python爬虫已经不能满足企业需求时，很多程序员会进行改代码或者增加服务器数量，这样虽说也能达到效果，但是对于人力物力也是一笔不小的消耗。如果是技术牛点的，正常都......
  • 按下空格发射子弹 pygame 230210
    逻辑捕捉用户按下空格的事件创建一个子弹对象在游戏循环中让子弹往上飞行定义子弹模板按下空格拷备子弹让子弹显示并飞......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99