Java学习笔记（并发包介绍）

用户5513909

发布于 2023-04-25 11:27:51

2360

发布于 2023-04-25 11:27:51

并发包的来历：

在实际开发中如果不需要考虑线程安全问题，大家不需要做线程安全，因为如果做了反而性能不好！但是开发中有很多业务是需要考虑线程安全问题的，此时就必须考虑了。否则业务出现问题。Java为很多业务场景提供了性能优异，且线程安全的并发包，程序员可以选择使用！

ConcurrentHashMap介绍

为什么要用ConcurrentHashMap？

HashMap线程不安全因为多线程环境下，使用Hashmap进行put操作可能会引起死循环，导致CPU利用率接近100%，所以在并发情况下不能使用HashMap。
Hashtable线程安全但效率低下 HashTable容器使用synchronized来保证线程安全，但在线程竞争激烈的情况下HashTable的效率非常低下。因为当一个线程访问HashTable的同步方法，其他线程也访问HashTable的同步方法时，会进入阻塞状态。如线程1使用put进行元素添加，线程2不但不能使用put方法添加元素，也不能使用get方法来获取元素，所以竞争越激烈效率越低。

package 并发包;

import java.util.HashMap;
import java.util.Hashtable;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class ConcurrentHashMapDemo {
    public static Map<String, String> maps = new ConcurrentHashMap<>();

    public static void main(String[] args) {
        Runnable target = new MyRunnable();
        Thread t1 = new Thread(target, "线程-1");
        Thread t2 = new Thread(target, "线程-2");
        Thread t3 = new Thread(target, "线程-3");
        t1.start();
        t2.start();
        t3.start();
        try{
            t1.join();//t1跑完，主线程不能竞争t1cpu
            t2.join();
            t3.join();
        }catch (Exception e){
            e.printStackTrace();
        }
        System.out.println("元素个数：" + maps.size());
    }

}

class MyRunnable implements Runnable{
    @Override
    public void run(){
        for (int i = 0; i < 500000; i++) {
            ConcurrentHashMapDemo.maps.put(Thread.currentThread().getName()+i, Thread.currentThread().getName()+i);
        }
    }
}

ConcurrentHashMap高效的原因：CAS + 局部(synchronized)锁定

CountDownLatch介绍

CountDownLatch允许一个或多个线程等待其他线程完成操作，再执行自己。

例如：线程1要执行打印：A和C，线程2要执行打印：B，但线程1在打印A后，要线程2打印B之后才能打印C，所以：线程1在打印A后，必须等待线程2打印完B之后才能继续执行。执行结果：会保证按：A B C的顺序打印。说明： CountDownLatch中count down是倒数的意思，latch则是门栓，的含义。整体含义可以理解为倒数的门栓，似乎有一点“三二一，芝麻开门”的感觉。 CountDownLatch是通过一个计数器来实现的，每当一个线程完成了自己的任务后，可以调用countDown()方法让计数器-1，当计数器到达0时，调用CountDownLatch。 await()方法的线程阻塞状态解除，继续执行。

package 并发包;

import java.util.concurrent.CountDownLatch;

public class CountDownLatchDemo {
    public static void main(String[] args) {
        //创建一个CountDownLatch用于监督A\B
        CountDownLatch c = new CountDownLatch(1);
        new ThreadA(c).start();
        new ThreadB(c).start();
    }
}

class ThreadA extends Thread{
    private CountDownLatch c;
    public ThreadA(CountDownLatch c){
        this.c = c;
    }

    @Override
    public void run() {
        System.out.println("A");
        //等到自己
        try {
            c.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("C");
    }
}


class ThreadB extends Thread{

    private CountDownLatch c;
    public ThreadB(CountDownLatch c){
        this.c = c;
    }

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

CyclicBarrier介绍

CyclicBarrier的字面意思是可循环使用（Cyclic）的屏障（Barrier）。它要做的事情是，让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续运行。

例如：公司召集5名员工开会，等5名员工都到了，会议开始。

我们创建5个员工线程，1个开会线程，几乎同时启动，使用CyclicBarrier保证5名员工线程全部执行后，再执行开会线程。使用场景：CyclicBarrier可以用于多线程计算数据，最后合并计算结果的场景。

需求：使用两个线程读取2个文件中的数据，当两个文件中的数据都读取完毕以后，进行数据的汇总操作

package 并发包;

import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierDemo {
    public static void main(String[] args) {
        //创建一个循环屏障对象，等五个线程执行完毕，触发一次
        CyclicBarrier c = new CyclicBarrier(5,new Meeting());
        for (int i = 0; i < 10; i++) {
            new EmpThread("员工"+i,c).start();
        }
    }
}

class Meeting implements Runnable{

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName()+"开始组织会议");
    }
}

class EmpThread extends Thread{
    private CyclicBarrier c;
    public EmpThread(String name, CyclicBarrier c) {
        super(name);
        this.c = c;
    }

    @Override
    public void run() {
        try {
            Thread.sleep(3000);
            System.out.println(Thread.currentThread().getName() + "正在进入会议室");
            c.await();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

Semaphore介绍

Semaphore（发信号）的主要作用是控制线程的并发数量。

synchronized可以起到"锁"的作用，但某个时间段内，只能有一个线程允许执行。

Semaphore可以设置同时允许几个线程执行。

Semaphore字面意思是信号量的意思，它的作用是控制访问特定资源的线程数目。

package 并发包;

import java.util.concurrent.Semaphore;

public class SemaphoreDemo {
    public static void main(String[] args) {
        Service service = new Service();
        for (int i = 0; i < 5; i++) {
            MyThread a = new MyThread(service);
            a.setName("线程-"+i);
            a.start();
        }
    }
}

class MyThread extends Thread{
    private Service service;

    public MyThread(Service service) {
        this.service = service;
    }

    @Override
    public void run() {
        service.login();
    }
}


class Service{
    private Semaphore semaphore = new Semaphore(1);
    //登录
    public void login(){
        try {
            semaphore.acquire();
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName()
        + "进入， 时间=" + System.currentTimeMillis());
        try {
            Thread.sleep(1000);
            System.out.println(Thread.currentThread().getName()+"登录成功");
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName()
                + "离开， 时间=" + System.currentTimeMillis());
        semaphore.release();
    }
}

Exchanger介绍

概述： Exchanger（交换者）是一个用于线程间协作的工具类。Exchanger用于进行线程间的数据交换。

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

package 并发包;

import java.util.concurrent.Semaphore;

public class SemaphoreDemo {
    public static void main(String[] args) {
        Service service = new Service();
        for (int i = 0; i < 5; i++) {
            MyThread a = new MyThread(service);
            a.setName("线程-"+i);
            a.start();
        }
    }
}

class MyThread extends Thread{
    private Service service;

    public MyThread(Service service) {
        this.service = service;
    }

    @Override
    public void run() {
        service.login();
    }
}


class Service{
    private Semaphore semaphore = new Semaphore(1);
    //登录
    public void login(){
        try {
            semaphore.acquire();
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName()
        + "进入， 时间=" + System.currentTimeMillis());
        try {
            Thread.sleep(1000);
            System.out.println(Thread.currentThread().getName()+"登录成功");
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName()
                + "离开， 时间=" + System.currentTimeMillis());
        semaphore.release();
    }
}

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2020-08-11，如有侵权请联系 cloudcommunity@tencent.com 删除

java