l.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; public class MyBlockingQueue extends Thread { public static BlockingQueue queue = new LinkedBlockingQueue (3); private int index; public MyBlockingQueue(int i) { this.index = i; } public void run() { try { queue.put(String.valueOf(this.index)); System.out.println("{" + this.index + "} in queue!"); } catch (Exception e) { e.printStackTrace(); } } public static void main(String args[]) { ExecutorService service = Executors.newCachedThreadPool(); for (int i = 0; i < 10; i++) { service.submit(new MyBlockingQueue(i)); } Thread thread = new Thread() { public void run() { try { while (true) { Thread.sleep((int) (Math.random() * 1000)); if (MyBlockingQueue.queue.isEmpty()) break; String str = MyBlockingQueue.queue.take(); System.out.println(str + " has take!"); } } catch (Exception e) { e.printStackTrace(); } } }; service.submit(thread); service.shutdown(); } }
---------------------执行结果-----------------
{0} in queue!
{1} in queue!
{2} in queue!
{3} in queue!
0 has take!
{4} in queue!
1 has take!
{6} in queue!
2 has take!
{7} in queue!
3 has take!
{8} in queue!
4 has take!
{5} in queue!
6 has take!
{9} in queue!
7 has take!
8 has take!

5 has take!
9 has take!
-----------------------------------------

CompletionService

我们现在在Java中使用多线程通常不会直接用Thread对象了，而是会用到java.util.concurrent包下的ExecutorService类来初始化一个线程池供我们使用。之前我一直习惯自己维护一个list保存submit的callable task所返回的Future对象。在主线程中遍历这个list并调用Future的get()方法取到Task的返回值。但是，我在很多地方会看到一些代码通过CompletionService包装ExecutorService，然后调用其take()方法去取Future对象。以前没研究过这两者之间的区别。今天看了源代码之后就明白了。这两者最主要的区别在于submit的task不一定是按照加入自己维护的list顺序完成的。
从list中遍历的每个Future对象并不一定处于完成状态，这时调用get()方法就会被阻塞住，如果系统是设计成每个线程完成后就能根据其结果继续做后面的事，这样对于处于list后面的但是先完成的线程就会增加了额外的等待时间。而CompletionService的实现是维护一个保存Future对象的BlockingQueue。只有当这个Future对象状态是结束的时候，才会加入到这个Queue中，take()方法其实就是Producer-Consumer中的Consumer。它会从Queue中取出Future对象，如果Queue是空的，就会阻塞在那里，直到有完成的Future对象加入到Queue中。
所以，先完成的必定先被取出。这样就减少了不必要的等待时间。ExecutorCompletionService 类提供了此方法的一个实现。

import java.util.concurrent.Callable;

import java.util.concurrent.CompletionService;

import java.util.concurrent.ExecutorCompletionService;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

public class MyCompletionService implements Callable
  
    {
	private int id;
	public MyCompletionService(int i) {
		this.id = i;
	}

	public static void main(String[] args) throws Exception {
		ExecutorService service = Executors.newCachedThreadPool();
		CompletionService
   
     completion = new ExecutorCompletionService
    
     (service); for (int i = 0; i < 10; i++) { completion.submit(new MyCompletionService(i)); } for (int i = 0; i < 10; i++) { System.out.println(completion.take().get()); } service.shutdown(); } public String call() throws Exception { Integer time = (int) (Math.random() * 1000); try { System.out.println(this.id + " start"); Thread.sleep(time); System.out.println(this.id + " end"); }catch (Exception e) { e.printStackTrace(); } return this.id + ":" + time; } }

CountDownLatch

一个同步辅助类，在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待。
用给定的计数初始化 CountDownLatch。由于调用了 countDown() 方法，所以在当前计数到达零之前，await 方法会一直受阻塞。
之后，会释放所有等待的线程，await 的所有后续调用都将立即返回。这种现象只出现一次――计数无法被重置。如果需要重置计数，请考虑使用 CyclicBarrier。
CountDownLatch 是一个通用同步工具，它有很多用途。将计数 1 初始化的 CountDownLatch 用作一个简单的开/关锁存器，
或入口：在通过调用 countDown() 的线程打开入口前，所有调用 await 的线程都一直在入口处等待。
用 N 初始化的 CountDownLatch 可以使一个线程在 N 个线程完成某项操作之前一直等待，或者使其在某项操作完成 N 次之前一直等待。
CountDownLatch 的一个有用特性是，它不要求调用 countDown 方法的线程等到计数到达零时才继续，而在所有线程都能通过之前，它只是阻止任何线程继续通过一个 await。

下面的例子是别人写的，非常形象。

import java.util.concurrent.CountDownLatch;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

public class TestCountDownLatch {

	public static void main(String[] args) throws InterruptedException {
		// 开始的倒数锁
		final CountDownLatch begin = new CountDownLatch(1);
		// 结束的倒数锁
		final CountDownLatch end = new CountDownLatch(10);
		// 十名选手
		final ExecutorService exec = Executors.ne

Java中的并发库学习总结(三)

CompletionService

CountDownLatch