10.读写锁
package test02;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.ReentrantReadWriteLock;
/**
-
独占锁（写锁） 一次只能被一个线程占用
-
共享锁（读锁） 多个线程可以同时占有
-
ReadWriteLock
-
读-读 可以共存！
-
读-写 不能共存！
-
写-写 不能共存！
*/
public class ReadWriteLockDemo {
public static void main(String[] args) {
// MyCache myCache = new MyCache();
MyCacheLock myCache = new MyCacheLock();
for (int i = 1 ; i <= 5 ; i++) {
final int temp=i;
new Thread(()->{
myCache.put(temp+"",temp+"");
},String.valueOf(i)).start();
}
for (int i = 1 ; i <= 5 ; i++) {
final int temp=i;
new Thread(()->{
myCache.get(temp+"");
},String.valueOf(i)).start();
}
}
}
class MyCacheLock{
private volatile Map<String,Object> map=new HashMap<>();
//读写锁，更加细粒度的控制
private ReentrantReadWriteLock readWriteLock= new ReentrantReadWriteLock();
//存 写入的时候 ，只希望只有一个线程去写
public void put(String key,Object value){
readWriteLock.writeLock().lock();
try{
System.out.println(Thread.currentThread().getName()+"写入"+key);
map.put(key, value);
System.out.println(Thread.currentThread().getName()+"写入OK");
}catch (Exception e){
e.printStackTrace();
}finally {
readWriteLock.writeLock().unlock();
}
}
//取 ,读 ，所有人都可以读
public void get(String key){
readWriteLock.readLock().lock();
try{
System.out.println(Thread.currentThread().getName()+"读取"+key);
map.get(key);
System.out.println(Thread.currentThread().getName()+"读取OK");
}catch (Exception e){
e.printStackTrace();
}finally {
readWriteLock.readLock().unlock();
}
}
}
class MyCache{
private volatile Map<String,Object> map=new HashMap<>();
//存
public void put(String key,Object value){
System.out.println(Thread.currentThread().getName()+"写入"+key);
map.put(key, value);
System.out.println(Thread.currentThread().getName()+"写入OK");
}
//取
public void get(String key){
System.out.println(Thread.currentThread().getName()+"读取"+key);
map.get(key);
System.out.println(Thread.currentThread().getName()+"读取OK");
}
}
11.阻塞队列
阻塞
队列
BlockingQueue 不是新的东西
什么情况下我们会使用阻塞队列 ：多线程并发处理，线程池！
学会使用队列
添加 移除
四组API
-
抛出异常
-
不会抛出异常
-
阻塞 等待
-
超时等待
| 方式 | 抛出异常 | 有返回值，不抛出异常 | 阻塞 等待 | 超时等待 |
| --- | --- | --- | --- | --- |
| 添加 | add() | offer() | put() | offer(“d”, 2,TimeUnit.SECONDS) |
| 移除 | remove() | poll() | take() | poll(2,TimeUnit.SECONDS) |
| 判断队列首 | element() | peek() | - | - |
package test03;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
public class Test {
public static void main(String[] args) throws InterruptedException {
test4();
}
/*
- 抛出异常
*/
public static void test1(){
//队列的大小
ArrayBlockingQueue<Object> blockingQueue = new ArrayBlockingQueue<>(3);
System.out.println(blockingQueue.add("a"));
System.out.println(blockingQueue.add("b"));
System.out.println(blockingQueue.add("c"));
System.out.println(blockingQueue.element());//检测队首元素
System.out.println("------------------------");
//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<Object> blockingQueue = new ArrayBlockingQueue<>(3);
System.out.println(blockingQueue.offer("a"));
System.out.println(blockingQueue.offer("b"));
System.out.println(blockingQueue.offer("c"));
System.out.println(blockingQueue.offer("d")); //false 不抛出异常
System.out.println(blockingQueue.peek());//检测队首元素
System.out.println("-------------------------");
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());//null
}
/*
- 等待 ，阻塞 （一直阻塞）
*/
public static void test3() throws InterruptedException {
//队列的大小
ArrayBlockingQueue<Object> blockingQueue = new ArrayBlockingQueue<>(3);
//一直阻塞
blockingQueue.put("a");
blockingQueue.put("b");
blockingQueue.put("c");
// blockingQueue.put("d");//程序卡住，不停止
System.out.println("-------------------------");
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<Object> blockingQueue = new ArrayBlockingQueue<>(3);
System.out.println(blockingQueue.offer("a"));
System.out.println(blockingQueue.offer("b"));
System.out.println(blockingQueue.offer("c"));
// System.out.println(blockingQueue.offer("d", 2,TimeUnit.SECONDS));//等待超过2s就退出
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));//等待超过2s就退出
}
}
12.同步队列 SynchronousQueue
没有容量
进去一个元素，必须等待取出来之后，才能再往里面放一个元素！
put、take
package test03;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
/**
-
同步队列
-
和其他的BlockingQueue 不一样
-
put了一个元素，必须先take出来，否则不能在put进去值！
*/
public class SynchronousQueueDemo {
public static void main(String[] args) {
BlockingQueue blockingQueue = new SynchronousQueue();
new Thread(()->{
try{
System.out.println(Thread.currentThread().getName()+"--->put 1");
blockingQueue.put("1");
System.out.println(Thread.currentThread().getName()+"--->put 1");
blockingQueue.put("2");
System.out.println(Thread.currentThread().getName()+"--->put 1");
blockingQueue.put("3");
}catch (Exception e) {
e.printStackTrace();
}
},"T1").start();
new Thread(()->{
try {
TimeUnit.SECONDS.sleep(3);
System.out.println(Thread.currentThread().getName()+blockingQueue.take());
TimeUnit.SECONDS.sleep(3);
System.out.println(Thread.currentThread().getName()+blockingQueue.take());
TimeUnit.SECONDS.sleep(3);
System.out.println(Thread.currentThread().getName()+blockingQueue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
},"T2").start();
}
}
