标签：并发
委托时创建线程安全类的一个最有效的策略，只需让现有的线程安全类管理所有的状态即可。
平台类库中包含了一个并发构建块的丰富集合，如线程安全的容器与同步工具。
分两部分，一是JDK1.0的Vector与Hashtable，另一个是JDK1.2才被加入的同步包装类Collections.synchronizedXxx工厂方法创建的。Collections.synchronizedXxx工厂方法构造出的容器返回的List与Set的iterator()与listIterator()（List集合）没有使用同步。
这些类实现线程安全的方式是：将它们的状态封装起来，并对每个公有方法都进行同步，使得每次只有一个线程能访问容器的状态。
同步容器类都是线程安全的，但在某些情况下可能需要额外的客户端加锁来保护复合操作。容器上常见的复合操作包括：迭代（反复访问元素，直到遍历完容器中的所有元素）。跳转（根据指定顺序找到当前元素的下一个元素）以及条件运算，例如：”若没有则添加“（检查在Map中是否存在键值K，如果没有，就加入二元组（K，V））。在同步容器类中，这些复合操作在没有客户端加锁的情况下仍然是线程安全的，但当其他线程并发地修改容器时，它们可能会出现意料之外的行为。
操作Vector（同步容器）的复合操作可能导致混乱的结果：
public static Object getLast(Vector list) {
int lastIndex = list.size() - 1;
return list.get(lastIndex);
}在多线程的环境下可能会抛出ArrayIndexOutOfBoundsException，因为其他线程可能会在size()与get中修改Vector，但单线程下是不会有问题的。
由于同步容器要遵守同步策略，即支持客户端加锁，因此可能会创建一些新的操作，只要我们知道应该使用哪一个锁，那么这些新操作就与容器的其他操作一样都是原子操作。同步容器类通过其自身的锁来保护它的每个方法。使用客户端加锁，对Vector进行复合操作：
public static Object getLast(Vector list) {
synchronized (list) {
int lastIndex = list.size() - 1;
return list.get(lastIndex);
}
}在调用size和相应的get之间，Vector的长度可能会发生变化，这种风险在对Vector中的元素进行迭代时仍然会出现，多线程环境下迭代过程中也可能抛出ArrayIndexOutOfBoundsException：
for (int i = 0; i < vector.size(); i++)
doSomething(vector.get(i));迭代过程中可能抛出异常，但并不意味着Vector就不是线程安全。Vector的状态仍然是有效的，事实上异常恰好使它保持规范的一致性。然而，在正常或迭代读过程中抛出异常的确不是人们所期望的。
造成迭代不可靠的问题同样可以通过在客户端加锁来完成，这要增加一些开销，像下那样，通过在迭代期间持有Vector的锁，我们防止其他线程在迭代期间修改Vector，这样完全阻止了其他线程在这期间访问它，如果集合很大或者对每个元素执行的任务耗时比较长，这会削弱并发性。
synchronized (vector) {
for (int i = 0; i < vector.size(); i++)
doSomething(vector.get(i));//还要持有另一个锁，这是一个产生死锁风险的因素
}尽管上面讨论的Vector是“遗留”下来的容器类，这只是说明同步容器有这样的问题。其实，“现代”的容器类也并没有消除复合操作产生的问题，比如迭代复合操作，当其他线程并发修改容器时，使用迭代器仍然避免不了在使用的地方加锁，在设计同步容器返回迭代器时，并没有使用同步（注，这里讲的是说返回的迭代器不是线程安全，而不是指返回迭代器的方法iterator() 没有使用同步，它本身就是经过同步了的。），因为他们是“及时失败”——只要有其他线程修改容器结果，立马就会抛出未检查性异常ConcurrentModificationException。
这种”及时失败“的迭代器并不是一种完备的处理机制，而只是”善意地“捕获并发错误，因此只能作为并发问题的预警指示器。它们采用的实现方式是，将计算器的变化与容器关联起来：如果在迭代期间计数器被修改，那么hasNext或next将抛出ConcurrentModificationException。然而，这种检查是在没有同步的情况下进行的，因此可能会看到失效的计数值，而迭代器可能并没有意识到已经发生了修改。这是设计上得权衡，从而降低并发修改并发操作的检测代码对程序性能带来的影响。
注：ConcurrentModificationException也可能出现在单线程的代码中，如果对象不是调用Iterator.remove，而是直接从容器中删除就会出现这种情况。
1.5中的for-each循环语法对容器进行迭代时，也是隐式地用到了Iterator，从内部来看，javac将生成使用Iterator的代码，反复调用hasNext和next来迭代List对象，与迭代Vector一样，想要避免ConcurrentModificationException，就必须在迭代过程持有容器的锁：
List<Widget> widgetList = Collections.synchronizedList(new ArrayList<Widget>());
...
// May throw ConcurrentModificationException
for (Widget w : widgetList)
doSomething(w);有时候开发人员并不希望在迭代期间对容器加锁，例如，某些线程在可以访问容器之前，必须等待迭代过程结束，如果容器规模很大，或者在每个元素上执行操作的时间很长，那么这些线程将长时间等待。即使不存在饥饿或者死锁等风险，长时间地对容器加锁也会降低程序的可伸缩性。持有锁的时间越长，那么在锁上的竞争就可能越激烈，如果许多线程在等待锁被释放，那么将极大地降低吞吐量和CPU的利用率。
如果不希望在迭代期间对容器加锁，那么一种替代方法就是”克隆“容器，并在副本上进行迭代。由于副本被封闭在线程内，因此其他线程不会在迭代期间对其进行修改，这样就避免了抛出ConcurrentModificationException（在克隆过程中仍然需要对容器加锁）。在克隆容器时存在显著地性能开销。这种方式的好坏取决于多个因素，包括容器的大小，在每个元素上执行的工作，迭代操作相对于容器其他操作的调用频率，以及在响应时间和吞吐量等方面的需求。
在一个可能发生迭代的共享容器中，各处都需要锁，这是一个棘手的问题，因为迭代器有时是隐藏的，就像下面代码一样，容器的toString方法的实现是通过迭代容器中的每个元素。编译器将字符串的连接操作转换为调用StringBuilder.append(Object),而这个方法又会调用容器的toString方法，标准容器的toString方法将迭代容器，并在每个元素上调用toString来生成容器内容的格式化表示。
public class HiddenIterator {
private final Set<Integer> set = new HashSet<Integer>();
public synchronized void add(Integer i) { set.add(i); }
public void addTenThings() {
Random r = new Random();
for (int i = 0; i < 10; i++)
add(r.nextInt());
System.out.println("DEBUG: added ten elements to " + set);//这里会隐式地使用迭代
}
}toString对容器进行迭代。当然真正的问题是HiddenIterator不是线程安全的。在使用println的set之前必须首先获取HiddenIterator的锁，但在调试代码和日志代码中通常会忽视这个要求。
如果状态与保护它的同步代码之间相隔越远，那么开发人员就越容易忘记在访问状态时使用正确的同步。如果将HashSet包装为synchronizedSet，并且对同步代码进行封装，就不会出现ConcurrentModificationException异常了。
正如封装对象的状态有助于维持不变性条件一样，封装对象的同步机制同样有助于确保实施同步策略。
容器的hashCode和equals等方法也会间接地执行迭代操作，当容器作为另一个容器的元素或键值时，就会出现这种情况。同样，containsAll、removeAll和retainAll等方法，以及把容器作为参数的构造函数，都会对容器进行迭代。所有这些间接地迭代操作都可能抛出ConcurrentModificationException。
1.5提供了几个并发的容器类来改进同步容器。同步容器通过对容器的进行串行访问，从而实现了它们的线程安全。这样做虽然是绝对的安全，但代价是削弱了并发性，当多个线程共同竞争容器级的锁时，吞吐量会降低。
并发容器是为多线程并发访问而设计的。1.5添加了ConcurrentHashMap，来替代同步的哈希Map实现；当大多数的操作是读操作时（因为如果有很多写操作会引起内部对原来集合进行复制，从而带来开销），CopyOnWriteArrayList是List相应的同步实现，同样CopyOnWriteArraySet是Set相应的同步实现（内部是以CopyOnWriteArrayList来实现的）。并且在ConcurrentMap接口还加入了对常见复合操作的支持，如“缺少即加入 put-if-absent”、替换和条件删除。
通过并发容器来代替同步容器，可以极大地提高伸缩性并降低风险。
1.5同时增加了两个新的容器类型：Queue和BlockingQueue（Queue接口继承了Collection接口）。有几种实现，一个传统意义上（入队与出队不会被阻塞，是相对阻塞队列来说的）的FIFO队列ConcurrentLinkedQueue，底层是基于链表结构；一个是有优先级顺序的队列PriorityQueue（注，它不支持非并发）。Queue的操作不会阻塞，如果队列是空，那么从队列中获取时返回null。尽管可以使用List来模拟Queue的类——事实上，LinkedList就已实现了Queue（如果我们只需要一个单纯的或者是传统意义上的队列时，我们应该使用LinkedList，如果我们需要在并发环境下，则使用ConcurrentLinkedQueue来代替它）——但我们还是需要Queue的类，因为如果忽略掉List的随机访问需求的话，使用Queue能得到高效的并发实现。
Queue接口：
BlockingQueue接口：
BlockingQueue扩展了Queue，增加了可阻塞的插入（put）和获取操作（take）。如果队列为空，则take阻塞；如果队列满（对于有限队列：LinkedBlockingQueue—可以不指定，不指定时容量为最大的Integer.MAX_VALUE、ArrayBlockingQueue构造时则一定要指定大小），put操作会阻塞直到有空间，而对于无界队列（PriorityBlockingQueue、DelayQueue），放入时不会被阻塞，直到OutOfMemoryError。阻塞队列在生产者——消费者设计中非常有用。
正如ConcurrentHashMap作为同步的哈希Map的一个替代，1.6加入了ConcurrentSkipListMap和ConcurrentSkipListSet，用来作为同步的SortedMap和SortedSet的并发替代品（用synchronizedMap包装的TreeMap或TreeSet）。
同步容器类在每个操作的执行期间都持有一个锁。比如HashMap.get或者List.contains操作，可能包含大量的工作：当遍历散列桶或列表来查找某个特定对象时，必须在许多元素上调用equals（而equals本身还包含一定的计算量）。在基于散列的容器中，如果hashCode不能很均匀地分布散列值，那么容器中的元素就不会均匀地分布在整个容器中。再调用它们的过程中可能需要很长一段时间，并且在这段时间内，其他线程都不能访问这个容器。
ConcurrentHashMap和HashMap一样是一个哈希表，但是它使用完全不同的锁策略，可以提供更好的并发性和或伸缩性。以前的同步容器在内部只有一把锁，即容器自身，而ConcurrentHashMap使用一个更加细化的锁机制，名叫“锁分离或分段锁”。这种机制允许更深层次的共享访问。任意数量的读线程可以并发访问Map，读和写线程可以并发访问Map，并且有限数量的写线程还可以并发修改Map。这样为并发访问带来了更高的吞吐量，同时几乎没有损失单个线程访问的性能。
ConcurrentHashMap提供了不会抛出ConcurrentModificationException异常的迭代器，因此不需要在容器迭代时加锁访问，它所返回的迭代器是弱一致性的，而非“及时失败”的。弱一致性的迭代可以允许并发修改，当迭代器被创建时，它会遍历已有的元素，并且可以（但是不保证）在迭代器被构造后将修改操作反映给容器。
尽管有这么多改进，但有一些还是需要权衡的地方。那些对整个Map进行操作的方法，如size和isEmpty，它们的语义在反映容器并发特性被弱化了。因为size的结果相对于在计算的时候可能已经过期，它仅仅只是一个估算值，所以允许size返回一个近似值而不是一个精确的值。这在一开始会让人有些困扰，不过事实上像size和isEmpty这样的方法在并发环境下几乎没有什么用处，因为它们的返回值总在不断变化，它们的目标是在于并发的读与写，所以这些操作的原子性被弱化了。相反，应该保证对最重要的操作进行性能优化，最重要的是get、put、containsKey和remove等。
相比于Hashtable和synchronizedMap，ConcurrentHashMap有很多的优势，因此大多数情况下ConcurrentHashMap取代同步Map实现只会带来更好的可伸缩性。只有当应用程序需要加锁Map以进行独占访问时，才应该放弃使用ConcurrentHashMap。
因为ConcurrentHashMap不能被独占访问，所以我们不能在客户端加锁来创建新的原子操作，比如我在前面对Vector复合操作施加的原子性。不过一些常的复合操作，如“若没有则添加”、“若相等则移除”和“若相等则替换”等都已被实现为原子操作。如果你正在已同步Map中加入这些功能时，你可能考虑使用ConcurrentHashMap来替代同步的Map。
public interface ConcurrentMap<K,V> extends Map<K,V>
{
// 仅当K没有响应映射值时才插入
public V putIfAbsent(K key, V value);
// 仅当K被映射到V时才移除
public boolean remove(Object key, Object value);
// 仅当K被映射到oldValue时才替换为newValue
public boolean replace(K key, V oldValue, V newValue);
// 仅当K被映射到某个值时才替换为newValue
public V replace(K key, V value);
}CopyOnWriteArrayList用于替代同步List，在某些情况下它提供了更好地并发性能，并且在迭代期间不需要对容器进行加锁或复制。（类似地，CopyOnWriteArraySet的作用时替代同步Set，它是对CopyOnWriteArrayList包装，所有的操作都是转换给CopyOnWriteArrayList，与CopyOnWriteArrayList没什么区别）
“写入时复制（Copy-On-Write）”容器的线程安全性在于，只要正确地发布一个事实不可变的对象，那么在访问该对象时就不再需要进一步的同步。在每次修改时，都会创建并重新发布一个新的容器副本，从而实现可变性。“写入时复制”容器的迭代器保留一个指向底层基数数组的引用，这个数组当前位于迭代器的起始位置，由于它不会被修改，因此在对其进行同步时只需确保数组内容的可见性。因此，多个线程可以同时对这个容器进行迭代，而不会彼此干扰或者与修改容器的线程相互干扰。“写入时复制”容器返回的迭代器不会抛出ConcurrentModificationException，并且返回的元素与迭代器创建时的元素完全一致，而不必考虑之后修改操作所带来的影响。
CopyOnWriteArrayList<StringBuffer> cwa = new CopyOnWriteArrayList<StringBuffer>();
cwa.add(new StringBuffer("0"));
Iterator<StringBuffer> it = cwa.iterator();
cwa.get(0).append("1");
cwa.add(new StringBuffer("3"));
while (it.hasNext()) {
// 不会抛异常，那怕在迭代创建后修改了结构，并只输出 01
System.out.println(it.next());
}显然，每当修改容器时都会复制底层数组，这需要一定的开销，特别是当容器的规模较大时，仅当迭代操作远远多于修改操作时，才应该使用“写入时复制”容器。这个准则很好地描述了许多时间通知系统：在分发通知时需要迭代已注册监听器链表，并调用每一个监听器，在大多数情况下，注册和注销事件监听器的操作远少于接受事件通知的操作。
Queue继承体系结构：
队列是一种数据结构，它有两个基本操作：在队列尾部加人一个元素，和从队列头部移除一个元素就是说，队列以一种先进先出的方式管理数据，如果你试图向一个已经满了的阻塞队列中添加一个元素或者是从一个空的阻塞队列中移除一个元索，将导致线程阻塞．在多线程进行合作时，阻塞队列是很有用的工具。工作者线程可以定期地把中间结果存到阻塞队列中而其他工作者线线程把中间结果取出并在将来修改它们。队列会自动平衡负载。如果第一个线程集运行得比第二个慢，则第二个线程集在等待结果时就会阻塞。如果第一个线程集运行得快，那么它将等待第二个线程集赶上来。下表显示了jdk1.5中的阻塞队列的操作：
add增加一个元素如果队列已满，则抛出一个IIIegaISlabEepeplian异常
remove移除并返回队列头部的元素如果队列为空，则抛出一个NoSuchElementException异常
element返回队列头部的元素如果队列为空，则抛出一个NoSuchElementException异常
offer添加一个元素并返回true如果队列已满，则返回false
poll移除并返问队列头部的元素如果队列为空，则返回null
peek返回队列头部的元素如果队列为空，则返回null
put添加一个元素如果队列满，则阻塞
take移除并返回队列头部的元素如果队列为空，则阻塞remove、element、offer 、poll、peek 其实是属于Queue接口，这些方法都不会阻塞。
队列Queue接口与List、Set同一级别，都是继承了Collection接口。LinkedList现已经实现了Queue接口。Queue接口窄化了对LinkedList的方法的访问权限（即在方法中的参数类型如果是Queue时，就完全只能访问Queue接口所定义的方法了，而不能直接访问LinkedList的非Queue的方法），以使得只有恰当的方法才可以使用。BlockingQueue 继承自Queue接口。
阻塞队列的操作可以根据它们的响应方式分为以下三类：aad、removee和element操作在你试图为一个已满的队列增加元素或从空队列取得元素时抛出异常。当然，在多线程程序中，队列在任何时间都可能变成满的或空的，所以你可能想使用offer、poll、peek方法。这些方法在无法完成任务时只是给出一个出错示而不会抛出异常。
注意：poll和peek方法出错进返回null。因此，向队列中插入null值是不合法的。
还有带超时的offer和poll方法变种，例如，下面的调用：
boolean success = q.offer(x,100,TimeUnit.MILLISECONDS);
尝试在100毫秒内向队列尾部插入一个元素。如果成功，立即返回true；否则，当到达超时进，返回false。同样地，调用：
Object head = q.poll(100, TimeUnit.MILLISECONDS);
如果在100毫秒内成功地移除了队列头元素，则立即返回头元素；否则在到达超时时，返回null。
阻塞队列提供了可阻塞的put和take方法，以及支持定时的offer和poll方法。如果队列已经满了，那么put方法将阻塞直到有空间可用：如果队列为空，那么take方法将会阻塞直到有元素可用。队列可以使有界的也可以是无界的，无界队列永远不会充满，因此无界队列上的put方法也永远不会阻塞。
阻塞队列支持生产者-消费者这种设计模式。该模式将“找出需要完成的工作”与“执行工作”这两个过程分离开来，并把工作项放入一个“待完成”列表中以便在随后处理，而不是找出后立即处理。生产者-消费者模式能简化开发过程，因此它消除了生产者类和消费者类之间的代码依赖性，此外，该模式还将生产数据的过程与消费数据的过程解耦开来以简化工作负载的管理。因为这两个过程在处理数据的速率上有所不同。
在基于阻塞队列构建的生产者-消费者设计中，当数据生成时，生产者把数据放入队列，而当消费者准备处理数据时，将从队列中获取数据。生产者不需要知道消费者的标示或数量，或者它们是否是唯一的生产者，而只需将数据放入队列即可。同样，消费者也不需要知道生产者是谁，或者工作来自何处。Blockingqueue简化了生产者-消费者设计的实现过程，它支持任意数量的生产者和消费者。
最常见的生产者-消费者设计是将线程池与工作队列相结合起来（据我所知，Executor框架有以下地方用到了池：池中的线程就是放在队列中的，但Executor的任务提交后不是会放入队列中，而是立刻准备执行；ScheduledExecutorService定制的计划任务会放入工作队列中，等到延迟到达后执行；另外CompletionService处理完后的结果会放在队列中），讲述Executor任务执行框架时会具体介绍这个模式。
“生产者”和“消费者”的角色是相对的，某种环境中的消费者在另一种不同的环境中可能会成为生产者。
阻塞队列简化了消费者程序的编码，因为take操作会一直阻塞知道有可用的数据。如果生产者不能尽快地产生工作项能使消费者保持忙碌，那么消费者就只能一直等待，直到有工作可做。在某些情况下，这种方式是非常合适的（例如，在服务器应用程序中，没有任何客户请求服务），而在其他一些情况下，这也表示需要调整生产者线程数量和消费者线程数量之间的比率，从而实现更高的资源利用率（例如，在“网页爬虫”或其他应用程序中，有无穷的工作需要完成）。
如果生产者生成工作的速率比消费者处理工作的速率快，那么工作项会在队列中累积起来，最终耗尽内存。同样，put方法的阻塞特性也极大地简化了生产者的编码。如果使用有界队列，那么当队列充满时，生产者将阻塞并且不能继续生成工作，而消费者就有时间来赶上工作处理进度。
阻塞队列同样提供了一个offer方法。如果数据项不能被添加到队列中，那么将返回一个失败状态。这样你就能够创建更多灵活的策略来处理负荷过载的情况，例如减轻负载，将多余的工作项序列化并写入磁盘，减少生产者线程的数量，或者通过某种方式来抑制生产者线程。
在构建高可靠的应用程序时，有界队列是一种强大的资源管理工具：它们能抑制并防止产生过多的工作项，使应用程序在负荷过载的情况下变得更加健壮。
开发人员总会假设消费者处理工作的速率赶上生产者生成工作项的速率，因此通常不会为工作队列的大小设置边界，但这将导致在之后需要重新设计架构。因此，应该尽早通过阻塞队列在设计中构建资源管理机制——这件事情做得越早，就越容易。许多情况下，阻塞队列能使这项工作更加简单，如果阻塞队列并不完全符合设计需求那么还可以通过信号量（Semaphore）来创建其他的阻塞数据结构。
类库中中包含一些BlockingQueue的实现：
LinkedBlockingQueue，底层基于链表结构的阻塞队列，默认情况下容量是没有上限的（说的不准确，在不指定时容量为Integer.MAX_VALUE，要不然的话在put时怎么会受阻呢），但是也可以选择指定其最大容量，它是基于链表的队列，此队列按 FIFO（先进先出）排序元素。
ArrayBlockingQueue，底层基于数组结构的阻塞队列，在构造时需要指定容量，并可以选择是否需要公平性，如果公平参数被设置true，等待时间最长的线程会优先得到处理（其实就是通过将ReentrantLock设置为true来达到这种公平性的：即等待时间最长的线程会先操作）。通常，公平性会使你在性能上付出代价，只有在的确非常需要的时候再使用它。它的底层基于数组的阻塞循环队列，此队列按 FIFO（先进先出）原则对元素进行排序。
PriorityBlockingQueue，它是基于PriorityQueue来实现的，而PriorityQueue优先队列底层是基于堆数据结构的，是一个带优先级的队列，而不是先进先出队列。元素按优先级顺序被移除，该队列也没有上限（因PriorityQueue是没有容量限制的，与ArrayList一样，所以在优先阻塞队列上put时是不会受阻的。虽然此队列逻辑上是无界的，但是由于资源被耗尽，所以试图执行添加操作可能会导致 OutOfMemoryError），但是如果队列为空，那么取元素的操作take就会阻塞，所以它的检索操作take是受阻的。另外，放入的元素要具有比较力或构建队列时指定一个Comparator比较器。
DelayQueue，也是基于PriorityQueue来实现的，是一个存放Delayed 元素的无界阻塞队列，只有在延迟期满时才能从中提取元素。该队列的头部是延迟期满后保存时间最长的 Delayed 元素。如果延迟都还没有期满，则队列没有头部，并且poll将返回null。当一个元素的 getDelay方法返回一个小于或等于零的值时，则出现期满，poll就以移除这个元素了。此队列不允许使用 null 元素。下面是Delayed接口：
public interface Delayed extends Comparable<Delayed> {
long getDelay(TimeUnit unit);
}最后一个BlockingQueue的实现是SynchronousQueue，但它根本上不是一个真正的队列，因为这个类没有存储元素的空间。不过，它维护一个排队的线程清单，这些线程等待把元素加入队列或者移出队列。这好比在洗盘子时，没有盘架子（在生产者与消费者模式中相当于它们之间的缓冲区）一样，却是直接将洗好的盘子放入烘干机。这种直接地移交工作，减少了在生产者和消费者之间移动数据的延迟时间（在传统的队列中，在一个工作单元可以交付之前，必须通过串行方式首先完成入列或者出列等操作）。另外直接移交任务同样会给生产者带来更多关于任务状态的反馈信息，当移交被接受，它就知道消费者已经得到了任务，而不是简单地把任务放在一个队列或是什么其他地方。因为SynchronousQueue没有存储能力，所以除非另一个线程已经准备好参与移交工作，否则put和take会一直阻塞。仅当有足够多的消费者，并且总是有一个消费者准备好获取交付的工作时，才适合使用同步队列。（而生产者是主动发起，发不发起及什么时候发起移交动作由生产者来决定，所以只需要有够的消费者，而生产者多还是少不重要）。具体示例请参考：SynchronousQueue（同步队列）章节。
示例：LinkedBlockingQueue
/**
* @author jiangzhengjun
* @date 2010-6-11
*/
public class TestLinkedBlockingQueue {
// 随机获取字母
private static char getChar() {
return (char) (Math.random() * 26 + 65);
}
// 随机睡几秒
private static void sleep() {
try {
Thread.sleep((long) (Math.max(500, Math.random() * 1000)));
} catch (InterruptedException e) {
e.printStackTrace();
}
}
private static class Producer implements Runnable {// 生产者
private BlockingQueue<String> sq;
public Producer(BlockingQueue<String> d) {
this.sq = d;
}
public void run() {
try {
while (!Thread.currentThread().isInterrupted()) {
sq
.put(Thread.currentThread().getName() + " - "
+ getChar());
sleep();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
private static class Consumer implements Runnable {// 消费者
private BlockingQueue<String> sq;
public Consumer(BlockingQueue<String> d) {
this.sq = d;
}
public void run() {
try {
while (!Thread.currentThread().isInterrupted()) {
System.out.println(Thread.currentThread().getName() + " - "
+ sq.take());
sleep();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) {
// LinkedBlockingQueue阻塞队列，可换成其他阻塞队列
BlockingQueue<String> sq = new LinkedBlockingQueue<String>();
Thread t = null;
for (int i = 0; i < 2; i++) {
t = new Thread(new Producer(sq));
t.setName("Producer -" + i + "- ");
t.start();
t = new Thread(new Consumer(sq));
t.setName("Consumer -" + i + "- ");
t.start();
}
}
}示例：PriorityBlockingQueue
/**
* @author jiangzhengjun
* @date 2010-6-11
*/
public class TestPriorityBlockingQueue {
// 随机获取字母
private static char getChar() {
return (char) (Math.random() * 26 + 65);
}
public static void main(String[] args) throws Exception {
// 优先级队列
BlockingQueue<Character> sq = new PriorityBlockingQueue<Character>();
Map<Character, Character> hashMap = new HashMap<Character, Character>();
System.out.print("put - ");
for (int i = 0; i < 26; i++) {
char c = getChar();
while (hashMap.containsKey(c)) {
c = getChar();
}
hashMap.put(c, c);
System.out.print(c + " ");
sq.put(c);
}
System.out.println();
System.out.print("take - ");
for (int i = 0; i < 26; i++) {
System.out.print(sq.take() + " ");
}
}
} /*
put - D G Q X P I Z L K E T O W F Y N M B V S U J H A R C
take - A B C D E F G H I J K L M N O P Q R S T U V W X Y Z
*/从上面的输出可以看出队与入队的顺序是不一样的，在入队时会将元素进行排序，这里按照字母的自然顺序排列，所以出出队时就是有顺的了。
示例：DelayQueue
/**
* @author jzj
* @date 2010-6-11
*/
public class TestDelayQueue {
// 放入DelayQueue的对象需实现Delayed接口
static class DelayObj implements Delayed {
long time;// 可从队列中取出的时间点
int id;
public int getId() {
return id;
}
// time的单位是秒
DelayObj(long time, int id) {
this.time = System.nanoTime() + TimeUnit.SECONDS.toNanos(time);
this.id = id;
}
public int compareTo(Delayed y) {
long i = time;
long j = ((DelayObj) y).time;
if (i < j)
return -1;
if (i > j)
return 1;
return 0;
}
public boolean equals(Object other) {
return ((DelayObj) other).time == time;
}
public long getDelay(TimeUnit unit) {
long n = time - System.nanoTime();// 剩余延迟时间
return unit.convert(n, TimeUnit.NANOSECONDS);
}
public long getTime() {
return time;
}
public String toString() {
return String.valueOf(time);
}
}
public static void main(String args[]) throws InterruptedException {
Random random = new Random();
DelayQueue<DelayObj> queue = new DelayQueue<DelayObj>();
for (int i = 0; i < 5; i++) {
queue.add(new DelayObj(random.nextInt(5), i));
}
long last = 0;
for (int i = 0; i < 5; i++) {
// 只有延迟时间点到后才会取出
DelayObj delay = queue.take();
//从延时时间到达至从队列中取出来使用所相隔时间
long tmpTime = System.nanoTime() - delay.getTime();
long t = delay.getTime();
System.out.println("DelayObj_" + delay.getId() + "- "
+ delay.getTime() + " " + tmpTime);
if (i != 0) {
// 打印后一个比前一个延迟了多少
System.out.println("Delay last: " + (t - last));
}
last = t;
}
}
} /*
DelayObj_2- 10380274830293 40575
DelayObj_0- 10381273985449 414310
Delay last: 999155156
DelayObj_3- 10381274835355 15126560
Delay last: 849906
DelayObj_4- 10381274839786 15205837
Delay last: 4431
DelayObj_1- 10384274818037 15111507
Delay last: 2999978251
*/从上面的输出来看，取出的顺序与放入的顺序是完全不一样的，取出的顺序是依赖于DelayObj的延时时间点time的值，即哪个时间点在前就先取出哪个。
有一种类型的程序适合被分解为生产者和消费者，例如代理程序，它将扫描本地驱动器上的文件并建立索引以便随后进行搜索，类似于某些桌面搜索程序或者Windows索引服务。
public class IndexingService {
private static final int CAPACITY = 1000;
private static final File POISON = new File("");
private final IndexerThread consumer = new IndexerThread();
private final CrawlerThread producer = new CrawlerThread();
private final BlockingQueue<File> queue;
private final FileFilter fileFilter;
private final File root;
public IndexingService(File root, final FileFilter fileFilter) {
this.root = root;
this.queue = new LinkedBlockingQueue<File>(CAPACITY);
this.fileFilter = new FileFilter() {
public boolean accept(File f) {
return f.isDirectory() || fileFilter.accept(f);
}
};
}
private boolean alreadyIndexed(File f) {
return false;
}
class CrawlerThread extends Thread {
public void run() {
try {
crawl(root);
} catch (InterruptedException e) { /* fall through */
} finally {
while (true) {
try {
queue.put(POISON);
break;
} catch (InterruptedException e1) { /* retry */
}
}
}
}
private void crawl(File root) throws InterruptedException {
File[] entries = root.listFiles(fileFilter);
if (entries != null) {
for (File entry : entries) {
if (entry.isDirectory())
crawl(entry);
else if (!alreadyIndexed(entry))
queue.put(entry);
}
}
}
}
class IndexerThread extends Thread {
public void run() {
try {
while (true) {
File file = queue.take();
if (file == POISON)
break;
else
indexFile(file);
}
} catch (InterruptedException consumed) {
}
}
public void indexFile(File file) {
/*...*/
};
}
public void start() {
producer.start();
consumer.start();
}
public void stop() {
producer.interrupt();
}
public void awaitTermination() throws InterruptedException {
consumer.join();
}
}将文件遍历与建立索引等功能分解为独立的组件，比如将所有功能都放到一个操作中实现有着更高的代码可读性与可重用性：每个操作只需完成一个任务，并且阻塞队列将负责所有的控制流，因此每个功能的代码都更加简单和清晰。
生产者-消费者模式带来许多性能优势。生产者和消费者可以并发地执行。如果一个是I/O密集型，另一个时CPU密集型，那么并发执行的吞吐率要高于串行执行的吞吐率。如果生产者和消费者的并行度不同，那么将它们紧密耦合在一起会把整体并行度降低为二者中更小的并行度。
在java.util.concurrent中实现的各种阻塞队列都包含了足够的内部同步机制，从而安全地将对象从生产者发布到消费者。
线程封闭对象只能由单个线程拥有，但可以通过安全地发布该对象来“转移”所有权。在转移所有权后，也只有另一个线程能获得这个对象的访问权限，并且发布对象的线程不会再访问它。这种安全的发布确保了对象状态对于新的所有者来说是可见的，并且由于最初的所有者不会再访问它，因此对象将被封闭在新的线程中。新的所有者线程可以对该对象做任意修改，因为它具有独占的访问权。
对象池利用了串行线程封闭，将对象“借给”一个请求线程，只要对象池包含足够的内部同步来安全地发布池中的对象，并且只要客户端代码本身不会发布池中的对象，或者在将对象返回给对象池后就不再使用它，那么就可以安全地在线程之间传递所有权。
我们也可以使用其他发布机制来传递可变对象的所有权，但必须确保只有一个线程能接受被转移的对象。阻塞队列简化了这项工作，除此之外，还可以通过ConcurrentMap的原子方法remove或者AtomicReference的原子方法compareAndSet来完成这项工作。
Java6增加了两种容器类型，Deque和BlockingDeque，它们分别对应Queue和BlockingQueue进行了扩展。Deque是一个双端队列，实现了在队列头和队列尾的高效插入和移除。具体实现包括ArrayDeque和LinkedBlockingDeque
正如阻塞队列适用于生产者 - 消费者模式，双端队列同样适用于另一种相关模式，即工作密取（Work Stealing）。在生产者 - 消费者设计中，所有消费者有一个共享的工作队列，而在工作密取设计中，每个消费者都有各自的双端队列。如果一个消费者完成了自己双端队列中的全部工作，那么它可以从其他消费者双端队列末尾秘密地获取工作。密取工作模式比传统的生产者-消费者模式具有更高的可伸缩性，这是因为工作者线程不会在单个共享的任务队列上发生竞争。在大多数时候，它们都只是访问自己的双端队列，从而极大地减少了竞争。当工作者线程需要访问另一个队列时，它会从队列的尾部而不是从头部获取工作，因此进一步降低了队列的竞争程度。
工作密取非常适用于既是消费者也是生产者问题——当执行某个工作时可能导致出现更多地工作。例如，网络爬虫程序中处理一个页面时，通常会发现有更多地页面需要处理。类似的还有许多搜索图的算法，例如在垃圾回收阶段对堆进行标记，都可以通过工作密取机制来实现高效并行。当一个工作线程找到新的任务单元时，它会将其放到自己队列的末尾（或者在工作共享设计模式中，放入其他工作者线程的队列中）。当双端队列为空时，它会在另一个线程的队列末尾查找新的任务，从而确保每个线程都保持忙碌状态。
线程可能会阻塞或暂停执行，原因有多重：等待I/O操作结束，等待获得一个锁，等待从Thread.sleep方法中醒来，或是等待另一个线程的计算结果。当线程阻塞时，它通常被挂起，并处以某种阻塞状态（BLOCKED，WAITING或TIMED_WAITING）。阻塞操作与执行时间很长的普通操作的差别在于，被阻塞的线程必须等待某个不受它控制的事件发生后才能继续执行，例如等待I/O操作完成，等待某个锁编程可用，或者等待外部计算的结束。当某个外部事件发生时，线程被置回RUNNABLE状态，并可以再次被调度执行。
BlockingQueue的put和take方法会抛出一个受检查的InterruptedException，这与类库中的其他方法的做法是相同的，比如Thread.sleep。当一个方法能够抛出中断异常时，是在告诉你这个方法是一个阻塞方法，并且，如果这个方法被中断，将它将努力提前结束阻塞状态。
Thread提供了interrupt方法，用于中断线程或者查询线程是否已经被中断。每个线程都有一个布尔的属性，表示线程的中断状态，当中断线程时将设置这个状态。
中断是一种协作机制。一个线程不能强制其他线程停止正在执行的操作而去执行其他的操作。当线程A中断B时，A仅仅是要求B在执行到某个可以暂停的地方停止正在执行的操作——前提是如果线程B愿意停止下来。虽然在API或者语言规范中并没有为中断定义任何特定应用级别的语义，但最常使用中断的情况就是取消某个操作。方法对中断请求的响应速度越高，就越容易即使取消那些执行时间很长的操作。
当在代码中调用了一个将抛出InterruptedException异常的方法时，你自己的方法也就变成了一个阻塞方法，并且必须要处理对中断的响应。对于库代码来说，有两种基本选择：
public class TaskRunnable implements Runnable {
BlockingQueue<Task> queue;
public void run() {
try {
processTask(queue.take());
} catch (InterruptedException e) {
// 恢复被中断的状态，即重新设置中断标记
Thread.currentThread().interrupt();
}
}
void processTask(Task task) {
// Handle the task
}
interface Task {
}
}还可以采用一些更复杂的中断处理方法，但上述两种方法已经可以应付大多数情况了，然而在出现InterruptedException时不应该做得事情是，捕获它不作出任何响应。这将使调用栈上更高层的代码无法对中断采取处理措施，因为线程被中断的证据已经丢失。只有在一种特殊的情况中才能屏蔽中断，即对Thread进行扩展，并且能控制调用栈上所有更高层的代码。
同步工具类可以是任何一个对象，只要它根据其自身的状态来协调线程的控制流。阻塞队列可以作为同步工具类，其他类型的同步工具类还包括信号量（Semaphore）、（Barrier）以及闭锁（Latch）等，如果无法满足需要，可以创建自己的同步工具类。
java.util.concurrent包包含了若干能够帮助人们来管理线程相互合作的类。如果有一个相互合作的线程集，它又满足这些行为模式中的一种，那么应该直接重用合适的库类而不要去试图手工维护。
所有的同步工具类都包含了一些特定的结构化属性：它们封装了一些状态，这些状态将决定执行同步工具类的线程是继续执行还是等待，此外还提供了一些方法对状态进行操作，以及另一些方法用于高效地等待同步工具类进入预期状态。
Class
它能做什么
何时使用它
CyclicBarrier
允许一个线程集等待直至其中预定数量的线程达到一个公共检障栅为止，然后可以选择执行一个处理障栅的动作
当大量的线程需要在它们的结果可用之前完成时
CountDownLatch
允许一个线程集等待直到计数器减为0为止
当一个或多个线程需要等待直到制定数量的结果可用为止
Exchanger
允许两个线程在要交换的对象准备好时交换对象
当两个线程工作在同一个数据结构的两个实例上时，一个向实例中添加数据，另一个将数据从实例中清除
SynchronousQueue
允许一个线程将对象交给另一线程
在没有同步的情况下，当两个线程准备好将一个对象从一个线程传给另一个时
Semaphore
允许线程集等待直到允许继承运行为止
用来限制访问资源的线程总数。如果许可是1，则阻塞线程直到另一个线程给出许可为止
闭锁是一种同步工具类，可以延迟线程的进度直到其到达终止状态。闭锁的作用相当于一扇门：在闭锁到达结束状态之前，这扇门一直是关闭的，并且没有任何线程能通过，当到达结束状态时，这扇门会打开并允许所有的线程通过。当闭锁到达结束状态后，将不会再改变状态，因此这扇门将永远保持打开状态。闭锁可以用来确保某些活动直到其他活动都完成后才继续执行，例如：
CountDownLatch是一种灵活的闭锁实现，可以在上述的各种情况下使用，它可以使一个或多个线程等待一组事件发生。闭锁状态包括一个计数器，该计数器被初始化为一个正数，表示需要等待的事情数量。countDown方法递减计数器，表示有一个事件已经发生，而await方法等待计数器到达零，这表示所有需要等待的事件都已经发生。如果计数器的值非零，那么await会一直阻塞直到计数器为零，或者等待中的线程中断，或者等待超时。
TestHarness展示了两种常见的用法。它有两个闭锁，一个是“开始阀门”和一个“结束阀门”，开始阀门将计数器初始化为1，结束阀门将计数器初始化为工作线程的数量。每个线程首先要做的事就是在启动阀门上等待，从而确保所有线程都就绪后才开始执行。而每个线程要做的最后一件事情时将调用结束门的countDown方法减1，这能使主线程高效地等待直到所有工作线程都执行完成，因此可以统计所消耗的时间。
public class TestHarness {
public static long timeTasks(int nThreads, final Runnable task)
throws InterruptedException {
final CountDownLatch startGate = new CountDownLatch(1);// 开始阀门
final CountDownLatch endGate = new CountDownLatch(nThreads);// 结束阀门
for (int i = 0; i < nThreads; i++) {
Thread t = new Thread() {
public void run() {
System.out.println(" " + System.currentTimeMillis());
try {
// 开始阀门，等待外界将阀门打开后（即所有线程都准备好后）
// 开始向下执行
startGate.await();
try {
task.run();
} finally {
// 当执行完后计数器减1
endGate.countDown();
}
} catch (Exception ignored) {
}
}
};
t.start();
}
long start = System.nanoTime();
startGate.countDown();//打开开始阀门
// 结束阀门，等待所有线程都完成后会自动打开
System.out.println("- " + System.currentTimeMillis());
endGate.await();
long end = System.nanoTime();
return end - start;
}
public static void main(String[] args) throws Exception {
System.out.println(timeTasks(10, new Runnable() {
public void run() {
}
}));
}
}/*
- 1276925505488
1276925505488
1276925505489
1276925505489
1276925505489
1276925505489
1276925505490
1276925505490
1276925505491
1276925505491
1276925505491
4738512
*/上面为什么不在线程创建后就立即运行，还要像上面那样使用闭锁的方式？如果我们简单地创建线程并启动线程，那么先启动的就比后启动的具有“领先优势”，并且根据活动线程数据的增加或者减少，这样的竞争度也在不断改变。而开始阀门让控制线程能够同时释放所有工作线程，结束阀门让控制线程能够等待最后一个线程完成任务，而不需要自己去判断每个线程是否完成。
？？但是要注意的是，上面还是不精确，因为从输出的结果可以看出startGate.countDown()是完全有可能在所有任务线程还没有开始前就已经先执行了，所有任务线程都聚集在一点后再同时执行在这里好像没有作到？？下面是我改进后的结果：
public static long timeTasks(int nThreads, final Runnable task)
throws InterruptedException {
final CountDownLatch startGate = new CountDownLatch(1);// 开始阀门
final CountDownLatch taskStartGate = new CountDownLatch(nThreads);// 任务开始阀
final CountDownLatch endGate = new CountDownLatch(nThreads);// 结束阀门
for (int i = 0; i < nThreads; i++) {
Thread t = new Thread() {
public void run() {
System.out.println(" " + System.currentTimeMillis());
//只有所有任务线程执行此句后主线程才能继续执行
taskStartGate.countDown();
try {
// 开始阀门，等待外界将阀门打开后（即所有线程都准备好后）
// 开始向下执行
startGate.await();
try {
task.run();
} finally {
// 当执行完后计数器减1
endGate.countDown();
}
} catch (Exception ignored) {
}
}
};
t.start();
}
//确保所有任务线程都启动后（即进行了run方法）开始往后执行
taskStartGate.await();
System.out.println("- " + System.currentTimeMillis());
long start = System.nanoTime();
startGate.countDown();//打开开始阀门
// 结束阀门，等待所有线程都完成后会自动打开
endGate.await();
long end = System.nanoTime();
return end - start;
}闭锁与障栅（CyclicBarrier）有下面向个不同点：
1、 不是所有线程（根据指定的线程数）都需要等待到闭锁打开为止
2、 闭锁可以由外部事件打开
3、 倒计时闭锁是一次性的，一旦计数器到达0，就不能再重用它了
FutureTask也可以用作闭锁。（FutureTask实现了Future语义，表示一种抽象的可生成结果的计算）。FutureTask表示的计算是通过Callable来实现的，相当于一种可生成结果的Runnable，并且可以处于以下3种状态：等待运行（Waiting to run），正在运行（Running）和运行完成（Completed）。“执行完成”表示计算的所有可能结束方式，包括正常结束、由于取消而结束和由于异常而结束等。当FutureTask进入完成状态后，它会永远停止在这个状态上。
Future.get的行为取决于任务的状态。如果任务已经完成，那么get会立即返回结果，否则get将阻塞直到任务进入完成状态，然后返回结果或者抛出异常。FutureTask将计算结果从执行计算的线程传递到获取这个结果的线程，而FutureTask的规范确保了这种传递过程能实现结果的安全发布。
FutureTask在Executor框架中表示异步任务，此外还可以用来表示一些时间较长的计算，这些计算可以在使用计算结果之前启动，并且计算结果将在稍后使用。通过提前启动计算，可以减少在等待结果时需要的时间。
public class Preloader {
private final FutureTask<ProductInfo> future =
new FutureTask<ProductInfo>(new Callable<ProductInfo>() {
public ProductInfo call() throws DataLoadException {
return loadProductInfo();
}
});
private final Thread thread = new Thread(future);
//不要在构造器中或静态初始化方法中启动线程，所以提供了这个单独的启动方法
public void start() { thread.start(); }//可以在get方法之前预先异步执行，等到调用get时结果已经准备后或者只需要等待很短的时间了
public ProductInfo get() throws DataLoadException, InterruptedException {
try {
//get会抛出三种异常：ExecutionException - 如果计算抛出异常；InterruptedException - 如果当前的线程在等待时被中断；以及非捕获性异常CancellationException - 如果计算被取消。
return future.get();
} catch (ExecutionException e) {// ExecutionException表示只要是计算过程的异常都会封装在这里面，再被Future.get重新抛出
Throwable cause = e.getCause();
if (cause instanceof DataLoadException)
throw (DataLoadException) cause;
else
throw launderThrowable(cause);
}
}
}
/** 如果Throwable是Error则抛出；如果是
* RuntimeException 则返回；其他情况抛出IllegalStateException
*/
public static RuntimeException launderThrowable(Throwable t) {
if (t instanceof RuntimeException)
return (RuntimeException) t;
else if (t instanceof Error)
throw (Error) t;
else
throw new IllegalStateException("Not unchecked", t);
}Callable表示的任务可以抛出受检查的或未受检查的异常，并且热河代码都可能抛出一个Error。无论任务代码抛出什么异常都会被封装到一个ExecutionException中，并在Future.get中重新抛出。
在Preloader中，当get方法抛出ExecutionException时，可能是以下三种情况之一：Callable抛出受检查异常，RuntimeException，以及Error。我们必须对每种情况进行单独处理，使用launderThrowable辅助方法来封装一些复杂的异常处理逻辑。在调用launderThrowable之前，Preloader会首先检查已知的受检查异常，并重新抛出它们。剩下的时未检查异常，Preloader将调用launderThrowable并抛出结果。如果Throwable传递给launderThrowable的是一个Error，那么launderThrowable将直接再次抛出它；如果不是RuntimeException，那么将抛出一个IllegalStateException表示这是一个逻辑错误。剩下的RuntimeException，launderThrowable将把它们返回给调用者，而调用者通常会重新抛出它们。
计数信号量用来控制同时访问某个特定资源的操作数量，或者同时执行某个特定操作的数量。计数信号量还可以用来实现某种资源池，或者对容器施加边界。
Semaphore中管理着一组虚拟的许可（permit），许可的初始数量可通过构造函数来指定。在执行操作时可以首先获取许可（只要还有剩余的许可），并在使用以后释放许可。如果没有许可，那么acquire将阻塞直到有许可（或者直到被中断或者操作超时）。release方法将返回一个许可给信号量。计算信号量的一种简化形式是二值信号量，即初始值为1的Semaphore。二值信号量可以用做互斥体，并具备不可重入的加锁语义：谁拥有这个唯一的许可，谁就拥有了互斥锁。
通俗一点讲，许可就像一个令牌，谁拿到令牌(acquire)就可以去执行了，如果没有令牌则需要等待。执行完毕，一定要归还(release)令牌，否则令牌会被很快用光，别的线程就无法获得令牌而执行下去了。
Semaphore可以用于实现资源池，例如数据库连接池。我们可以可以构造一个固定长度的资源池，当池为空时，请求资源将会失败，但你真正希望看到的行为是阻塞而不是失败，并且当池非空时解除阻塞。如果将通俗一点讲，许可就像一个令牌，谁拿到令牌(acquire)就可以去执行了，如果没有令牌则需要等待。执行完毕，一定要归还(release)令牌，否则令牌会被很快用光，别的线程就无法获得令牌而执行下去了。的计数值初始化为池的大小，并在池中获取一个资源之前首先调用acquire方法获取一个许可，在将资源池返回给池之后调用release释放许可，那么acquire将一直阻塞直到资源池不为空。（在构造阻塞对象池时，一种更简单的方法是使用BlockingQueue来保存池的资源。）
下面使用Semaphore把任何容器转换为有界的阻塞容器：
public class BoundedHashSet<T> {
private final Set<T> set;
private final Semaphore sem;
public BoundedHashSet(int bound) {
this.set = Collections.synchronizedSet(new HashSet<T>());
sem = new Semaphore(bound);// bound为容器允许的最大容量
}
public boolean add(T o) throws InterruptedException {
sem.acquire();//如果容器满后则阻塞
boolean wasAdded = false;
try {
wasAdded = set.add(o);
return wasAdded;
}
finally {
if (!wasAdded)//如果添加不成功，则释放刚刚获取的信号量
sem.release();
}
}
public boolean remove(Object o) {
boolean wasRemoved = set.remove(o);
if (wasRemoved)//如果删除成功，则需要释放一个信号量
sem.release();
return wasRemoved;
}
}栅栏类似于闭锁，它能阻塞一组线程直到某个事情发生。栅栏与闭锁的关键区别在于，所有的线程必须同时到达栅栏位置，才能继续执行。闭锁用于等待事件，而栅栏用于等待其他线程。栅栏用于实现一些协议，例如几个家庭决定在某个地方集合：”所有人6：00在麦当劳碰头，到了之后要等其他人，之后再讨论下一步要做的事情。”
CyclicBarrier可以使一定数量的参与方反复地在栅栏位置汇集，它在并行迭代算法中非常有用：这种算法通常将一个问题拆分成一些列相互独立的子问题。当线程到达栅栏位置时将调用await方法，这个方法将阻塞直到所有线程都到达栅栏位置。如果所有线程都到达了栅栏位置，那么栅栏将打开，此时所有线程都被释放，而栅栏将被重置以便下次使用。如果对await的调用超时，或者await阻塞的线程被中断，那么栅栏就被认为是打破了，所有阻塞的await调用都将终止并抛出BrokenBarrierException。如果成功地通过栅栏，那么await将为每个线程返回一个唯一的到达索引号（其中，索引 getParties() - 1 指示将到达的第一个线程，零指示最后一个到达的线程），我们可以利用这些索引“选举”产生一个领导线程，并在下一次迭代中由该领导线程执行一些特殊的工作。CyclicBarrier还可以使你将一个栅栏操作传递给构造函数，这是一个Runnable，当成功通过栅栏时会（在一个子任务线程中）执行它，但在阻塞线程被释放之前是不能执行的。在一组线程中的最后一个线程到达之后（但在释放所有线程之前），该命令只在每个屏障点运行一次，若在任何参与线程继续执行之前更新共享状态，此屏障操作很有用。
在模拟程序中通常需要使用栅栏，例如某个步骤中的计算可以并行执行，但必须等到该步骤中的所有计算都执行完毕才能进入下一个步骤。通过每两次更新之间等待栅栏，能够确保在第k步中的所有更新操作都已经计算完毕，才进入k+1步。
下面是并行求和，它将数据根据CPU的内核数量分成多个任务后求和，一旦所有分段计算任务计算出结果后，就可以将每个任务的和进行汇总：
//并行计算
public class ConcurrentCal {
private final CyclicBarrier barrier;// 屏障
private final int cpuCoreNumber;// 待计算数组
private final int[] data;// 等计算数组
private final AtomicInteger sum = new AtomicInteger();// 和
// 计算任务
class SumCal implements Runnable {
private int[] numbers;
private int start;
private int end;
public SumCal(int num, final int[] numbers, int start, int end) {
this.numbers = numbers;
this.start = start;
this.end = end;
}
public void run() {// 求每小段数组的和
int tmpSum = 0;
for (int i = start; i < end; i++) {
tmpSum += numbers[i];
}
sum.getAndAdd(tmpSum);//存储每段和
try {
/*
* 在最后一个线程调用 await 方法之前，都将一直等待。只有当所有线
* 程都到达这里后才能通过
*/
barrier.await();
} catch (InterruptedException ex) {
return;
} catch (BrokenBarrierException ex) {
return;
}
}
}
public ConcurrentCal(int[] data) {
this.data = data;
cpuCoreNumber = Runtime.getRuntime().availableProcessors();// CPU内核数
this.barrier = new CyclicBarrier(cpuCoreNumber, new Runnable() {// 计算完后输出结果
/*
* 在最后一个线程到达之后，即触发点是最后一个线程调用await方法，并在该
* 方法中调用直接调用这个run方法（注，不会启动另一线程，而是直接在调用
* await方法线程中执行），且该命令只在每个屏障点运行一次
*/
public void run() {
System.out.println("计算完毕 sum=" + sum);
}
});
}
public void start() {
// 根据CPU核心个数拆分任务
for (int i = 0; i < cpuCoreNumber; i++) {
int increment = (data.length + 1) / cpuCoreNumber;
int start = increment * i;
int end = increment * i + increment;
if (end > data.length) {
end = data.length;
}
new Thread(new SumCal(i, data, start, end)).start();
}
}
public static void main(String[] args) {
int[] numbers = new int[] { 1, 2, 3, 4, 5, 6, 7, 8, 10, 11 };
new ConcurrentCal(numbers).start();
}
}Exchanger是两个线程可以交换对象的同步点。每个线程都在进入 exchange 方法时给出某个对象，相互接受对方准备的对象。
它是一种两方栅栏，各方在栅栏位置上交换数据。当两方执行不对称操作时，Exchanger会非常有用，例如当一个线程向缓冲区写入数据，而另一个线程从缓冲区中读取数据。这些线程可以使用Exchanger来汇合，并将满的缓冲区与空的缓冲区交换。当两个线程通过Exchanger交换对象时，这种交换就把这两个对象安全地发布给另一方。
数据交换的时机取决于应用程序的响应需求。最简单的方案是，当缓冲区被填满时，由填充任务进行交换，当缓冲区为空时，由清空任务进行交换。这样会把需要交换的次数降至最低，但如果新数据的到达率不可预测，那么一些数据的处理过程就将延迟。另一个方法是，不仅当缓冲被填满时进行交换，并且当缓冲被填充到一定程度并保持一定时间后，也进行交换。
Exchanger 可能被视为 SynchronousQueue 的双向形式。
/**
* 生产线程不停地向未满绥存中添加元素直到满为止，
* 消费线程不停地从非空缓存读取元素直到缓冲空为止，
* 当生产线程的缓存满且消费线程的缓存空时，将两者
* 的缓存互换，就这样一直下去
* @author jiangzhengjun
* @date 2010-6-9
*/
public class TestExchanger {
//交换器
static Exchanger<DataBuffer> exchanger = new Exchanger<DataBuffer>();
static DataBuffer emptyBuffer = new DataBuffer(10);
static DataBuffer fullBuffer = new DataBuffer(10);
public static void main(String[] args) {
new Thread(new ProducerThread()).start();
new Thread(new ConsumerThread()).start();
}
//随机获取字母
private static char getChar() {
return (char) (Math.random() * 26 + 65);
}
//随机睡几秒
private static void sleep() {
try {
Thread.sleep((long) (Math.max(500, Math.random() * 1000)));
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//生产者 —— 不停的向缓冲中添加字母
static class ProducerThread implements Runnable {
public void run() {
int index = 0;
while (!Thread.currentThread().isInterrupted()) {
try {
// 如果发现数据缓冲满后，则不能再向缓冲添加元素，准备与消费线程中的空缓冲交换
if (emptyBuffer.isFull()) {
// 等待一个满的数据缓冲，一旦生产线程准备好，则互换，即将满的数所缓存
// 传递给消费线程，并获取消费线程会传进来的空的数据缓存
emptyBuffer = exchanger.exchange(emptyBuffer);
System.out.println("ProducerThread.capacity - "
+ emptyBuffer.data.size());
index = 0;
} else {//如果数据缓冲不满，则直到添加满为止
char c = getChar();
System.out.println("ProducerThread - " + (index++) + " - " + c);
emptyBuffer.put(c);
sleep();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
//消费者 —— 不停的从缓冲中获取字母
static class ConsumerThread implements Runnable {
public void run() {
int index = fullBuffer.capacity;
while (!Thread.currentThread().isInterrupted()) {
try {
// 如果发现数据缓冲为空，则不能再读取元素，准备与生产消费线程中的满缓冲交换
if (fullBuffer.isEmpty()) {
// 等待一个满的数据缓冲，一旦生产线程准备好，则互换，即将空的数所缓存
// 传递给生产线程，并获取生产线程会传进来的满的数据缓存
fullBuffer = exchanger.exchange(fullBuffer);
System.out.println("ConsumerThread.capacity - "
+ fullBuffer.data.size());
index = fullBuffer.capacity;
} else {//如果缓冲不为空，则读到缓冲为空止
System.out.println("ConsumerThread - " + (--index) + " - "
+ fullBuffer.get());
sleep();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
//数据缓冲
static class DataBuffer {
private List<Character> data;
private int capacity;//缓冲允许最大容量
public DataBuffer(int capacity) {
this.capacity = capacity;
data = new ArrayList<Character>();
}
public void put(char c) {
if (data.size() < capacity) {
data.add(c);
}
}
public char get() {
if (data.size() >= 0) {
return data.remove(data.size() - 1);
} else {
return 0;
}
}
public boolean isFull() {
return data.size() == capacity ? true : false;
}
public boolean isEmpty() {
return data.size() == 0 ? true : false;
}
}
}一个阻塞队列，在每次的put操作中必须等待另一线程执行对应的take操作，反之亦然。同步队列没有任何内部容量，甚至连一个队列的容量都没有。SynchronousQueue允许我们在两个线程之间交换单个元素。
同步队列类似于 CSP 和 Ada 中使用的 rendezvous 信道。它非常适合于传递性设计，在这种设计中，在一个线程中运行的对象要将某些信息、事件或任务传递给在另一个线程中运行的对象，它就必须与该对象同步（即信息在两个线程间是相同的，这与普通的列队缓存是不一样的，这个是即放即取，信息对象不会停留在同步队列中）。
对于正在等待的生产者和消费者线程而言，SynchronousQueue的构造器还支持可选的公平排序策略。默认构造器是下不保证这种排序。使用公平设置为 true 所构造的队列可保证线程以 FIFO 的顺序进行访问。公平通常会降低吞吐量，但是可以减小可变性并可避免得不到服务。
另外SynchronousQueue详细的介绍请参考：阻塞队列和生产者—消费者模式章节。
/**
* @author jiangzhengjun
* @date 2010-6-9
*/
public class TestSynchronousQueue {
// 随机获取字母
private static char getChar() {
return (char) (Math.random() * 26 + 65);
}
// 随机睡几秒
private static void sleep() {
try {
Thread.sleep((long) (Math.max(500, Math.random() * 1000)));
} catch (InterruptedException e) {
e.printStackTrace();
}
}
private static class Producer implements Runnable {// 生产者
private BlockingQueue<String> sq;
public Producer(BlockingQueue<String> d) {
this.sq = d;
}
public void run() {
try {
while (!Thread.currentThread().isInterrupted()) {
// 但不一定能做到即消即产，因为生产者远少于消费者
sq.put(Thread.currentThread().getName() + " - "
+ getChar());
sleep();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
private static class Consumer implements Runnable {// 消费者
private BlockingQueue<String> sq;
public Consumer(BlockingQueue<String> d) {
this.sq = d;
}
public void run() {
try {
while (!Thread.currentThread().isInterrupted()) {
System.out.println(Thread.currentThread().getName() + " - "
+ sq.take());// 即产即消，因为消费者比生产者充足
sleep();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) {
// 同步队列
BlockingQueue<String> sq = new SynchronousQueue<String>();
Thread t = null;
//我们只创建二个生产者
for (int i = 0; i < 2; i++) {
t = new Thread(new Producer(sq));
t.setName("Producer -" + i + "- ");
t.start();
}
//创建多个消费者
for (int i = 0; i < 10; i++) {
t = new Thread(new Consumer(sq));
t.setName("Consumer -" + i + "- ");
t.start();
}
}
}几乎每个服务器应用程序都使用某种形式的调整缓存。利用已有的计算结果可以降低延迟，提高吞吐量，代价是占用更多的内存。
Computable
public interface Computable<A, V> {//具有计算能力接口
V compute(A arg) throws InterruptedException;
}
//实现计算接口
class ExpensiveFunction implements Computable<String, BigInteger> {
public BigInteger compute(String arg) {//假设这是一个耗时的计算实现
// after deep thought...
return new BigInteger(arg);
}
}
class Memoizer1<A, V> implements Computable<A, V> {
//使用HashMap来建立缓存
private final Map<A, V> cache = new HashMap<A, V>();
private final Computable<A, V> c;//
public Memoizer1(Computable<A, V> c) {
this.c = c;
}
//注意这里需要同步HashMap
public synchronized V compute(A arg) throws InterruptedException {
V result = cache.get(arg);//缓存中是否已有
if (result == null) {//如果缓存中没有则重新计算
result = c.compute(arg);
cache.put(arg, result);
}
return result;
}
}上面实现中Memoizer1中compute方法被整个同步了，这保证了线程安全，但是却带来一个明显的可伸缩性问题：一次只有一个线程能够执行compute。如果另外一个线程正忙于计算结果，其他调用compute的线程可能被阻塞很长时间。那么，compute可能会比不使用缓存花费更长的时间。这显示不是我们希望的结果。
下面使用ConcurrentHashMap取代HashMap，改进Memoizer1中这种糟糕的并发行为。因为ConcurrentHashMap是线程安全的，所以不需要同步。Memoizer2与Memoizer1相比，毫无疑问具有更好的并发性：多线程可以真正并发地使用它了。但是它作为调整缓存仍然存在缺陷——当两个线程同时调用compute时，存在一个漏洞，可能同时计算相同的值。这对于这种备忘录形式的缓存，这仅仅是效率的问题（因为缓存的作用是避免相同的数据被计算多次），但是对于像一个缓存对象仅仅只能被初始化一次的缓存，这个漏洞就会带来安全性问题了。
public class Memoizer2<A, V> implements Computable<A, V> {
private final Map<A, V> cache = new ConcurrentHashMap<A, V>();
private final Computable<A, V> c;
public Memoizer2(Computable<A, V> c) { this.c = c; }
public V compute(A arg) throws InterruptedException {
V result = cache.get(arg);
if (result == null) {//计算相同参数的线程可能在相隔不长的时间内都到达
result = c.compute(arg);
cache.put(arg, result);
}
return result;
}
}Memoizer2的问题在于，如果某个线程启动了一个开销很大的计算，而其他线程并不知道这个计算正在执行，那么很可能会重复这个计算。我们希望通过某种方法来表达“线程X正在计算f（27）”这种情况，这样当另一个线程查询f（27）时，它能够知道最高效的方法是等待线程X计算结束，然后再去查询缓存“f（27）的结果是多少“。
有一个类能基本实现这个功能，：FutureTask。FatureTask表示一个计算的过程，这个过程可能已经计算完成，也可能正在进行。如果有结果可用，那么FutureTask.get将立即返回结果，否则它会一直阻塞，直到结果计算出来再将其返回。
Memoizer3将用于缓存值得Map重新定义为ConcurrentHashMap<A,Future<V>>,替换原来得ConcurrentHashMap<A,V>。Memoizer3首先检查某个相应的计算是否已经开始（Memoizer2与之相反，它首先判断某个计算是否已经已经完成）。如果还没有启动，那么就创建一个FutureTask，并注册到Map中，然后启动计算；如果已经启动，那么等待现有计算的结果。如果可能很快会得到，也可能还在运算过程中，但这对于Future.get的调用者来说是透明的。
class Memoizer3<A, V> implements Computable<A, V> {
//缓存的不是值，而是一个计算的任务过程，它可能正在计算，也可能已计算完成
private final Map<A, Future<V>> cache = new ConcurrentHashMap<A, Future<V>>();
private final Computable<A, V> c;
public Memoizer3(Computable<A, V> c) {
this.c = c;
}
public V compute(final A arg) throws InterruptedException {
Future<V> f = cache.get(arg);
if (f == null) {
Callable<V> eval = new Callable<V>() {
public V call() throws InterruptedException {
return c.compute(arg);
}
};
FutureTask<V> ft = new FutureTask<V>(eval);
f = ft;
cache.put(arg, ft);//这里与上面if一起使用时是一个复合操作，所以还是有问题
ft.run(); // call to c.compute happens here
}
try {
return f.get();//如果还没有计算完则等待，否则马上返回
} catch (ExecutionException e) {
e.printStackTrace();
}
return null;
}
}Memorizer3的实现几乎是完美的：它表现出了非常好的并发性（基本上是源于ConcurrentHashMap高效地并发性），若结果已经计算出来，那么将立即返回。如果其他线程正在计算结果，那么新到的线程将一直等待这个结果被计算出来。它只有一个缺陷，即仍然存在两个线程计算出相同值得漏洞。这个漏洞的发生概率要远小于Memoizer2中发生的概率，但由于compute方法中的if代码块仍然是非原子的“先检查再执行”操作，因此两个线程仍有可能在同一时间内调用compute来计算相同的值，即二者都没有在缓存中找到期望的值，因此都开始计算。
Memoizer3中存在这个问题的原因是，复合操作（”若没有则添加“）是在底层的Map对象上执行的，而这个对象无法通过加锁来确保原子性，如下程序段使用了ConcurrentHashMap中的原子方法putIfAbsent，避免了Memoizer3的漏洞。
使用putIfAbsent方法来替换上面的复合操作：
public class Memoizer <A, V> implements Computable<A, V> {
private final ConcurrentMap<A, Future<V>> cache
= new ConcurrentHashMap<A, Future<V>>();
private final Computable<A, V> c;
public Memoizer(Computable<A, V> c) {
this.c = c;
}
public V compute(final A arg) throws InterruptedException {
while (true) {
Future<V> f = cache.get(arg);
if (f == null) {
Callable<V> eval = new Callable<V>() {
public V call() throws InterruptedException {
return c.compute(arg);
}
};
FutureTask<V> ft = new FutureTask<V>(eval);
f = cache.putIfAbsent(arg, ft);//重新检测if条件不是否满足
if (f == null) {//只有在成功放入后才开始计算
f = ft;
ft.run();
}
}
try {
return f.get();//注，计算取消或失败时都需要从缓存中移除，因为取消与失败的任务下次还是不会计算成功
} catch (CancellationException e) {
cache.remove(arg, f);//如果计算被取消，则移除，以便下次新的计算
} catch (ExecutionException e) {
cache.remove(arg, f);//如果计算失败，也要移除，以便下次新的计算
throw LaunderThrowable.launderThrowable(e.getCause());//对ExecutionException异常进行分解
}
}
}
}当缓存的是Future而不是值时，将导致缓存污染问题：如果计算被取消或者失败，那么在计算这个结果时将指明计算过程被取消或者失败。为了避免这种情况，如果Memoizer发现计算被取消，那么将Future从缓存中移除。如果检测到RuntimeException，那么也会移除Future，这样将来的计算才可能成功。
Memoizer同样没有解决缓存逾期问题，但它可以通过使用FutureTask的子类来解决，在子类中为每个结果指定一个逾期时间，并定期扫描缓存中逾期的元素。（同样，它也没有解决缓存清理的问题，即移除旧的计算结果以便为新的计算结果腾出空间。从而使缓存不会消耗过多的内存。）
使用Memoizer为因式分解的servlet缓存结果：
@ThreadSafe
public class Factorizer implements Servlet {
private final Computable<BigInteger, BigInteger[]> c =
new Computable<BigInteger, BigInteger[]>() {
public BigInteger[] compute(BigInteger arg) {
return factor(arg);//计算某个数的因子，可能耗时比较长
}
};
private final Computable<BigInteger, BigInteger[]> cache
= new Memoizer<BigInteger, BigInteger[]>(c);//高速缓存
public void service(ServletRequest req, ServletResponse resp) {
try {
BigInteger i = extractFromRequest(req);
encodeIntoResponse(resp, cache.compute(i));//从缓存中取或重新计算
} catch (InterruptedException e) {
encodeError(resp, "factorization interrupted");
}
}
}参考：Java并发编程
版权声明：本文为博主原创文章，未经博主允许不得转载。
[Java Concurrency in Practice]第五章 基础构建模块
标签：并发
原文地址：http://blog.csdn.net/coslay/article/details/47704673
