Java并发编程实战(第五章1)

5.1 同步容器类

5.1.1 同步容器类的问题
同步容器类&＃xff08;Vector、Hashtable&＃xff09;都是线程安全的&＃xff0c;但在某些情况下可能需要额外的客服端加锁来保护复合操作。
复合操作&＃xff1a;
1&＃xff09;迭代&＃xff08;反复访问元素&＃xff0c;直到遍历完容器中所有元素&＃xff09;&＃xff1b;
2&＃xff09;跳转&＃xff08;根据指定顺序找到当前元素的下一个元素&＃xff09;以及条件运算。
在同步容器中&＃xff0c;这些复合操作在没有客服端加锁的情况下任然是线程安全的&＃xff0c;但但其他线程并发地修改容器时&＃xff0c;它们可能会表现出意料之外的行为。

package chapter5;
import java.util.Vector;/*** 在调用size与调用getLast这两个操作之间&＃xff0c;Vector变小了&＃xff0c;* 因此在调用size时得到的索引值将不再有效---将抛出ArrayIndexOutOfBoundsException异常*/
public class DemoVector {public static Object getLast(Vector vector){// 同步容器类通过其自身的锁来保护它的每个方法。通过获得容器类的锁&＃xff0c;我们可以使getLast// 和deleteLast成为原子操作&＃xff0c;并确保Vector大小在调用size和get之间不会发生变化。synchronized (vector) {int lastIndex &＃61; vector.size() - 1;return vector.get(lastIndex);}}public static Object deleteLast(Vector vector){synchronized (vector) {int lastIndex &＃61; vector.size() - 1;Object object &＃61; vector.remove(lastIndex);return object ;}}// 在调用size和相应的get之间&＃xff0c;vector的长度可能发生变化&＃xff0c;这种风险在对Vector中元素// 进行迭代时任然会出现&＃xff0c;与getLast一样&＃xff0c;如果在对Vector进行迭代时&＃xff0c;另一个线程删除了一个// 一个元素&＃xff0c;并且这两个操作交替执行&＃xff0c;那么这种迭代方法将抛出ArrayIndexOutOfBoundsExceptionpublic static void doSomething(Vector vector){// 通过在迭代期间持有Vector的锁&＃xff0c;可以防止其他线程在迭代期间修改vector。然而&＃xff0c;这同样会导致// 其他线程在迭代期间无法访问它&＃xff0c;因此降低了并发性。synchronized (vector) {System.out.println(Thread.currentThread().getName() &＃43; ":" &＃43; vector.toString());}}public static void main(String[] args) throws InterruptedException {final Vector vector &＃61; new Vector();for (int i &＃61; 0; i < 10; i &＃43;&＃43;){vector.add(i) ;}for (int j &＃61; 0; j < 5; j &＃43;&＃43;) {Thread.sleep(1000);new Thread(new Runnable() {public void run() {System.out.println(Thread.currentThread().getName()&＃43; ":" &＃43;getLast(vector));}}).start();}for (int j &＃61; 0; j < 5 ; j &＃43;&＃43;) {new Thread(new Runnable() {public void run() {System.out.println(Thread.currentThread().getName() &＃43; " delete:" &＃43; deleteLast(vector));}}).start();}for (int j &＃61; 0 ; j < 5; j &＃43;&＃43;) {Thread.sleep(100);new Thread(new Runnable() {public void run() {doSomething(vector);}}).start();}}
}
/**
output:~Thread-0:9Thread-1:9Thread-2:9Thread-3:9Thread-4:9Thread-5 delete:9Thread-8 delete:8Thread-7 delete:6Thread-9 delete:5Thread-6 delete:7Thread-10:[0, 1, 2, 3, 4]Thread-11:[0, 1, 2, 3, 4]Thread-12:[0, 1, 2, 3, 4]Thread-13:[0, 1, 2, 3, 4]Thread-14:[0, 1, 2, 3, 4]*/

5.1.2 迭代器与ConcurrentModificationException
有时候开发人员并不希望在迭代期间对容器加锁。如果容器的规模很大&＃xff0c;或者在某个元素上执行操作的时间很长&＃xff0c;那么这些线程将长时间等待。即使不存在饥饿或者死锁等风险&＃xff0c;长时间地对容器加锁也会降低程序的可伸缩性。持有锁的时间越长&＃xff0c;那么在锁上的竞争就可能越激烈&＃xff0c;如果许多线程都在等待死锁被释放&＃xff0c;那么将极大地降低吞吐量和CPU的利用率。
如果不希望在迭代期间对容器加锁&＃xff0c;那么一种可替代方法是**“克隆”容器并在容器上进行迭代。副本被封锁在线程内&＃xff0c;其它线程不会在迭代期间对其进行修改–避免了抛出ConcurrentModificationException。&＃xff08;在克隆的过程中任然要对容器加锁**&＃xff0c;并且克隆容器时的开销由其大小、在每个元素上执行的操作&＃xff09;

5.1.3 隐藏迭代器
虽然加锁可以防止迭代器抛出ConcurrentModificationException&＃xff0c;但你必须要记住在所有对共享容器进行迭代的地方都需要加锁。 然而实际情况要更加复杂&＃xff0c;因为在某些情况下&＃xff0c;迭代器会隐藏起来。

package chapter5;import java.util.HashSet;
import java.util.Random;
import java.util.Set;/*** 虽然加锁可以防止迭代器抛出ConcurrentModificationException&＃xff0c;* 但你必须要记住在所有对共享容器进行迭代的地方都需要加锁。* * warning:* 容器的hashCode和equals等方法也会间接地执行迭代操作&＃xff1b;* 当容器作为另一个容器的元素或键时&＃xff0c;也会出现这种情况&＃xff1b;* 同样&＃xff0c;containsAll、removeAll和retainsAll等&＃xff1b;* 以及把容器作为参数的构造函数&＃xff0c;都会对容器进行迭代。*/
public class HiddenIterator {private final Set<Integer> set &＃61; new HashSet<Integer>();public synchronized void add(Integer i) {set.add(i);}public synchronized void remove(Integer i) {set.remove(i);}public void addTenThings() {Random r &＃61; new Random();for (int i &＃61; 0; i < 1000; i&＃43;&＃43;) {add(r.nextInt()); // 进行了同步}// 在使用println中的set之前必须首先获取HiddenIterator的锁&＃xff0c;但在// 但在调试代码和日志代码中通常会忽略这个要求System.out.println("DEBUG: added ten elements to " &＃43; set);}
}

正如封装对象的状态有助于维持不变性条件一样&＃xff0c;封装对象的同步机制同样有助于确保实施同步策略。

5.2 并发容器

通过并发容器来代替同步容器&＃xff0c;可以极大地提高伸缩性并降低风险。
5.2.1 ConcurrentHashMap
同步容器类在执行每个操作期间都持有一个锁。
与HashMap一样&＃xff0c;ConcurrentHashMap也是一个基于散列的Map&＃xff0c;但是它使用了一种完全不同的加锁机制。
ConcurrentHashMap并不是将每个方法都在同一个锁上同步并使得每次只能有一个线程访问容器&＃xff0c;而是使用一种粒度更细的加锁机制来实现更大程度的共享&＃xff0c;这种机制称为分段锁。
ConcurrentHashMap与其他并发容器一起增强了同步容器类&＃xff1a;它们提供的迭代器不会抛出ConcurrentModificationException&＃xff0c;因此不需要在迭代过程中对容器加锁。只有当应用程序需要加锁Map以进行独占访问时&＃xff0c;才应该放弃使用ConcurrentHashMap.

5.2.2 额外的原子Map操作
由于ConcurrentHashMap不能被加锁来执行独占访问&＃xff0c;因此我们无法使用客户端加锁来创建新的原子操作。

5.2.3 CopyOnWriteArrayList
CoppyOnWriteArrayList用于替代同步List&＃xff0c;在某些情况下它提供了更好的并发性能&＃xff0c;并且在迭代期间不需要对容器进行加锁或复制。&＃xff08;类似地&＃xff0c;CopyOnWriteArraySet的作用是替代同步Set。&＃xff09;
“写入时复制”容器的线程安全性在于&＃xff0c;只要正确地发布一个事实不可变的对象&＃xff0c;那么在访问该对象时&＃xff0c;就不再需要进一步的同步。在每次修改时&＃xff0c;都会创建并重新发布一个新的容器副本&＃xff0c;从而实现可变性。

5.3 阻塞队列和生产者-消费者模式

阻塞队列提供了可阻塞的put和take方法&＃xff0c;以及支持定时的offer和poll方法。队列已满–阻塞put直到有空间可用。队列为空–阻塞take直到有元素可用。
在构建高可靠的应用程序时&＃xff0c;有界队列是一种强大的资源管理工具&＃xff1a;它们能抑止并防止产生过多的工作项&＃xff0c;使应用程序在负载过高的情况下变得跟家健壮。

5.3.2串行线程封闭
在java.util.concurrent中实现的各种阻塞队列都包含了足够的内部同步机制&＃xff0c;从而安全地将对象从生产者线程发布到消费者线程。

5.4 阻塞方法与中断方法

线程可能会阻塞或者暂停执行&＃xff0c;原因有多种&＃xff1a;等待I/O操作结束&＃xff0c;等待获得一个锁&＃xff0c;等待从Thread.sleep方法中醒来&＃xff0c;或是等待另一个线程的计算结果。当线程阻塞时&＃xff0c;它通常被挂起&＃xff0c;并处于某种阻塞状态(BLOCKED、WAITING或TIME_WAITING&＃xff09;。
阻塞操作与执行时间很长的普通操作的差别在于&＃xff0c;被阻塞的线程必须等待某个不受它控制的事件发生后才能继续执行&＃xff0c;例如等待I/O操作完成&＃xff0c;等待某个锁变成可用&＃xff0c;或者等待外部计算的结束。
一个线程不能强制其它线程停止正在执行的操作而去执行其他的操作。方法对中断请求的响应度越高&＃xff0c;就越容易及时取消那些执行时间长的操作。

public class TaskRunnable implements Runnable{BlockingQueue<Tast> queuqu;...public void run(){try{proccessTask(queue.take()) ;} catch(InterruptedException e) {// 恢复中断的状态Thread.currentThread().interrupt() ;}}
}


5.5 同步工具类

阻塞队列&＃xff1a;不仅能作为保存对象的容器&＃xff0c;还能协调生产者和消费者之间的控制流&＃xff0c;因为take和put等方法将阻塞&＃xff0c;直到队列达到期望的状态(队列既非空&＃xff0c;也非满)。

1. 闭锁
   在闭锁到达结束状态之前&＃xff0c;这扇门一直是关闭的&＃xff0c;并且没有任何线程能通过&＃xff0c;当到达结束状态时&＃xff0c;这扇门会打开并释放所有的线程通过。

import java.util.concurrent.CountDownLatch;
public class TestHarness{public long timeTasks(int nThreads, final Runnable task) throws InterruptedException{final CountDownLatch startGate &＃61; new CountDownLatch(1); // 初始化闭锁final CountDownLatch endGate &＃61; new CountDownLatch(nThreads);for (int i &＃61; 0; i < nThreads; i &＃43;&＃43;){Thread t &＃61; new Thread(){public void run(){try{startGate.await(); // 在启动门上等待try{task.run();}finally{endGate.countDown(); // 调用结束门的countDown方法减一}}catch(InterruptedException ignored){}}};t.start();}long start &＃61; System.nanoTime();startGate.countDown();endGate.await();long end &＃61; System.nanoTime();return end-start;}
}