目录
一.ConcurrentHashMap
1.为什么要使用ConcurrentHashMap?
2.ConcurrentHashMap的类图
3.ConcurrentHashMap的结构图
二.阻塞队列
Java中的7个阻塞队列
ArrayBlockingQueue:一个由数组结构组成的有界阻塞队列。
LinkedBlockingQueue:一个由链表结构组成的有界阻塞队列。可以指定容量也可以无界。
PriorityBlockingQueue:一个支持优先级排序的无界阻塞队列。
DelayQueue:一个使用优先级队列实现的无界阻塞队列。 用于按照指定延迟时间对元素进行排序的阻塞队列。
SynchronousQueue:一个不存储元素的阻塞队列。常用于线程间的手递手传递。
LinkedBlockingDeque:一个由链表结构组成的双向阻塞队列。
一.ConcurrentHashMap
1.为什么要使用ConcurrentHashMap?
-
不安全的HashMap
Java 中的
HashMap
是非线程安全的,这意味着如果多个线程同时访问和修改同一个HashMap
实例,可能会导致不一致的结果或抛出异常。以下是一个示例代码,展示了HashMap
的线程不安全行为:
public static void main(String[] args) {// 创建一个 HashMapMap<Integer, String> map = new HashMap<>();// 创建一个线程池ExecutorService executorService = Executors.newFixedThreadPool(2);// 向 HashMap 中添加键值对的任务Runnable task = () -> {for (int i = 0; i < 1000; i++) {map.put(i, "Value " + i);}};// 启动两个线程同时执行添加任务executorService.submit(task);executorService.submit(task);// 等待线程池执行完毕executorService.shutdown();// 等待一段时间以确保线程池完成try {Thread.sleep(2000);} catch (InterruptedException e) {e.printStackTrace();}// 输出 HashMap 的大小System.out.println("HashMap size: " + map.size());}
在上述示例中,我们创建了一个包含两个线程的线程池,并让它们同时向同一个 HashMap
实例中添加键值对。由于 HashMap
不是线程安全的,这样的并发写入操作可能导致不一致的结果。在某些情况下,可能会抛出 ConcurrentModificationException
异常。
-
效率低下的HashTable
HashTable容器使用synchronized来保证线程安全,但在线程竞争激烈的情况下HashTable 的效率非常低下。因为当一个线程访问HashTable的同步方法,其他线程也访问HashTable的同 步方法时,会进入阻塞或轮询状态。如线程1使用put进行元素添加,线程2不但不能使用put方 法添加元素,也不能使用get方法来获取元素,所以竞争越激烈效率越低。
-
ConcurrentHashMap的锁分段技术可有效提升并发访问率
HashTable容器在竞争激烈的并发环境下表现出效率低下的原因是所有访问HashTable的 线程都必须竞争同一把锁。
ConcurrentHashMap所使用的锁分段技术。首先将数据分成一段一段地存 储,然后给每一段数据配一把锁,当一个线程占用锁访问其中一个段数据的时候,其他段的数据也能被其他线程访问。
2.ConcurrentHashMap的类图
3.ConcurrentHashMap的结构图
二.阻塞队列
阻塞队列(BlockingQueue)是一个支持两个附加操作的队列。这两个附加的操作支持阻塞的插入和移除方法。
-
支持阻塞的插入方法:当队列满时,会阻塞插入元素的线程,直到队列不满。
-
支持阻塞的移除方法:当队列为空时,获取元素的线程会等待队列变为非空。
阻塞队列常用于生产者和消费者场景。生产者就是添加元素的线程,消费者就是获取元素的线程,阻塞队列就是生产者存放元素、消费者获取元素的容器。插入和移除操作的4中处理方式:
方法/处理方式 | 抛出异常 | 返回特殊值 | 一直阻塞 | 超时退出 |
---|---|---|---|---|
插入方法 | add(e) | offer(e) | put(e) | offer(e,time,unit) |
移除方法 | remove() | poll() | take() | poll(time,unit) |
检查方法 | element() | peek() | 不可用 | 不可用 |
-
抛出异常:当队列满了,再插入元素时,会抛出IllegalStateException(“Queue full”)异常。当队列为空,再获取元素,会抛出NoSuchElementException异常。
-
返回特殊值:当队列插入元素时,会返回元素是否插入成功,成功返回true。如果是移除方法,则是从队列去除元素,如果不存在则返回null。
-
一直阻塞:当队列满时,往队列put元素,队列会一直阻塞添加元素的线程,知道队列可用或者响应中断退出。当队列为空时,如果从队列中take元素,队列会阻塞获取元素的线程,知道队列不为空。
-
超时退出:当队列满时,如果插入元素,队列会阻塞插入元素的线程一段时间,超过了指定时间,线程就会退出。
Java中的7个阻塞队列
如果是无界阻塞队列,队列不可能会出现满的情况,所以使用put或offer方法永 远不会被阻塞,而且使用offer方法时,该方法永远返回true。
-
ArrayBlockingQueue:一个由数组结构组成的有界阻塞队列。
public class ArrayBlockingQueueExample {public static void main(String[] args) throws InterruptedException {ArrayBlockingQueue<Integer> arrayBlockingQueue = new ArrayBlockingQueue(5); // 生产者Thread produce = new Thread(()->{try {for (int i = 0; i < 10; i++) {arrayBlockingQueue.put(i);System.out.println("produced:" + i);}} catch (InterruptedException e) {e.printStackTrace();}}); // 消费者Thread consume = new Thread(()->{try {while (true) {int i = arrayBlockingQueue.take();System.out.println("consumed" + i);}} catch (InterruptedException e) {e.printStackTrace();}}); produce.start();consume.start(); Thread.sleep(2000);produce.interrupt();consume.interrupt();} }
-
LinkedBlockingQueue:一个由链表结构组成的有界阻塞队列。可以指定容量也可以无界。
public class LinkedBlockingQueueExample {public static void main(String[] args) throws InterruptedException {LinkedBlockingQueue<Integer> queue = new LinkedBlockingQueue(5); // 同上,创建生产者和消费者线程并启动 // 主线程等待,中断生产者和消费者Thread.sleep(2000);produce.interrupt();consume.interrupt();} }
-
PriorityBlockingQueue:一个支持优先级排序的无界阻塞队列。
public class PriorityBlockingQueueExample {public static void main(String[] args) throws InterruptedException {PriorityBlockingQueue<Integer> queue = new PriorityBlockingQueue<>(); // 同上,创建生产者和消费者线程并启动 // 主线程等待,中断生产者和消费者Thread.sleep(2000);produce.interrupt();consume.interrupt();} }
-
DelayQueue:一个使用优先级队列实现的无界阻塞队列。 用于按照指定延迟时间对元素进行排序的阻塞队列。
public class DelayQueueExample {public static void main(String[] args) throws InterruptedException {DelayQueue<DelayedElement > queue = new DelayQueue<DelayedElement >(); // 创建消费者线程Thread consume = new Thread(() ->{try {while (true) {DelayedElement element = queue.take();System.out.println("consume:" + element.getValue());}} catch (InterruptedException e) {e.printStackTrace();}});consume.start(); // 生产者添加元素queue.put(new DelayedElement("value 5", 1, TimeUnit.SECONDS));queue.put(new DelayedElement("value 4", 2, TimeUnit.SECONDS));queue.put(new DelayedElement("value 3", 3, TimeUnit.SECONDS));queue.put(new DelayedElement("value 2", 4, TimeUnit.SECONDS));queue.put(new DelayedElement("value 1", 5, TimeUnit.SECONDS)); Thread.sleep(10000); consume.interrupt();} static class DelayedElement implements Delayed {private String value;private long delayTime; public String getValue() {return value;} public DelayedElement(String value, long delayTime, TimeUnit timeUnit) {this.value = value;this.delayTime = System.currentTimeMillis() + timeUnit.toMillis(delayTime);} @Overridepublic long getDelay(TimeUnit unit) {return delayTime - System.currentTimeMillis();} @Overridepublic int compareTo(Delayed o) {return Long.compare(this.delayTime, ((DelayedElement) o).delayTime);}} }
-
SynchronousQueue:一个不存储元素的阻塞队列。常用于线程间的手递手传递。
public class SynchronousQueueExample {public static void main(String[] args) throws InterruptedException {SynchronousQueue<Integer> queue = new SynchronousQueue<>();// 同上,创建生产者和消费者线程并启动 // 主线程等待,中断生产者和消费者Thread.sleep(2000);produce.interrupt();consume.interrupt();} }
-
LinkedTransferQueue:一个由链表结构组成的无界阻塞队列。
public class LinkedTransferQueueExample {public static void main(String[] args) throws InterruptedException {LinkedTransferQueue<Integer> queue = new LinkedTransferQueue<>(); // 同上,创建生产者和消费者线程并启动 // 主线程等待,中断生产者和消费者Thread.sleep(2000);produce.interrupt();consume.interrupt();} }
-
LinkedBlockingDeque:一个由链表结构组成的双向阻塞队列。
public class LinkedBlockingDequeExample {public static void main(String[] args) throws InterruptedException {LinkedBlockingDeque<Integer> deque = new LinkedBlockingDeque<>(); // 同上,创建生产者和消费者线程并启动 // 主线程等待,中断生产者和消费者Thread.sleep(2000);produce.interrupt();consume.interrupt();} }