文章目录

阻塞队列BlockingQueue传统版生产消费者模式阻塞版生产消费者模式 synchronized & lock

阻塞队列

阻塞队列，首先是一个队列，而一个阻塞队列在数据结构中所起的作用大致如图：

当阻塞队列是空，从队列中获取元素的操作将会被阻塞；当阻塞队列为满，往队列中添加元素的操作将会被阻塞。

阻塞，在多线程环境中会挂起线程（线程阻塞），一旦条件满足，被挂起的线程会被自动唤醒。

BlockingQueue

BlockingQueue让我们不需要关心什么时候需要阻塞线程，什么时候需要唤醒线程:

public class BlockingQueueDemo { public static void main(String[] args) throws Exception{ // List list = null; BlockingQueue<String> 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(blockingQueue.remove()); System.out.println(blockingQueue.remove()); } }

继承树：

ArrayBlockingQueue由数组结构组成的有界阻塞队列LinkedBlockingQueue由链表结构组成的有界阻塞（默认大小Integer.MAX_VALUE()）队列PriorityBlockingQueue支持优先级排序的无界阻塞队列DelayQueue使用优先队列实现延迟无界阻塞队列SynchronizedQueue不存储元素的阻塞队列，也即单个元素的队列LinkedTransferQueue由链表结构组成的无界阻塞队列LinkedBlockingDeque由链表结构组成的双向阻塞队列 方法类型抛出异常特殊值阻塞超时插入add(e)offer(e)put(e)offer(e,time,unit)移除remove()poll()take()poll(time,unit)检查element()peek()不可用不可用 抛出异常当阻塞队列满时,再往队列里面add插入元素会抛IllegalStateException: Queue full当阻塞队列空时,再往队列Remove元素时候回抛出NoSuchElementException特殊值插入方法,成功返回true 失败返回false 移除方法,成功返回元素,队列里面没有就返回null一直阻塞当阻塞队列满时,生产者继续往队列里面put元素,队列会一直阻塞直到put数据or响应中断退出当阻塞队列空时,消费者试图从队列take元素,队列会一直阻塞消费者线程直到队列可用.超时退出当阻塞队列满时,队列会阻塞生产者线程一定时间,超过后限时后生产者线程就会退出 SynchronousQueue 示例 public class SynchronousQueueDemo { public static void main(String[] args) { BlockingQueue<String> blockingQueue = new SynchronousQueue<>(); new Thread(()->{ try{ System.out.println(Thread.currentThread().getName()+"\t put 1"); blockingQueue.put("1"); System.out.println(Thread.currentThread().getName()+"\t put 2"); blockingQueue.put("2"); System.out.println(Thread.currentThread().getName()+"\t put 3"); blockingQueue.put("3"); }catch (Exception e){ e.printStackTrace(); } },"AAA").start(); new Thread(()->{ try{ try{ TimeUnit.SECONDS.sleep(5); } catch (InterruptedException e){ e.printStackTrace(); } System.out.println(Thread.currentThread().getName()+"\t"+blockingQueue.take()); try{ TimeUnit.SECONDS.sleep(5); } catch (InterruptedException e){ e.printStackTrace(); } System.out.println(Thread.currentThread().getName()+"\t"+blockingQueue.take()); try{ TimeUnit.SECONDS.sleep(5); } catch (InterruptedException e){ e.printStackTrace(); } System.out.println(Thread.currentThread().getName()+"\t"+blockingQueue.take()); } catch (InterruptedException e){ e.printStackTrace(); } },"BBB").start(); } }

传统版生产消费者模式

public class ProdConsumer_TraditionDemo { public static void main(String[] args) { ShareData shareData = new ShareData(); new Thread(()->{ for(int i = 1;i<=5;i++){ shareData.increment(); } },"AAA").start(); new Thread(()->{ for(int i =1;i<=5;i++){ shareData.decrement(); } },"BBB").start(); } } class ShareData{ private int number= 0; private Lock lock = new ReentrantLock(); private Condition condition = lock.newCondition(); public void increment(){ lock.lock(); try{ while (number != 0){ condition.await(); } number ++; System.out.println(Thread.currentThread().getName()+"\t"+number); condition.signalAll(); }catch (Exception e){ e.printStackTrace(); }finally { lock.unlock(); } } public void decrement(){ lock.lock(); try{ while (number == 0){ condition.await(); } number--; System.out.println(Thread.currentThread().getName()+"\t"+number); condition.signalAll(); }catch (Exception e){ e.printStackTrace(); }finally { lock.unlock(); } } }

阻塞版生产消费者模式

public class ProdConsumer_BlockQueueDemo { public static void main(String[] args) throws InterruptedException { MyResource myResource = new MyResource(new ArrayBlockingQueue<>(10)); new Thread(()->{ System.out.println(Thread.currentThread().getName()+"\t 生产线程启动"); try { myResource.myProd(); } catch (InterruptedException e) { e.printStackTrace(); } },"prod").start(); new Thread(()->{ System.out.println(Thread.currentThread().getName()+"\t 消费线程启动"); try { myResource.myConsumer(); } catch (InterruptedException e) { e.printStackTrace(); } },"consumer").start(); TimeUnit.SECONDS.sleep(5); System.out.println("5秒钟到，main停止"); myResource.stop(); } } class MyResource{ private volatile boolean FLAG = true; private AtomicInteger atomicInteger = new AtomicInteger(); BlockingQueue<String> blockingQueue = null; public MyResource(BlockingQueue<String> blockingQueue){ this.blockingQueue = blockingQueue; System.out.println(blockingQueue.getClass().getName()); } public void myProd() throws InterruptedException { String data = null; boolean retValue ; while (FLAG){ data = atomicInteger.incrementAndGet()+"" ; retValue = blockingQueue.offer(data,2L, TimeUnit.SECONDS); if(retValue){ System.out.println(Thread.currentThread().getName()+"\t插入队列 "+data+" 成功"); }else{ System.out.println(Thread.currentThread().getName()+"\t插入队列 "+data+" 失败"); } TimeUnit.SECONDS.sleep(1); } System.out.println(Thread.currentThread().getName()+"\t生产停止"); } public void myConsumer() throws InterruptedException { String result = null; while (FLAG){ result = blockingQueue.poll(2L,TimeUnit.SECONDS); if(null == result || result.equalsIgnoreCase("")){ FLAG = false; System.out.println(Thread.currentThread().getName()+"\t 超过2秒，消费退出"); return; } System.out.println(Thread.currentThread().getName()+"\t消费队列 "+result+" 成功"); } } public void stop(){ this.FLAG = false; } }

synchronized & lock

synchronized和lock的区别：

synchronizedlock原始构成synchronized属于JVM层面，底层通过 monitorenter 和 monitorexit 两个指令实现lock是JUC提供的具体类，是API层面的东西用法synchronized 不需要用户手动释放锁，当 synchronized 代码执行完毕之后会自动让线程释放持有的锁lock 需要一般使用try-finally模式去手动释放锁，并且加锁-解锁数量需要一直，否则容易出现死锁或者程序不终止现象等待是否可中断synchronized是不可中断的，除非抛出异常或者程序正常退出lock可中断: 1. 设置超时方法tryLock(time, unit)；2. 使用lockInterruptibly，调用iterrupt方法可中断；是否公平锁synchronized是非公平锁lock默认是非公平锁，但是可以通过构造函数传入boolean类型值更改是否为公平锁锁是否能绑定多个条件（condition)synchronized没有condition的说法，要么唤醒所有线程，要么随机唤醒一个线程lock可以使用condition实现分组唤醒需要唤醒的线程，实现精准唤醒 /* * 题目：多线程之间按顺序调用，实现A->B->C三个线程启动，要求如下： * A打印5次，B打印10次，C打印15次 * 紧接着 * A打印5次，B打印10次，C打印15次 * 。。。。。 * 打印10轮 * * */ public class SyncAndReentrantLockDemo { public static void main(String[] args) { ShareResource shareResource = new ShareResource(); new Thread(()->{ for(int i =1;i<=10;i++){ shareResource.printA(); } },"A").start(); new Thread(()->{ for(int i = 1;i<=10;i++){ shareResource.printB(); } },"B").start(); new Thread(()->{ for(int i =1;i<=10;i++){ shareResource.printC(); } },"C").start(); } } class ShareResource{ private int number = 1; private Lock lock = new ReentrantLock(); private Condition cond1 = lock.newCondition(); private Condition cond2 = lock.newCondition(); private Condition cond3 = lock.newCondition(); public void printA(){ lock.lock(); try{ while (number != 1){ cond1.await(); } for(int i =1;i<=5;i++){ System.out.println(Thread.currentThread().getName()+"\t"+i); } number = 2; cond2.signal(); }catch (Exception e){ e.printStackTrace(); }finally { lock.unlock(); } } public void printB(){ lock.lock(); try{ while (number != 2){ cond2.await(); } for(int i = 1;i<=10;i++){ System.out.println(Thread.currentThread().getName()+"\t"+i); } number = 3; cond3.signal(); }catch (Exception e){ e.printStackTrace(); }finally { lock.unlock(); } } public void printC(){ lock.lock(); try{ while (number != 3){ cond3.await(); } number = 1; cond1.signal(); }catch (Exception e){ e.printStackTrace(); }finally { lock.unlock(); } } }