java并发编程(五)
PriorityBlockingQueue，DelayQueue
阻塞队列：在某些情况下，会挂起线程，一旦条件满足，被挂起的线程会自动唤醒。而阻塞队列无需关心什么时候阻塞，什么时候唤醒。
ArrayBlockingQueue：由数组结构组成的有界阻塞队列
LinkedBlockingQueue：由链表结构组成的有界阻塞队列
SynchronousQueue:不存储元素的阻塞队列，即单个元素队列
PriorityBlockingQueue：支持优先级排序的无界阻塞队列
DelayQueue：使用优先级队列实现的延迟无界阻塞队列
阻塞队列的通用Api
抛异常 特殊值 阻塞 超时
插入add(e) offer(e) put(e) offer(e,time,unit)
移除 remove() poll() take() poll(time,unit)
检查 element() peek(） X X
demo
public class BlockingQueueDemo {
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 (InterruptedException e) {
e.printStackTrace();
}
}, "A").start();
new Thread(() -> {
try {
TimeUnit.SECONDS.sleep(5);
System.out.println(Thread.currentThread().getName()+"\t get"+blockingQueue.take());
TimeUnit.SECONDS.sleep(5);
System.out.println(Thread.currentThread().getName()+"\t get"+blockingQueue.take());
TimeUnit.SECONDS.sleep(5);
System.out.println(Thread.currentThread().getName()+"\t get"+blockingQueue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}, "B").start();
}
}SynchronousQueue:不存储元素的阻塞队列，即单个元素队列，每次put进一个元素，另一个线程5秒后取值，每次take到的都是put进的值
Connected to the target VM, address: '127.0.0.1:53549', transport: 'socket'
A put 1
B get1
A put 2
B get2
A put 3
B get3
Disconnected from the target VM, address: '127.0.0.1:53549', transport: 'socket'当队列满时，你还add，就会抛异常，队列满了
当队列空时，你还remove，就会抛异常，没有元素给你
//ArrayBlockingQueue 容量仅为2 放入3个 B能取到吗？能取到几个？
public class BlockingQueueDemo {
public static void main(String[] args) {
//BlockingQueue<String> blockingQueue = new SynchronousQueue<>();
ArrayBlockingQueue<String> blockingQueue = new ArrayBlockingQueue<String>(2);
new Thread(() -> {
try {
System.out.println(Thread.currentThread().getName() + "\t put 1");
blockingQueue.add("1");
System.out.println(Thread.currentThread().getName() + "\t put 2");
blockingQueue.add("2");
System.out.println(Thread.currentThread().getName() + "\t put 3");
blockingQueue.add("3");
} catch (Exception e) {
e.printStackTrace();
}
}, "A").start();
new Thread(() -> {
try {
TimeUnit.SECONDS.sleep(5);
System.out.println(Thread.currentThread().getName()+"\t get"+blockingQueue.poll());
TimeUnit.SECONDS.sleep(5);
System.out.println(Thread.currentThread().getName()+"\t get"+blockingQueue.poll());
TimeUnit.SECONDS.sleep(5);
System.out.println(Thread.currentThread().getName()+"\t get"+blockingQueue.poll());
} catch (InterruptedException e) {
e.printStackTrace();
}
}, "B").start();
}
}
A put 1
A put 2
A put 3
java.lang.IllegalStateException: Queue full
at java.util.AbstractQueue.add(AbstractQueue.java:98)
at java.util.concurrent.ArrayBlockingQueue.add(ArrayBlockingQueue.java:312)
at com.kk.thread.BlockingQueueDemo.lambda$main$0(BlockingQueueDemo.java:25)
at java.lang.Thread.run(Thread.java:748)
B get1
B get2
B getnull问题:那么当ArrayBlockingQueue用put放入元素时会报full queu 异常吗？为什么？
将Add换成put，虽然限制容量为2，但是继续放元素，队列会阻塞直到put数据，B仍能取到所有元素
看下ArraBlockingQueue的put源码
/**
* Inserts the specified element at the tail of this queue, waiting
* for space to become available if the queue is full.
*在队列尾部插入指定元素，等待队列已满时空间变为可用
* @throws InterruptedException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
*/
public void put(E e) throws InterruptedException {
checkNotNull(e);//检查是否元素为null
final ReentrantLock lock = this.lock;//可重入锁
lock.lockInterruptibly();//实际判断的是Thread.interrupted()
try {
while (count == items.length)队列元素满了就等待
notFull.await();
enqueue(e);否则进入队列，调用singal方法通知线程
} finally {
lock.unlock();
}
}放入1
检查线程是否中断
设置线程状态
放入队列
此时调用unlock waitstatus=-1
而unlock最终会unpark当前线程
如果线程被阻塞*{@code park}然后它将解除锁定。保证下一个调用
到{@code park}不会阻塞
也就是说put方法在队列满了以后会阻塞，等待没满的唤醒
31:CountDownLatch,CyclicBarrier,Semaphore
CountDownLatch:让一些线程阻塞直到另一些线程完成一系列操作后才被唤醒
public class CountDownLatchDemo
{
public static void main(String[] args) throws InterruptedException
{
CountDownLatch latch = new CountDownLatch(6);
for (int i = 1; i <= 6; i++) {
new Thread(() -> {
System.out.println(Thread.currentThread().getName()+"\t 车次离开");
latch.countDown();
}, String.valueOf(i)).start();
}
latch.await();
System.out.println(Thread.currentThread().getName()+"\t 锁定");
}
}
1 车次离开
2 车次离开
3 车次离开
4 车次离开
5 车次离开
6 车次离开
main 锁定CyclicBarrier：让一组线程到达一个屏障（也可以叫同步点）时被阻塞，
直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才
会继续干活。
public class CyclicBarrierDemo
{
private static final int NUMBER = 7;
public static void main(String[] args)
{
//CyclicBarrier(int parties, Runnable barrierAction)
CyclicBarrier cb = new CyclicBarrier(NUMBER, new Runnable() {
@Override
public void run() {
System.out.println("车次"+NUMBER+"已满");
}
});
for (int i = 1; i <=7; i++) {
new Thread(() -> {
try {
System.out.println(Thread.currentThread().getName()+"\t 车次到达");
cb.await();
} catch (InterruptedException | BrokenBarrierException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}, String.valueOf(i)).start();
}
}
1 车次到达
2 车次到达
3 车次到达
4 车次到达
5 车次到达
6 车次到达
7 车次到达
车次7已满Semaphore:一个是用于多个共享资源的互斥使用，另一个用于并发线程数的控制，互斥，当一个线程使用资源时获取信号量，互斥线程需要等待，线程释放，将信号量加1，唤醒互斥资源
public class SemaphoreDemo
{
public static void main(String[] args)
{
//3个停车位
Semaphore semaphore = new Semaphore(3);
//模拟6部汽车
for (int i = 1; i <= 6; i++) {
new Thread(() -> {
try {
semaphore.acquire();
System.out.println(Thread.currentThread().getName()+"\t 号停入停车位");
TimeUnit.SECONDS.sleep(3);
semaphore.release();
System.out.println(Thread.currentThread().getName()+"\t 号离开停车位");
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}, String.valueOf(i)).start();
}
}
}1 号停入停车位
2 号停入停车位
3 号停入停车位
4 号停入停车位
2 号离开停车位
3 号离开停车位
5 号停入停车位
6 号停入停车位
1 号离开停车位
4 号离开停车位
5 号离开停车位
6 号离开停车位
32:Exchanger交换器
一个同步点，在这个点上线程可以成对地交换元素。每个线程在{@link#exchange}方法的条目上显示一些对象，与伙伴线程匹配，并在返回时接收其伙伴的对象。交换器可以看作{@link SynchronousQueue}的双向形式。Exchangers may be useful in applications such as genetic algorithms and pipeline designs，应用于遗传算法和管道设计
A synchronization point at which threads can pair and swap elements
* within pairs. Each thread presents some object on entry to the
* {@link #exchange exchange} method, matches with a partner thread,
* and receives its partner's object on return. An Exchanger may be
* viewed as a bidirectional form of a {@link SynchronousQueue}.
* Exchangers may be useful in applications such as genetic algorithms
* and pipeline designs.用于两个线程之间的数据交换
public class ExchangerDemo {
public static void main(String[] args) {
ExecutorService executor = Executors.newCachedThreadPool();
final Exchanger exchanger = new Exchanger();
executor.execute(()->{
String data1="int 1";
exchangeDate(data1, exchanger);
});
executor.execute(()->{
String data1="int 2";
exchangeDate(data1, exchanger);
});
executor.shutdown();
}
private static void exchangeDate(String data1, Exchanger exchanger) {
try {
System.out.println(Thread.currentThread().getName() + "交换数据" + data1 + " 交换");
Thread.sleep(5L);
String data2 = (String) exchanger.exchange(data1);
System.out.println(Thread.currentThread().getName() + "交易得到" + data2);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
pool-1-thread-1交换数据int 1 交换
pool-1-thread-2交换数据int 2 交换
pool-1-thread-1交易得到int 2
pool-1-thread-2交易得到int 1