package cn.qlq.thread.thirteen;import java.util.LinkedList;
import java.util.Queue;public class Demo1 {public static void main(String[] args) {Queue<String> queue = new LinkedList<String>();String poll = queue.poll();System.out.println(poll);String remove = queue.remove();System.out.println(remove);}
}

package java.util.concurrent;import java.util.Collection;
import java.util.Queue;public interface BlockingQueue<E> extends Queue<E> {boolean add(E e);boolean offer(E e);void put(E e) throws InterruptedException;boolean offer(E e, long timeout, TimeUnit unit)throws InterruptedException;E take() throws InterruptedException;E poll(long timeout, TimeUnit unit)throws InterruptedException;int remainingCapacity();boolean remove(Object o);public boolean contains(Object o);int drainTo(Collection<? super E> c);int drainTo(Collection<? super E> c, int maxElements);
}

    final ReentrantLock lock;private final Condition notEmpty;private final Condition notFull;public ArrayBlockingQueue(int capacity) {this(capacity, false);}public ArrayBlockingQueue(int capacity, boolean fair) {if (capacity <= 0)throw new IllegalArgumentException();this.items = new Object[capacity];lock = new ReentrantLock(fair);notEmpty = lock.newCondition();notFull =  lock.newCondition();}

package cn.qlq.thread.thirteen;import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;public class Demo2 {private static int num ;private static final Logger LOGGER = LoggerFactory.getLogger(Demo2.class);public static void main(String[] args) throws InterruptedException {final BlockingQueue<String> strings = new ArrayBlockingQueue<>(1);//必须指定容量(指定容器最多为1)Thread producer = new Thread(new Runnable() {@Overridepublic void run() {try {for  (int i=0;i<5;i++) {String ele = "ele"+(++num);strings.put(ele);LOGGER.info("ThreadName ->{} put ele->{}",Thread.currentThread().getName(),ele);}} catch (InterruptedException e) {e.printStackTrace();}}},"producer");producer.start();Thread consumer = new Thread(new Runnable() {@Overridepublic void run() {try {for (int i=0;i<5;i++) {Thread.sleep(1*1000);String take = strings.take();LOGGER.info("ThreadName ->{} take ele->{}",Thread.currentThread().getName(),take);}} catch (InterruptedException e) {e.printStackTrace();}}},"consumer");consumer.start();}
}

    private transient Node<E> head;
private transient Node<E> last;
private final ReentrantLock takeLock = new ReentrantLock();
private final Condition notEmpty = takeLock.newCondition();
private final ReentrantLock putLock = new ReentrantLock();
private final Condition notFull = putLock.newCondition();public LinkedBlockingQueue() {this(Integer.MAX_VALUE);}public LinkedBlockingQueue(int capacity) {if (capacity <= 0) throw new IllegalArgumentException();this.capacity = capacity;last = head = new Node<E>(null);}public LinkedBlockingQueue(Collection<? extends E> c) {this(Integer.MAX_VALUE);final ReentrantLock putLock = this.putLock;putLock.lock(); // Never contended, but necessary for visibilitytry {int n = 0;for (E e : c) {if (e == null)throw new NullPointerException();if (n == capacity)throw new IllegalStateException("Queue full");enqueue(new Node<E>(e));++n;}count.set(n);} finally {putLock.unlock();}}

package cn.qlq.thread.thirteen;import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;public class Demo3 {private static int num ;private static final Logger LOGGER = LoggerFactory.getLogger(Demo3.class);public static void main(String[] args) throws InterruptedException {final BlockingQueue<String> strings = new LinkedBlockingQueue<>(3);Runnable producerRun = new Runnable() {@Overridepublic synchronized void  run() {//加同步避免出现线程非安全try {for  (int i=0;i<5;i++) {Thread.sleep(1000);String ele = "ele"+(++num);strings.put(ele);LOGGER.info("ThreadName ->{} put ele->{}",Thread.currentThread().getName(),ele);}} catch (InterruptedException e) {e.printStackTrace();}}};Thread producer = new Thread(producerRun,"producer");producer.start();Thread producer2 = new Thread(producerRun,"producer2");producer2.start();Runnable consumerRun = new Runnable() {@Overridepublic void run() {try {for (int i=0;i<5;i++) {Thread.sleep(3000);String take = strings.take();LOGGER.info("ThreadName ->{} take ele->{}",Thread.currentThread().getName(),take);}} catch (InterruptedException e) {e.printStackTrace();}}};Thread consumer = new Thread(consumerRun,"consumer");Thread consumer1 = new Thread(consumerRun,"consumer1");consumer.start();consumer1.start();}
}

   /*** Default array capacity.*/private static final int DEFAULT_INITIAL_CAPACITY = 11;public PriorityBlockingQueue() {this(DEFAULT_INITIAL_CAPACITY, null);}public PriorityBlockingQueue(int initialCapacity) {this(initialCapacity, null);}public PriorityBlockingQueue(int initialCapacity,Comparator<? super E> comparator) {if (initialCapacity < 1)throw new IllegalArgumentException();this.lock = new ReentrantLock();this.notEmpty = lock.newCondition();this.comparator = comparator;this.queue = new Object[initialCapacity];}public PriorityBlockingQueue(Collection<? extends E> c) {this.lock = new ReentrantLock();this.notEmpty = lock.newCondition();boolean heapify = true; // true if not known to be in heap orderboolean screen = true;  // true if must screen for nullsif (c instanceof SortedSet<?>) {SortedSet<? extends E> ss = (SortedSet<? extends E>) c;this.comparator = (Comparator<? super E>) ss.comparator();heapify = false;}else if (c instanceof PriorityBlockingQueue<?>) {PriorityBlockingQueue<? extends E> pq =(PriorityBlockingQueue<? extends E>) c;this.comparator = (Comparator<? super E>) pq.comparator();screen = false;if (pq.getClass() == PriorityBlockingQueue.class) // exact matchheapify = false;}Object[] a = c.toArray();int n = a.length;// If c.toArray incorrectly doesn't return Object[], copy it.if (a.getClass() != Object[].class)a = Arrays.copyOf(a, n, Object[].class);if (screen && (n == 1 || this.comparator != null)) {for (int i = 0; i < n; ++i)if (a[i] == null)throw new NullPointerException();}this.queue = a;this.size = n;if (heapify)heapify();}

package cn.qlq.thread.thirteen;import java.util.concurrent.BlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;public class Demo4 {public static void main(String[] args) throws InterruptedException {BlockingQueue<Person> persons = new PriorityBlockingQueue<Person>(3);persons.put(new Person(20,"張三"));persons.put(new Person(22,"李四"));persons.put(new Person(21,"王五"));persons.put(new Person(18,"八卦"));System.out.println(persons.take());System.out.println(persons.take());System.out.println(persons.take());System.out.println(persons.take());}
}class Person implements Comparable<Person>{private int age;private String name;public int getAge() {return age;}public void setAge(int age) {this.age = age;}@Overridepublic String toString() {return "Person [age=" + age + ", name=" + name + "]";}public String getName() {return name;}public void setName(String name) {this.name = name;}public Person(int age, String name) {super();this.age = age;this.name = name;}@Overridepublic int compareTo(Person o) {//返回-1表示排在他前面，返回1表示排在他后面if(o.getAge() > this.getAge()  ){return 1;}else if(o.getAge() < this.getAge()){return -1;}return 0;}
}

    /** The queued items */final Object[] items;

    /*** Linked list node class*/static class Node<E> {E item;Node<E> next;Node(E x) { item = x; }}

private transient Object[] queue;

    public SynchronousQueue() {this(false);}public SynchronousQueue(boolean fair) {transferer = fair ? new TransferQueue() : new TransferStack();}

    abstract static class Transferer {/*** Performs a put or take.**/abstract Object transfer(Object e, boolean timed, long nanos);}

package cn.qlq.thread.thirteen;import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;public class Demo5 {private static final Logger LOGGER = LoggerFactory.getLogger(Demo5.class);public static void main(String[] args) throws InterruptedException {BlockingQueue<String> persons = new SynchronousQueue<String>();persons.put("1");LOGGER.info("放入元素 1");LOGGER.info("獲取元素 "+persons.take());}
}

package cn.qlq.thread.thirteen;import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;public class Demo5 {private static final Logger LOGGER = LoggerFactory.getLogger(Demo5.class);public static void main(String[] args) throws InterruptedException {final BlockingQueue<String> strings = new SynchronousQueue<String>();Thread consumer = new Thread(new Runnable() {@Overridepublic void run() {try {String take = strings.take();LOGGER.info("ThreadName ->{} take ele->{}",Thread.currentThread().getName(),take);} catch (InterruptedException e) {e.printStackTrace();}}},"consumer");consumer.start();strings.put("1");LOGGER.info("放入元素 1");}
}

public interface Delayed extends Comparable<Delayed> {
long getDelay(TimeUnit unit);
}

    private transient final ReentrantLock lock = new ReentrantLock();private final PriorityQueue<E> q = new PriorityQueue<E>();
private Thread leader = null;public DelayQueue() {}
public DelayQueue(Collection<? extends E> c) {this.addAll(c);}public E take() throws InterruptedException {final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {for (;;) {E first = q.peek();if (first == null)available.await();else {long delay = first.getDelay(TimeUnit.NANOSECONDS);if (delay <= 0)return q.poll();else if (leader != null)available.await();else {Thread thisThread = Thread.currentThread();leader = thisThread;try {available.awaitNanos(delay);} finally {if (leader == thisThread)leader = null;}}}}} finally {if (leader == null && q.peek() != null)available.signal();lock.unlock();}}

package cn.qlq.thread.thirteen;import java.util.Date;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;public class Demo6 {private static final Logger LOGGER = LoggerFactory.getLogger(Demo6.class);public static void main(String[] args) throws InterruptedException {final BlockingQueue<DelayObj> delayObjs = new DelayQueue<DelayObj>();DelayObj delayObj = new DelayObj("1");delayObjs.put(delayObj);LOGGER.info("放入元素->{}", delayObj);Thread.sleep(1 * 1000);DelayObj delayObj2 = new DelayObj("3");delayObjs.put(delayObj2);LOGGER.info("放入元素->{}", delayObj2);LOGGER.info("{}", delayObjs.take());LOGGER.info("{}", delayObjs.take());}
}class DelayObj implements Delayed {private Date createTime;private String name;public Date getCreateTime() {return createTime;}public void setCreateTime(Date createTime) {this.createTime = createTime;}public DelayObj(String name) {this.createTime = new Date();this.name = name;}public String getName() {return name;}public void setName(String name) {this.name = name;}@Overridepublic int compareTo(Delayed o) { // 返回负数表示在前面，返回正数表示在后面if (this.getDelay(TimeUnit.NANOSECONDS) > o.getDelay(TimeUnit.NANOSECONDS)) {// NANOSECONDS是十亿分之秒return -1;} else if (this.getDelay(TimeUnit.NANOSECONDS) < o.getDelay(TimeUnit.NANOSECONDS)) {return 1;}return 0;}@Overridepublic String toString() {return "DelayObj [createTime=" + createTime + ", name=" + name + "]";}@Overridepublic long getDelay(TimeUnit unit) {Date now = new Date();long diff = createTime.getTime() + 5 * 1000 - now.getTime();System.out.println(diff);return unit.convert(diff, TimeUnit.MILLISECONDS);}
}