《Java并发编程实践》（四）---- 构建阻塞

一，同步容器

同步容器类包括Vector和Hashtable。这些类实现线程安全的方式是：将它们的状态封装起来，并对每个公有方法都进行同步，使得每次只有一个线程能访问容器本身的状态。

1，同步容器的问题

同步容器类都是线程安全的，但在某些情况下可能需要额外的客户端加锁来保护复合操作。同步容器将所有对容器状态的访问都串行化，以实现它们的线程安全性，这样的代价就是严重降低并发性，当多个线程竞争容器的锁时，吞吐量将严重降低

例如：

public static Object getLast(Vector list) {
    int lastIndex = list.size() - 1;
    return list.get(lastIndex);
}

public static void deleteLast(Vector list) {
    int lastIndex = list.size() - 1;
    list.remove(lastIndex);
}

如果线程A在执行deleteLast, 线程B在执行getLast，list中有10个元素，刚好B在获取完lastIndex(9)和get(9)之间，线程A执行完了remove(9), 那么线程B在执行get(9)时就会抛出ArrayIndexOutOfBoundsException。

可以在客户端加锁来原子操作Vector上的复合操作：

public static Object getLast(Vector list) {
    synchronized(list) {
        int lastIndex = list.size() - 1;
        return list.get(lastIndex);
    }
}

public static void deleteLast(Vector list) {
    synchronized(list) {
        int lastIndex = list.size() - 1;
       list.remove(lastIndex);
    }
}

2, Iterators和ConcurrentModificationException

对容器类进行迭代的标准方式是使用Iterator，然而，如果有其他线程并发地修改容器，那么即使是使用迭代器也无法避免地需要在同步容器上加锁。在设计同步容器类的迭代器时并没有考虑到并发修改的问题，它们的迭代器是“及时失败”的，所以当它们发现容器在迭代过程中发生变化，就会抛出一个ConcurrentModificationException异常。这种fail-fast机制并不是一种完备的处理机制，而只是“善意地”捕获并发错误，因此只能作为并发问题的预警指示器。它们采取的实现方式是将计数器变化与容器关联起来：如果在迭代期间计数器被修改，那么hasNext或next将抛出ConcurrentModificationException。然而，这种检查是在没有同步的情况下进行的，因此可能会看到失效的值，而迭代器可能并没有意识到已经发生了修改。要想避免出现ConcurrentModificationException，就必须在迭代过程中持有容器的锁。

然而，有时候开发人员并不希望在迭代器间对容器加锁。例如，某些线程在可以访问容器之前，必须等待迭代过程结束，如果容器规模很大，或者在每个元素上执行操作的时间很长，那么这些线程就需要长时间等待。持有锁的时间越长，那么在锁上的竞争就越激烈，如果许多线程都在等待锁被释放，那么将极大地降低吞吐量和CPU的利用率。

另一种替代方法是“克隆”容器，并在副本上进行迭代。由于副本被封闭在线程内，因此其他线程不会在迭代期间对其进行修改，这样就避免了抛出ConcurrentModificationException，不过在克隆过程中仍然要加锁（以防在此期间被克隆容器被其他线程修改，那样克隆出来的容器就是失效的容器），所以也会增加性能开销。所以这种方法的好坏取决于多个因素：容器的大小，在每个元素上执行的操作，迭代操作相对于容器上其他操作被调用的频率，以及在响应时间和吞吐量等方面的需求。

3，隐藏迭代器

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

如下例，标准容器的toString方法将迭代容器，并在每个元素上调用toString来生成容器内容的格式化表示：

public class HiddenIterator {
    @GuardedBy("this") private final Set<Integer> set = new HashSet<Integer>();
    public synchronized void add(Integer i) {
        set.add(i);
    }

    public synchronized void remove(Integer i) {
        set.remove(i);
    }

    public void addTenThing() {
        Random r = new Random();
        for (int i=0; i< 10; i++) {
            set.add(r.nextInt());
        }
        System.out.println("DEBUG: added ten elements to " + set);
    }
}

addTenThings方法可能会抛出ConcurrentModificationException，因为toString对set进行了迭代，而且没加锁。

如果状态与保护它的同步代码之间相隔越远，那开发人员就越容易忘记在访问状态时使用正确的同步。如果HiddenIterator用synchronizedSet来包装HashSet，并且对同步代码进行封装，那么就不会抛出异常了。

容器的hashCode和equals等方法也会间接地执行迭代操作，同样，containsAll, removeAll和retainAll等方法，以及把容器作为参数的构造函数，都会对容器进行迭代，所有这些间接的迭代操作都可能抛出ConcurrentModificationException。

二，并发容器

同步容器将所有对容器状态的访问都串行化，以实现它们的线程安全性，这样的代价就是严重降低并发性，当多个线程竞争容器的锁时，吞吐量将严重降低。Java 5.0提供了多种并发容器类来改进同步容器的性能。通过并发容器来代替同步容器，可以极大地提高伸缩性并降低风险。

Java 5.0增加了ConcurrentHashMap，用来替代基于hash的同步map，增加了CopyOnWriteArrayList，用来替代以遍历操作为主要操作的同步List。在新的ConcurrentMap接口中增加了一些常用的复合操作，例如“putIfAbsent”,replace, 和 conditional remove.

Java 5.0还增加了两个新的集合类型，Queue和BlockingQueue。

Java 6.0增加了ConcurrentSkipListMap来替换同步的SortedMap，增加了ConcurrentSkipListSet替换SortedSet（例如TreeMap和TreeSet）

1，ConcurrentHashMap

同步容器类在执行每个操作期间都持有一个锁，在某些情况下会花费很长时间，而其他线程在这段时间内都不能访问该容器。

与HashMap一样，ConcurrentHashMap也是一个基于HashCode的Map，但它使用了一种完全不同的加锁策略来提供更高的并发性和伸缩性。ConcurrentHashMap并不是将每个方法都在同一把锁上同步并使得每次只有一个线程访问容器，而是使用一个种粒度更细的加锁机制来时间共享，叫做分段锁。在这种机制下，任意数量的读取线程可以并发地访问这个map，执行读取操作的线程和执行写入操作的线程可以并发地访问map，并且一定数量的写入线程可以并发地修改Map。

而且ConcurrentHashMap提供的迭代器不会抛出ConcurrentModificationException，因此不需要再迭代过程中对容器加锁。它返回的迭代器具有弱一致性，而并非“及时失败”。弱一致性的迭代器可以容忍并发的修改，当创建迭代器时会遍历已有的元素，并可以在迭代器被构造后将修改操作反映给容器。

但是对于一些需要在整个Map上进行计算的方法，如size和isEmpty，它们的返回结果有可能已经过时了，不过这些方法在并发环境下用处并不大，所以这些方法的精准性就被牺牲了以换取其他更重要操作的性能优化。

有一个小的缺陷是，在ConcurrentHashMap中没有实现对Map加锁以提供独占访问。在Hashtable和synchronized-Map中，获得Map的锁能防止其他线程访问这个Map，在一些不常见的情况下确实需要这种功能，例如通过原子方式添加一些mapping，或者遍历过map好几遍后仍然需要保持原来的顺序。

与Hashtable和synchronized-Map相比，ConcurrentHashMap有着更多的优势以及更少的劣势。因此在大多数情况下，用ConcurrentHashMap来代替同步Map能进一步提高代码的可伸缩性，只有当应用程序需要给map加锁以进行独占访问时，才应该放弃使用ConcurrentHashMap。

2，为Map添加额外的原子操作

由于ConcurrentHashMap不能被加锁来执行独占访问，因此也无法使用客户端加锁来创建新的原子操作。但是一些常见的复合操作，如“put-if-absent”,"remove-if-equals","replace-if-equals"等，都已经在ConcurrentMap接口中有声明，所以如果需要为现有的同步Map添加这样的功能，就应该考虑使用ConcurrentMap了。

3，CopyOnWriteArrayList

CopyOnWriteArrayList用于替代同步list，在某些情况下提供了更好的并发性能，并且在迭代器间不需要对容器进行加锁或复制。（类似地，CopyOnWriteArraySet的作用是替代同步set）

Copy-On-Write容器的线程安全性在于，只要正确地发布一个实际不可变的对象，那么在访问该对象时就不需要进一步的同步了。

Copy-On-Write从字面上看就是，Write的时候总是要Copy，所以在每次修改时，都会创建并重新发布一个新的容器副本。而CopyOnWriteArrayList容器的迭代器会保留一个指向原始数组的引用，遍历的也是原始数组，而其他线程修改的是这个原始数组的副本，所以也不会影响原始数组，原始数组不会改变，也就不会有ConcurrentModificationException了，并且返回的元素和迭代器创建时的元素完全一致。

显然，每当修改容器时都会复制原始数组，这需要一定开销，特别是当容器的规模较大时。仅当迭代器操作多于修改操作时，才应该使用“写入时复制”容器。

（参考http://www.cnblogs.com/xrq730/p/5020760.html）

CopyOnWriteArrayList这个并发组件，其实反映的是两个十分重要的分布式理念：

（1）读写分离

我们读取CopyOnWriteArrayList的时候读取的是CopyOnWriteArrayList中的Object[] array，但是修改的时候，操作的是一个新的Object[] array，读和写操作的不是同一个对象，这就是读写分离。这种技术数据库用的非常多，在高并发下为了缓解数据库的压力，即使做了缓存也要对数据库做读写分离，读的时候使用读库，写的时候使用写库，然后读库、写库之间进行一定的同步，这样就避免同一个库上读、写的IO操作太多

（2）最终一致

对CopyOnWriteArrayList来说，线程1读取集合里面的数据，未必是最新的数据。因为线程2、线程3、线程4四个线程都修改了CopyOnWriteArrayList里面的数据，但是线程1拿到的还是最老的那个Object[] array，新添加进去的数据并没有，所以线程1读取的内容未必准确。不过这些数据虽然对于线程1是不一致的，但是对于之后的线程一定是一致的，它们拿到的Object[] array一定是三个线程都操作完毕之后的Object array[]，这就是最终一致。最终一致对于分布式系统也非常重要，它通过容忍一定时间的数据不一致，提升整个分布式系统的可用性与分区容错性。当然，最终一致并不是任何场景都适用的，像火车站售票这种系统用户对于数据的实时性要求非常非常高，就必须做成强一致性的。

最后总结一点，随着CopyOnWriteArrayList中元素的增加，CopyOnWriteArrayList的修改代价将越来越昂贵，因此，CopyOnWriteArrayList适用于读操作远多于修改操作的并发场景中。

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

在构建高可靠的应用程序时，有界队列是一种强大的资源管理工具：它们能抑制并防止产生过多的工作项，使应用程序在负荷过载的情况下变得更加健壮。

BlockingQueue有多种实现：LinkedBlockingQueue和ArrayBlockingQueue是FIFO队列，与LinkedList和ArrayList相似，但比同步list有更好的并发性能。PriorityBlockingQueue是一个按优先级排序的队列，当你希望按照某种顺序而不是FIFO来处理元素时，这个队列非常有用，PriorityBlockingQueue既可以根据元素的自然顺序来比较元素，也可以使用Comparator来比较。最后一个BlockingQueue是SynchronousQueue，它并不是一个真正的队列，因为它不会为队列中元素维护存储空间。它维护的是一组线程，这些线程在等待着把元素加入或移出队列。以洗盘子为例，相当于没有盘架，直接将洗好的盘子放入下一个空闲的烘干机中，它可以直接交付工作，从而降低了将数据从生产者移动到消费者的延迟。因为SynchronousQueue没有存储功能，因此put和take会一直阻塞，直到有另一个线程已经准备好参与到交付过程中。仅当有足够多的消费者，并且总有一个消费者准备好获取交付的工作时，才适合使用同步队列。

1，串行线程封闭（serial thread confinement）

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

对于可变对象，生产者-消费者这种设计与阻塞队列组合在一起使得把对象从生产者转移给消费者变得容易。线程封闭对象只能由单个线程拥有，但可以通过安全地发布该对象来转移所有权。在所有权转移后，就只有新线程能获得这个对象的访问权限，并且发布对象的线程不会再访问它。这种安全的发布确保了对象状态对于新的所有者来说是可见的，并且由于最初的所有者不会再访问它，所以这个对象又被封闭在新的线程中，新线程可以对该对象做任意修改，因为它具有独占的访问权。

对象池利用了串行线程封闭，将对象借给一个请求线程。只要对象池包含足够的内部同步来安全地发布池中的对象，并且只要客户代码本身不会发布池中的对象，或者在江对象返回给对象池后就不再使用它，那么就可以安全地在线程之间传递所有权。

2，双端队列与work stealing

Java 6增加了两种容器类型，Deque&BlockingQueue。Deque是一个双端队列，实现了在队列头和队列尾的高效插入和移除。

BlockingQueue适用于生产者-消费者模式，双端队列适用于Work Stealing（工作密取）。在work stealing中，每个消费者有各自的双端队列。如果一个消费者完成了自己双端队列中的全部工作，那么它可以从其它消费者双端队列末尾steal 工作。密取工作模式比传统的生产者-消费者模式具有更高的可伸缩性，这是因为工作者线程不会在单个共享的任务队列上发生竞争。在大多数时候，它们都只是访问自己的双端队列，从而极大地减少了竞争。当工作者线程需要访问另一个队列时，它会从队列的尾部而不是头部获取工作，因此进一步降低了队列上的竞争程度。

 工作密取非常适合既是消费者又是生产者问题--当执行某个工作时可能导致出现更多的工作。例如，当一个工作线程找到新的任务单元时，它会将其放到自己队列的末尾（或者在工作共享设计模式中，放入其他工作者线程的队列中）。当双端队列为空时，它会在另一个线程队列队为查找新的任务，从而保证每个线程都保持忙碌状态。

四，阻塞方法和中断方法

线程可能会阻塞或暂停执行，原因有多种：等待I/O操作结束，等待获得一个锁，等待从Thread.sleep方法中醒来，或是等待另一个线程的计算结果。当线程阻塞时，它通常被挂起，并处于某种阻塞状态（BLOCKED/WATING/TIME_WAITING）。

阻塞操作与执行时间很长的普通操作的差别在于，被阻塞的线程必须等待某个不受它控制的事件发生后才能继续执行，例如等待I/O操作完成，等待某个锁变成可用，或者等待外部计算的结束。当某个外部事件发生时，线程被置回runnable状态，并可以再次被调度执行。

当某个方法会抛出InterruptedException时，表示该方法是一个阻塞方法，如果这个方法被中断，那么它将努力提前结束阻塞状态，以达到中断执行的目的。

Thread提供了interrupt方法，用于中断线程或者查询线程是否已经被中断，每个线程都有一个boolean类型的属性，表示线程的中断状态，interrupt方法会设置这个属性。

中断是一种协作机制。一个线程不能强制其它线程停止正在执行的操作而去执行其他的操作。当线程A中断线程B，A只是请求B在某个可以暂停的地方停止正在执行的操作--前提是线程B自己愿意停止。最常使用中断的情况就是取消某个操作。

当在代码中调用了一个将抛出InterruptedException的方法时，你自己的方法也就变成了一个阻塞方法，并且必须要对中断做出处理。对于library code来说，有两个选择：

• Propagate the InterruptedException. 这是最明智的策略，只需要把InterruptedException传递给方法的调用者，根本不捕获这个异常
• Restore the interrupt. 有时候我们不能抛出InterruptedException，比如code是放在runnable的run方法中。在这种情况下，就只能捕获InterruptedException并调用interrupt方法来执行当前线程的中断操作，这样在调用栈中更高层的代码将看到这个中断。例如：

• public class TaskRunnable implements Runnable {
      BlockingQueue<Task> queue;
      ...
      public void run() {
          try {
              processTask(queue.take());
          } catch (InterruptedException e) {
              Thread.currentThread().interrupt();
          }
      }
  }

  在出现InterruptedException时不应该做的事情是，捕获它但不做出任何响应。这将使调用栈上更高层的代码无法对中断采取处理措施，因为线程被中断的证据丢失了。

五，同步器

同步器可以是任何一个对象，只要它根据其自身的状态来协调线程的控制流就可以叫同步器。阻塞队列可以作为同步器，其他类型的同步器还包括信号量(Semaphore)/栅栏（Barrier）以及闭锁（Latch）。

所有的同步器都包含一些特定的结构化属性：它们封装了一些状态，这些状态将决定使用同步器的线程是继续执行还是等待，此外还提供了一些方法对状态进行操作，以及另一些方法用于高效地等待同步器进入到预期状态。

1，Latches

闭锁时一种同步器，可以延迟线程的进度直到线程到达终止状态。闭锁的作用相当于一扇门：在闭锁到达terminal状态前，这扇门一直是关闭的，没有任何线程通过，而当到达terminal状态时，这扇门就会打开允许所有线程通过。当闭锁达到terminal状态，它的状态就不会再改变，因此这扇门会永远打开。闭锁可以用来确保某些活动直到其他活动都完成后才继续执行。例如：

• 确保某个计算在其需要的所有资源都初始化之后才继续执行。
• 确保某个服务在其依赖的所有其他服务都已经启动之后才启动。
• 等待某个操作的所有参与者都就绪再继续执行。

CountDownLatch是一种灵活的闭锁，可以在上述各种情况中使用，它可以使一个或多个线程等待一组事件发生。countDown方法递减计数器，表示有一个事件已经发生了，而await方法等待计数器达到零，这表示所有需要等待的事件都已经发生。如果计数器的值非零，那么await会一直阻塞直到计数器为零，或者等待中的线程中断，或者等待超时。例如：

package com.ivy.thread;

import java.util.concurrent.CountDownLatch;

public class TestHarness {

    public 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() {
                    try {
                        startGate.await();
                        try {
                            task.run();
                        } finally {
                            endGate.countDown();
                        }
                    } catch (InterruptedException ignored) {}
                }
            };
            t.start();
        }
        long start = System.nanoTime();
        startGate.countDown();
        endGate.await();
        long end = System.nanoTime();
        return end-start;
    }
    
    public static void main(String[] args) {
        Runnable task = new Runnable() {
            
            @Override
            public void run() {
                System.out.println("haha");
                
            }
        };
        
        try {
            long interval = new TestHarness().timeTasks(3, task);
            System.out.println(interval);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}

TestHarness创建了nThreads数量的线程来执行task，它用了两个锁，startGate和endGate，startGate初始化为1，endGate初始化为nThreads。先是主线程开始计时，每个线程需要执行的第一件事是startGate.await()，等待startGate状态变为0，也就是等待startGate.countDown()执行后这个线程开始启动，这样确保所有线程在同一时刻启动。而每个线程要做的最后一件事是将调用endGate的countDown()减1，最后endGate.await()执行，等待endGate状态转为0，这样确保所有线程执行完成后主线程结束计时。

为什么要在TestHarness中使用闭锁而不是在线程创建后就立即启动呢？可能是想要测试n各线程并发执行某个任务时需要的时间。启动门将使得主线程能够同时释放所有执行task的线程，而结束门则使主线程能够等待最后一个线程执行完成，而不是顺序地等待每个线程执行完成。

2，Semaphore[ˈsɛməˌfɔr, -ˌfor]信号量

 计数信号量用来控制同时访问某个特定资源的操作数量，或者同时执行某个指定操作的数量。计数信号量还可以用来实现某种资源池，或者对容器施加边界。

Semaphore管理着一组虚拟的permits，许可的初始数量通过构造函数指定。在执行操作前先acquire permits（只要有剩余的许可就可以），在使用完后会release这个许可。如果没有获得permit，acquire方法将一直阻塞到有许可或指导被中断或超时。release方法将返回一个permit给信号量。

计数信号量的一种简化形式是binary信号量，即初始值是1的Semaphore，它可以用来做互斥体（mutext），并具备不可重入的特性：谁拥有这个唯一的许可，谁就拥有了互斥锁。

Semaphore可以用于实现资源池，例如数据库连接池。将Semaphore的计数值初始化为池的大小，并从池中获取一个资源前线调用acquire()获取一个许可，在将资源返回给池后调用release释放许可，那么acquire将一直阻塞直到资源池不为空。

Semaphore也可以将任何一种容器变成有界阻塞容器。信号量的计数值会初始化为容器容量的最大值，add操作在向容器添加一个元素之前，首先获取一个permit，然后再添加，如果添加失败，那么会释放许可，如果成功就不释放了。同样，remove操作会释放一个许可，来使更多的元素能够添加到容器中。代码如下：

package com.ivy.thread;

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.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);
    }
    
    public boolean add(T o) throws InterruptedException {
        sem.acquire();
        boolean wasAdded = false;
        try {
            wasAdded = set.add(o);
        } finally {
            if (!wasAdded) {
                sem.release();
            }
        }
        return wasAdded;
    }
    
    public boolean remove(Object o) {
        boolean wasRemoved = set.remove(o);
        if (wasRemoved) {
            sem.release();
        }
        return wasRemoved;
    }
}

3,栅栏Barrier

闭锁可以启动一组相关的操作，或者等待一组相关的操作结束。闭锁时一次性对象，一旦进入终止状态，就不能被重置。

栅栏类似于闭锁，它能阻塞一组线程直到某个事件发生。栅栏与闭锁的关键区别在于，所有线程必须同时到达栅栏位置，才能继续执行。闭锁用于等待事件发生，而栅栏用于等待其他线程。

1， 创建一个CyclicBarrier：

CyclicBarrier barrier = new CyclicBarrier()2;

需要等待两个线程

2， 在CyclicBarrier上等待：

barrier.await();

也可以为等待线程设置一个timeout，当timeout时间到了栅栏就打开了而不用等到N个threads都到达CyclicBarrier。

barrier.await(10, TimeUnit.SECONDS);

所有的线程会一直等在栅栏外，直到：

　　1， 最后一个线程到达

　　2， 当前线程被其他线程中断

　　3，其它等待线程被中断

　　4，其它等待线程timeout

　　5，CyclicBarrier.reset()被其他外部线程调用。

3，CyclicBarrier Action

CyclicBarrier支持一个栅栏操作，这个action是一个Runnable实现，一旦最后一个线程到达，这个action就会被调用。通过CyclicBarrier构造函数将action传递给Barrier：

Runnable barrierAction = new Runnable() {
    public void run() {
          ...  
    }
}

CyclicBarrier barrier = new CyclicBarrier(2, barrierAction);    

示例：

package com.ivy.thread;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

class CyclicBarrierWork implements Runnable {

    private int id;
    private CyclicBarrier barrier;
    
    public CyclicBarrierWork(int id, final CyclicBarrier barrier) {
        this.id = id;
        this.barrier = barrier;
    }
    
    @Override
    public void run() {
        // TODO Auto-generated method stub
            
            try {
                System.out.println(id + " Thread");
                barrier.await();
            } catch (InterruptedException|BrokenBarrierException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            } 
    }
    
}
public class TestCyclicBarrier {

    public static void main(String[] args) {
        int num = 10;
        CyclicBarrier barrier = new CyclicBarrier(num, new Runnable() {
            
            @Override
            public void run() {
                System.out.println("finished");
                
            }
        });

        for(int i=0;i<num;i++) {
            new Thread(new CyclicBarrierWork(i, barrier)).start();
        }
    }

}

10个线程在运行，约定所有线程运行完了之后才能继续。

结果如下：

0 Thread
1 Thread
2 Thread
5 Thread
4 Thread
3 Thread
7 Thread
8 Thread
6 Thread
9 Thread
finished

如果把for循环的num改为5，barrier会一直await，结果如下：

0 Thread
3 Thread
2 Thread
1 Thread
4 Thread

如果for循环的num改为12，barrier在10个线程完成后就不会再await了，结果如下：

0 Thread
2 Thread
3 Thread
1 Thread
4 Thread
5 Thread
6 Thread
7 Thread
9 Thread
8 Thread
finished
10 Thread
11 Thread