当前位置：首页>后端>正文

4.java并发包&并发队列

后端2024-05-04 04:27:24

并发包

同步容器类

Vector与ArrayList区别

ArrayList是最常用的List实现类，内部是通过数组实现的，它允许对元素进行快速随机访问。数组的缺点是每个元素之间不能有间隔，当数组大小不满足时需要增加存储能力，就要讲已经有数组的数据复制到新的存储空间中。当从ArrayList的中间位置插入或者删除元素时，需要对数组进行复制、移动、代价比较高。因此，它适合随机查找和遍历，不适合插入和删除。
Vector与ArrayList一样，也是通过数组实现的，不同的是它支持线程的同步，即某一时刻只有一个线程能够写Vector，避免多线程同时写而引起的不一致性，但实现同步需要很高的花费，因此，访问它比访问ArrayList慢

注意: Vector线程安全、ArrayList线程不安全

Vector的add方法

ArrayList的add方法

Vector的get方法

ArrayList的get方法

HashTable与HashMap

HashMap不是线程安全的
HastMap是一个接口是map接口的子接口，是将键映射到值的对象，其中键和值都是对象，并且不能包含重复键，但可以包含重复值。HashMap允许null key和null value，而Hashtable不允许。
HashTable是线程安全的一个Collection。
HashMap是Hashtable的轻量级实现（非线程安全的实现），他们都完成了Map接口，主要区别在于HashMap允许空)null)键值(key),由于非线程安全，效率上能高于Hashtable。
HashMap允许将null作为一个entry的key或者value，而Hashtable不允许。
HashMap把Hashtable的contains方法去掉了，改成containsvalue和containsKey。

注意: HashTable线程安全，HashMap线程不安全。

Hashtable存在contains方法

HashMap不存在contains方法

Hashtable的get方法

HashMap的get方法

Collections提供了将线程不安全的集合转换成线程安全的集合的方法

将HashMap转换成同步的HashMap

ConcurrentHashMap(JDK1.7)

ConcurrentMap接口下有俩个重要的实现 :
ConcurrentHashMap
ConcurrentskipListMap (支持并发排序功能。弥补ConcurrentHashMap)
ConcurrentHashMap内部使用段(Segment)来表示这些不同的部分，每个段其实就是一个小的HashTable,它们有自己的锁。只要多个修改操作发生在不同的段上，它们就可以并发进行。把一个整体分成了16个段(Segment.也就是最高支持16个线程的并发修改操作。
这也是在重线程场景时减小锁的粒度从而降低锁竞争的一种方案。并且代码中大多共享变量使用volatile关键字声明，目的是第一时间获取修改的内容，性能非常好。

分段锁怎么分段。
将一个整体拆分成多个小的HasTable,默认分成16段。

CountDownLatch

CountDownLatch类位于java.util.concurrent包下，利用它可以实现类似计数器的功能。比如有一个任务A，它要等待其他4个任务执行完毕之后才能执行，此时就可以利用CountDownLatch来实现这种功能了。

import java.util.concurrent.CountDownLatch;

public class CountDownLatchDemo {

    public static void main(String[] args) throws InterruptedException {
        System.out.println("等待子线程执行完毕...");
        CountDownLatch countDownLatch = new CountDownLatch(2);
        new Thread(() -> {
            System.out.println("子线程," + Thread.currentThread().getName() + "开始执行...");
            countDownLatch.countDown();// 每次减去1
            System.out.println("子线程," + Thread.currentThread().getName() + "结束执行...");
        }).start();

        new Thread(() -> {
            System.out.println("子线程," + Thread.currentThread().getName() + "开始执行...");
            countDownLatch.countDown();
            System.out.println("子线程," + Thread.currentThread().getName() + "结束执行...");
        }).start();

        countDownLatch.await();// 调用当前方法主线程阻塞  countDown结果为0, 阻塞变为运行状态
        System.out.println("两个子线程执行完毕....");
        System.out.println("继续主线程执行..");
    }
}

运行结果

主线程在两个子线程结束后才能执行

CyclicBarrier

CyclicBarrier初始化时规定一个数目，然后计算调用了CyclicBarrier.await()进入等待的线程数。当线程数达到了这个数目时，所有进入等待状态的线程被唤醒并继续。
CyclicBarrier就象它名字的意思一样，可看成是个障碍，所有的线程必须到齐后才能一起通过这个障碍。
CyclicBarrier初始时还可带一个Runnable的参数，此Runnable任务在CyclicBarrier的数目达到后，所有其它线程被唤醒前被执行。

需求：5个线程写入数据完毕后才打印所有线程执行完毕

import java.util.concurrent.CyclicBarrier;

class Writer extends Thread {
    private CyclicBarrier cyclicBarrier;

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

    @Override
    public void run() {
        System.out.println("线程" + Thread.currentThread().getName() + ",正在写入数据");
        try {
            Thread.sleep(2000);
        } catch (Exception e) {
        }
        System.out.println("线程" + Thread.currentThread().getName() + ",写入数据成功.....");

        try {
            cyclicBarrier.await();
        } catch (Exception e) {
        }
        System.out.println("所有线程执行完毕..........");
    }
}

public class CyclicBarrierDemo {
    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(5);
        for (int i = 0; i < 5; i++) {
            Writer writer = new Writer(cyclicBarrier);
            writer.start();
        }
    }
}

运行结果

Semaphore

Semaphore是一种基于计数的信号量。它可以设定一个阈值，基于此阈值多个线程竞争获取许可信号，做自己的申请后归还，超过阈值后，线程申请许可信号将会被阻塞。

Semaphore可以用来构建一些对象池，资源池之类的，比如数据库连接池，我们也可以创建计数为1的Semaphore，将其作为一种类似互斥锁的机制，这也叫二元信号量，表示两种互斥状态。

用法如下：

availablePermits函数用来获取当前可用的资源数量
wc.acquire(); //申请资源
wc.release();// 释放资源

需求

一个厕所只有3个坑位，但是有10个人来上厕所，那怎么办？

假设10的人的编号分别为1-10，并且1号先到厕所，10号最后到厕所。那么1-3号来的时候必然有可用坑位，顺利如厕，4号来的时候需要看看前面3人是否有人出来了，如果有人出来，进去，否则等待。同样的道理，4-10号也遵守先来先上的规则。

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

class Parent implements Runnable {
    private String name;
    private Semaphore wc;

    public Parent(String name, Semaphore wc) {
        this.name = name;
        this.wc = wc;
    }

    @Override
    public void run() {
        try {
            // 剩下的资源(剩下的茅坑)
            int availablePermits = wc.availablePermits();
            if (availablePermits > 0) {
                System.out.println(name + "有坑了...");
            } else {
                System.out.println(name + "怎么没有坑了...");
            }
            //申请茅坑 如果资源达到3次，就等待
            wc.acquire();
            System.out.println(name + "终于轮我上厕所了..");
            Thread.sleep(new Random().nextInt(1000)); // 模拟上厕所时间。
            System.out.println(name + "厕所上完了...");
            wc.release();

        } catch (Exception e) {

        }
    }
}

public class SemaphoreDemo2 {
    public static void main(String[] args) {
        Semaphore semaphore = new Semaphore(3);
        for (int i = 1; i <= 10; i++) {
            Parent parent = new Parent("第" + i + "个人,", semaphore);
            new Thread(parent).start();
        }
    }
}

开始时，前3个人占到坑位，后面7个人等待

如果有人上完的话，7个人依次占得空位

并发队列

在并发队列上JDK提供了两套实现，一个是以ConcurrentLinkedQueue为代表的高性能队列，一个是以BlockingQueue接口为代表的阻塞队列，无论哪种都继承自Queue。

ConcurrentLinkedDeque

ConcurrentLinkedQueue : 是一个适用于高并发场景下的队列，通过无锁的方式，实现
了高并发状态下的高性能，通常ConcurrentLinkedQueue性能好于BlockingQueue.它
是一个基于链接节点的无界线程安全队列。该队列的元素遵循先进先出的原则。头是最先加入的，尾是最近加入的，该队列不允许null元素。
ConcurrentLinkedQueue重要方法:
add()和offer()都是加入元素的方法(在ConcurrentLinkedQueue中这俩个方法没有任何区别)poll()和peek()都是取头元素节点，区别在于前者会删除元素，后者不会。

import java.util.concurrent.ConcurrentLinkedDeque;

public class ConcurrentLinkedDequeDemo {
    public static void main(String[] args) {
        ConcurrentLinkedDeque q = new ConcurrentLinkedDeque();
        q.offer("zhangsan");
        q.offer("lisi");
        q.offer("wangwu");
        q.offer("123");
        q.offer("456");
        //从头获取元素,删除该元素
        System.out.println(q.poll());
        //从头获取元素,不刪除该元素
        System.out.println(q.peek());
        //获取总长度
        System.out.println(q.size());
    }
}

BlockingQueue

阻塞队列(BlockingQueue)是一个支持两个附加操作的队列。这两个附加的操作是：
在队列空时，获取元素的线程会等待队列变为非空。
当队列满时，存储元素的线程会等待队列变为可用。
阻塞队列常用于生产者和消费者的场景，生产者是往队列里添加元素的线程，消费者是从队列里拿元素的线程。阻塞队列就是生产者存放元素的容器，而消费者也只从容器里拿元素。

BlockingQueue详解

BlockingQueue即阻塞队列，从阻塞这个词可以看出，在某些情况下对阻塞队列的访问可能会造成阻塞。被阻塞的情况主要有如下两种：

当队列满了的时候进行入队列操作
当队列空了的时候进行出队列操作

因此，当一个线程试图对一个已经满了的队列进行入队列操作时，它将会被阻塞，除非有另一个线程做了出队列操作；同样，当一个线程试图对一个空队列进行出队列操作时，它将会被阻塞，除非有另一个线程进行了入队列操作。
在Java中，BlockingQueue的接口位于java.util.concurrent 包中(在Java5版本开始提供)，由上面介绍的阻塞队列的特性可知，阻塞队列是线程安全的。
在新增的Concurrent包中，BlockingQueue很好的解决了多线程中，如何高效安全“传输”数据的问题。通过这些高效并且线程安全的队列类，为我们快速搭建高质量的多线程程序带来极大的便利。本文详细介绍了BlockingQueue家庭中的所有成员，包括他们各自的功能以及常见使用场景。

常用的队列主要有以下两种：(当然通过不同的实现方式，还可以延伸出很多不同类型的队列，DelayQueue就是其中的一种）

先进先出（FIFO）：先插入的队列的元素也最先出队列，类似于排队的功能。从某种程度上来说这种队列也体现了一种公平性。
后进先出（LIFO）：后插入队列的元素最先出队列，这种队列优先处理最近发生的事件。

多线程环境中，通过队列可以很容易实现数据共享，比如经典的“生产者”和“消费者”模型中，通过队列可以很便利地实现两者之间的数据共享。假设我们有若干生产者线程，另外又有若干个消费者线程。如果生产者线程需要把准备好的数据共享给消费者线程，利用队列的方式来传递数据，就可以很方便地解决他们之间的数据共享问题。但如果生产者和消费者在某个时间段内，万一发生数据处理速度不匹配的情况呢？

理想情况下，如果生产者产出数据的速度大于消费者消费的速度，并且当生产出来的数据累积到一定程度的时候，那么生产者必须暂停等待一下（阻塞生产者线程），以便等待消费者线程把累积的数据处理完毕，反之亦然。然而，在concurrent包发布以前，在多线程环境下，我们每个程序员都必须去自己控制这些细节，尤其还要兼顾效率和线程安全，而这会给我们的程序带来不小的复杂度。好在此时，强大的concurrent包横空出世了，而他也给我们带来了强大的BlockingQueue。（在多线程领域：所谓阻塞，在某些情况下会挂起线程（即阻塞），一旦条件满足，被挂起的线程又会自动被唤醒）

ArrayBlockingQueue

ArrayBlockingQueue是一个有边界的阻塞队列，它的内部实现是一个数组。有边界的意思是它的容量是有限的，我们必须在其初始化的时候指定它的容量大小，容量大小一旦指定就不可改变。
ArrayBlockingQueue是以先进先出的方式存储数据，最新插入的对象是尾部，最新移出的对象是头部。

例子：

import java.util.concurrent.TimeUnit;

public class BlockingQueueDemo {
    public static void main(String[] args) throws InterruptedException {
        ArrayBlockingQueue<String> arrays = new ArrayBlockingQueue<String>(3);
        arrays.add("李四");
        arrays.add("张军");
        arrays.add("张军");
        // 添加阻塞队列
        boolean offer = arrays.offer("张三", 1, TimeUnit.SECONDS);
        System.out.println(offer);
        System.out.println(arrays.size());
    }
}

运行结果

LinkedBlockingQueue

LinkedBlockingQueue阻塞队列大小的配置是可选的，如果我们初始化时指定一个大小，它就是有边界的，如果不指定，它就是无边界的。说是无边界，其实是采用了默认大小为Integer.MAX_VALUE的容量。它的内部实现是一个链表。

和ArrayBlockingQueue一样，LinkedBlockingQueue 也是以先进先出的方式存储数据，最新插入的对象是尾部，最新移出的对象是头部。

例子：

import java.util.concurrent.LinkedBlockingQueue;

public class BlockingQueueDemo2 {
    public static void main(String[] args) throws InterruptedException {
        LinkedBlockingQueue linkedBlockingQueue = new LinkedBlockingQueue(3);
        linkedBlockingQueue.add("张三");
        linkedBlockingQueue.add("李四");
        linkedBlockingQueue.add("李四");
        System.out.println(linkedBlockingQueue.size());
    }
}

运行结果

3

PriorityBlockingQueue

PriorityBlockingQueue是一个没有边界的队列，它的排序规则java.util.PriorityQueue一样。需要注意，PriorityBlockingQueue中允许插入null对象。
所有插入PriorityBlockingQueue的对象必须实现java.lang.Comparable接口，队列优先级的排序规则就是按照我们对这个接口的实现来定义的。
另外，我们可以从PriorityBlockingQueue获得一个迭代器Iterator，但这个迭代器并不保证按照优先级顺序进行迭代。

SynchronousQueue

SynchronousQueue队列内部仅允许容纳一个元素。当一个线程插入一个元素后会被阻塞，除非这个元素被另一个线程消费。

使用BlockingQueue模拟生产者与消费者

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

class ProducerThread implements Runnable {
    private BlockingQueue queue;
    private volatile boolean flag = true;
    private static AtomicInteger count = new AtomicInteger();

    public ProducerThread(BlockingQueue queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            System.out.println("生产线程启动...");
            while (flag) {
                System.out.println("正在生产数据....");
                String data = count.incrementAndGet() + "";
                // 将数据存入队列中
                boolean offer = queue.offer(data, 2, TimeUnit.SECONDS);
                if (offer) {
                    System.out.println("生产者,存入" + data + "到队列中,成功.");
                } else {
                    System.out.println("生产者,存入" + data + "到队列中,失败.");
                }
                Thread.sleep(1000);
            }
        } catch (Exception e) {

        } finally {
            System.out.println("生产者退出线程");
        }

    }

    public void stop() {
        this.flag = false;
    }
}

class ConsumerThread implements Runnable {
    private BlockingQueue<String> queue;
    private volatile boolean flag = true;

    public ConsumerThread(BlockingQueue<String> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        System.out.println("消费线程启动...");
        try {
            while (flag) {
                System.out.println("消费者,正在从队列中获取数据..");
                String data = queue.poll(2, TimeUnit.SECONDS);
                if (data != null) {
                    System.out.println("消费者,拿到队列中的数据data:" + data);
                    Thread.sleep(1000);
                } else {
                    System.out.println("消费者,超过2秒未获取到数据..");
                    flag = false;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            System.out.println("消费者退出线程...");
        }
    }
}

public class ProducerAndComsumer {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> queue = new LinkedBlockingQueue<>(10);
        ProducerThread producerThread1 = new ProducerThread(queue);
        ConsumerThread consumerThread1 = new ConsumerThread(queue);
        Thread t1 = new Thread(producerThread1);
        Thread c1 = new Thread(consumerThread1);
        t1.start();
        c1.start();

        // 执行10s
        Thread.sleep(10 * 1000);
        producerThread1.stop();
    }
}

运行结果

BlockingQueue的阻塞是让生产消费有序的关键

查看全文

https://www.xamrdz.com/backend/39e1941455.html

并发包

同步容器类

Vector与ArrayList区别

Vector的add方法

ArrayList的add方法

Vector的get方法

ArrayList的get方法

HashTable与HashMap

Hashtable存在contains方法

HashMap不存在contains方法

Hashtable的get方法

HashMap的get方法

Collections提供了将线程不安全的集合转换成线程安全的集合的方法

将HashMap转换成同步的HashMap

ConcurrentHashMap(JDK1.7)

CountDownLatch

运行结果

CyclicBarrier

运行结果

Semaphore

需求

并发队列

ConcurrentLinkedDeque

BlockingQueue

BlockingQueue详解

ArrayBlockingQueue

例子：

运行结果

LinkedBlockingQueue

例子：

运行结果

PriorityBlockingQueue

SynchronousQueue

使用BlockingQueue模拟生产者与消费者

运行结果

相关文章：