读写锁、阻塞队列和同步队列
目录
ReadWriteLock（读写锁） #
文档介绍 #
主要是提高效率，允许多个线程读资源，但是只允许一个线程写资源。
运行案例 #
import java.util.HashMap;
import java.util.Map;
public class Main {
public static void main(String[] args) {
MyCache myCache = new MyCache();
int num = 6;
for (int i = 1; i <= num; i++) {
int finalI = i;
new Thread(() -> {
myCache.write(String.valueOf(finalI), String.valueOf(finalI));
},String.valueOf(i)).start();
}
for (int i = 1; i <= num; i++) {
int finalI = i;
new Thread(() -> {
myCache.read(String.valueOf(finalI));
},String.valueOf(i)).start();
}
}
}
/**
* 方法未加锁，导致写的时候被插队
*/
class MyCache {
private Map<String, String> map = new HashMap<>();
public void write(String key, String value) {
System.out.println(Thread.currentThread().getName() + "线程开始写入");
map.put(key, value);
System.out.println(Thread.currentThread().getName() + "线程写入ok");
}
public void read(String key) {
System.out.println(Thread.currentThread().getName() + "线程开始读取");
map.get(key);
System.out.println(Thread.currentThread().getName() + "线程写读取ok");
}
}
运行结果：写的时候被插队
所以如果我们不加锁的情况，多线程的读写会造成数据不可靠的问题。我们也可以采用synchronized这种重量锁和轻量锁 lock去保证数据的可靠。
但是这次我们还可以采用更细粒度的锁：ReadWriteLock 读写锁来保证
改进案例 #
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
public class Main {
public static void main(String[] args) {
MyCache2 myCache = new MyCache2();
int num = 3;
for (int i = 1; i <= num; i++) {
int finalI = i;
new Thread(() -> {
myCache.write(String.valueOf(finalI), String.valueOf(finalI));
},String.valueOf(i)).start();
}
for (int i = 1; i <= num; i++) {
int finalI = i;
new Thread(() -> {
myCache.read(String.valueOf(finalI));
},String.valueOf(i)).start();
}
}
}
class MyCache2 {
private volatile Map<String, String> map = new HashMap<>();
private ReadWriteLock lock = new ReentrantReadWriteLock();
//写的时候不允许插队
public void write(String key, String value) {
lock.writeLock().lock(); // 写锁
try {
System.out.println(Thread.currentThread().getName() + "线程开始写入");
map.put(key, value);
System.out.println(Thread.currentThread().getName() + "线程写入ok");
}finally {
lock.writeLock().unlock(); // 释放写锁
}
}
//读的时候允许线程插队
public void read(String key) {
lock.readLock().lock(); // 读锁
try {
System.out.println(Thread.currentThread().getName() + "线程开始读取");
map.get(key);
System.out.println(Thread.currentThread().getName() + "线程写读取ok");
}finally {
lock.readLock().unlock(); // 释放读锁
}
}
}
BlockingQueue（阻塞队列） #
多线程环境中，通过队列可以很容易实现数据共享，比如经典的“生产者”和“消费者”模型中，通过队列可以很便利地实现两者之间的数据共享。
假设我们有若干生产者线程，另外又有若干个消费者线程。如果生产者线程需要把准备好的数据共享给消费者线程，利用队列的方式来传递数据，就可以很方便地解决他们之间的数据共享问题。
如果生产者和消费者在某个时间段内，万一发生数据处理速度不匹配的情况呢？理想情况下，如果生产者产出数据的速度大于消费者消费的速度，并且当生产出来的数据累积到一定程度的时候，那么生产者必须暂停等待一下(阻塞生产者线程)，以便等待消费者线程把累积的数据处理完毕，反之亦然。
然而，在concurrent包发布以前，在多线程环境下，我们每个程序员都必须去自己控制这些细节，尤其还要兼顾效率和线程安全，而这会给我们的程序带来不小的复杂度。好在此时，强大的concurrent包横空出世了，而他也给我们带来了强大的BlockingQueue。(在多线程领域：所谓阻塞，在某些情况下会挂起线程(即阻塞)，一旦条件满足，被挂起的线程又会自动被唤醒)，
文档介绍 #
BlockingQueue 是 Collection 的一个子类
什么情况下我们会使用阻塞队列
多线程并发处理、线程池
四组API #
BlockingQueue 方法以四种形式出现，对于不能立即满足但可能在将来某一时刻可以满足的操作，这四种形式的处理方式不同：
- 第一种是抛出一个异常
- 第二种是返回一个特殊值（
null或false，具体取决于操作） - 第三种是在操作可以成功前，无限期地阻塞当前线程（阻塞等待）
- 第四种是在放弃前只在给定的最大时间限制内阻塞（超时等待）
下表中总结了这些方法：
| 抛出异常 | 返回一个特殊值 | 阻塞 | 超时 | |
|---|---|---|---|---|
| 插入 | add(e) | offer(e) | put(e) | offer(e, time, unit) |
| 移除 | remove() | poll() | take() | poll(time, unit) |
| 获取队首元素 | element() | peek() | 不可用 | 不可用 |
运行案例 #
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
public class Main {
public static void main(String[] args) throws InterruptedException {
Main.test1();
System.out.println("======================");
Main.test2();
System.out.println("======================");
Main.test3();
System.out.println("======================");
Main.test4();
}
/**
* 抛出异常
*/
public static void test1(){
//需要初始化队列的大小
ArrayBlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
System.out.println(blockingQueue.add("a"));
System.out.println(blockingQueue.add("b"));
System.out.println(blockingQueue.add("c"));
//抛出异常：java.lang.IllegalStateException: Queue full
//System.out.println(blockingQueue.add("d"));
System.out.println(blockingQueue.remove());
System.out.println(blockingQueue.remove());
System.out.println(blockingQueue.remove());
//如果多移除一个
//这个会造成 java.util.NoSuchElementException 抛出异常
//System.out.println(blockingQueue.remove());
}
/**
* 不抛出异常，有返回值
*/
public static void test2(){
ArrayBlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
System.out.println(blockingQueue.offer("a"));
System.out.println(blockingQueue.offer("b"));
System.out.println(blockingQueue.offer("c"));
//添加 一个不能添加的元素 使用offer只会返回false 不会抛出异常
System.out.println(blockingQueue.offer("d"));
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());
//弹出 如果没有元素 只会返回null 不会抛出异常
System.out.println(blockingQueue.poll());
}
/**
* 等待一直阻塞
*/
public static void test3() throws InterruptedException {
ArrayBlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
//一直阻塞 不会返回
blockingQueue.put("a");
blockingQueue.put("b");
blockingQueue.put("c");
//如果队列已经满了， 再进去一个元素 这种情况会一直等待这个队列 什么时候有了位置再进去，程序不会停止
//blockingQueue.put("d");
System.out.println(blockingQueue.take());
System.out.println(blockingQueue.take());
System.out.println(blockingQueue.take());
//如果我们再来一个 这种情况也会等待，程序会一直运行 阻塞
System.out.println(blockingQueue.take());
}
/**
* 等待 超时阻塞
* 这种情况也会等待队列有位置 或者有产品 但是会超时结束
*/
public static void test4() throws InterruptedException {
ArrayBlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
blockingQueue.offer("a");
blockingQueue.offer("b");
blockingQueue.offer("c");
System.out.println("开始等待");
blockingQueue.offer("d",2, TimeUnit.SECONDS); //超时时间2s 等待如果超过2s就结束等待
System.out.println("结束等待");
System.out.println("===========取值==================");
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());
System.out.println("开始等待");
blockingQueue.poll(2,TimeUnit.SECONDS); //超过两秒 我们就不要等待了
System.out.println("结束等待");
}
}
SynchronousQueue（同步队列） #
文档介绍 #
同步队列没有容量，也可以视为容量为1的队列；
进去一个元素，必须等待取出来之后，才能再往里面放入一个元素；
put方法和take方法 #
Synchronized 和其他的 BlockingQueue 不一样 它不存储元素；put了一个元素，就必须从里面先take出来，否则不能再put进去值！并且 SynchronousQueue 的take是使用了lock锁保证线程安全的。
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;
public class Main{
public static void main(String[] args) {
BlockingQueue<String> synchronousQueue = new SynchronousQueue<>();
//添加元素
new Thread(() -> {
try {
System.out.println(Thread.currentThread().getName() + "put 01");
synchronousQueue.put("1");
System.out.println(Thread.currentThread().getName() + "put 02");
synchronousQueue.put("2");
System.out.println(Thread.currentThread().getName() + "put 03");
synchronousQueue.put("3");
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
// 取出元素
new Thread(()-> {
try {
System.out.println(Thread.currentThread().getName() + "take" + synchronousQueue.take());
System.out.println(Thread.currentThread().getName() + "take" + synchronousQueue.take());
System.out.println(Thread.currentThread().getName() + "take" + synchronousQueue.take());
}catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}
/**
Thread-0put 01
Thread-1take1
Thread-0put 02
Thread-1take2
Thread-0put 03
Thread-1take3
**/