你正在使用的浏览器版本过低，将不能正常浏览和使用知乎。java相关（10）
书的话，参考java并发编程实战。
博客可以参考这个
关于什么是thread safe：
根据《Java Concurrency in Practice》的定义，一个线程安全的 class 应当满足以下三个条件：o 多个线程同时访问时，其表现出正确的行为。o 无论操作系统如何调度这些线程， 无论这些线程的执行顺序如何交织（interleaving）。o 调用端代码无须额外的同步或其他协调动作。
首先我们要定义线程安全，我比较认可的是在《Java concurrency in practice》一书中的定义：一个不论运行时（Runtime）如何调度线程都不需要调用方提供额外的同步和协调机制还能正确地运行的类是线程安全的多线程的场景很多很复杂，难以穷尽地说那些条件下是或者不是线程安全的，但是有一些常用的肯定线程安全的场景：1.无状态的一定是线程安全的。这个很好理解，因为所谓线程不安全也就是一个线程修改了状态，而另一个线程的操作依赖于这个被修改的状态。2.只有一个状态，而且这个状态是由一个线程安全的对象维护的，那这个类也是线程安全的。比如你在数据结构里只用一个AtomicLong来作为计数器，那递增计数的操作都是线程安全的，不会漏掉任何一次计数，而如果你用普通的long做++操作则不一样，因为++操作本身涉及到取数、递增、赋值 三个操作，某个线程可能取到了另外一个线程还没来得及写回的数就会导致上一次写入丢失。3.有多个状态的情况下，维持不变性（invariant）的所有可变（mutable）状态都用同一个锁来守护的类是线程安全的。这一段有些拗口，首先类不变性的意思是指这个类在多线程状态下能正确运行的状态，其次用锁守护的意思是所有对该状态的操作都需要获取这个锁，而用同一个锁守护的作用就是所有对这些状态的修改实际最后都是串行的，不会存在某个操作中间状态被其他操作可见，继而导致线程不安全。所以这里的关键在于如何确定不变性，可能你的类的某些状态对于类的正确运行是无关紧要的，那就不需要用和其他状态一样的锁来守护。因此我们常可以看到有的类里面会创建一个新的对象作为锁来守护某些和原类本身不变性无关的状态。上面这三种只是一种归纳，具体到实际应用时，要看你的类哪些状态是必须用锁来守护的，灵活变通。
一、关于java.util.concurrent下面的类
1.BlockingQueue
The Java BlockingQueue interface in the java.util.concurrent package represents a queue which is thread safe to put into, and take instances from.
典型场景：
A BlockingQueue with one thread putting into it, and another thread taking from it.
该类中定义了四种方式的方法：
The 4 different sets of behaviour means this:
1.Throws Exception:If the attempted operation is not possible immediately, an exception is thrown.
2.Special Value:If the attempted operation is not possible immediately, a special value is returned (often true / false).
3.Blocks:If the attempted operation is not possible immedidately, the method call blocks until it is.
4.Times Out:If the attempted operation is not possible immedidately, the method call blocks until it is, but waits no longer than the given timeout. Returns a special value telling whether the operation succeeded or not (typically true / false).
并且不能在BlockingQueue中添加null，不然汇报空指针异常。
BlockingQueue为接口，下面是几种实现：
1.ArrayBlockingQueue
2.DelayQueue
3.LinkedBlockingQueue
4.PriorityBlockingQueue
5.SynchronousQueue
关于BlockingQueue的例子，就是上图的producer和consumer的一个实现：
public class BlockingQueueExample {
public static void main(String[] args) throws Exception {
BlockingQueue queue = new ArrayBlockingQueue(1024);
Producer producer = new Producer(queue);
Consumer consumer = new Consumer(queue);
new Thread(producer).start();
new Thread(consumer).start();
Thread.sleep(4000);
producer：
public class Producer implements Runnable{
protected BlockingQueue queue = null;
public Producer(BlockingQueue queue) {
this.queue =
public void run() {
queue.put("1");
Thread.sleep(1000);
queue.put("2");
Thread.sleep(1000);
queue.put("3");
} catch (InterruptedException e) {
e.printStackTrace();
consumer：
public class Consumer implements Runnable{
protected BlockingQueue queue = null;
public Consumer(BlockingQueue queue) {
this.queue =
public void run() {
System.out.println(queue.take());
System.out.println(queue.take());
System.out.println(queue.take());
} catch (InterruptedException e) {
e.printStackTrace();
那么BlockingQueue内部是怎么实现的呢？
public class BlockingQueue {
private List queue = new LinkedList();
private int
limit = 10;
public BlockingQueue(int limit){
this.limit =
public synchronized void enqueue(Object item)
throws InterruptedException
while(this.queue.size() == this.limit) {
if(this.queue.size() == 0) {
notifyAll();
this.queue.add(item);
public synchronized Object dequeue()
throws InterruptedException{
while(this.queue.size() == 0){
if(this.queue.size() == this.limit){
notifyAll();
return this.queue.remove(0);
注意在notifyAll();只有在queue空或者满的时候才调用，因为只有这两种情况下才会产生等待。
用synchronized关键字和wait，notify性能会有影响，看看ArrayBlockingQueue中如何来实现的。
1.ArrayBlockingQueue
The ArrayBlockingQueue stores the elements internally in FIFO (First In, First Out) order. The head of the queue is the element which has been in queue the longest time, and the tail of the queue is the element which has been in the queue the shortest time.
2.DelayQueue
The DelayQueue blocks the elements internally until a certain delay has expired. The elements must implement the interface java.util.concurrent.Delayed. Here is how the interface looks: 放进去的元素必须实现Delayed接口
它的内部实现使用了priorityqueue。
不清楚Delayed接口该如何实现？看看：
package com.
import java.util.concurrent.D
import java.util.concurrent.TimeU
public class DelayObject implements Delayed {
private long startT
public DelayObject(String data, long delay) {
this.data =
this.startTime = System.currentTimeMillis() +
public long getDelay(TimeUnit unit) {
long diff = startTime - System.currentTimeMillis();
return unit.convert(diff, TimeUnit.MILLISECONDS);
public int compareTo(Delayed o) {
if (this.startTime & ((DelayObject) o).startTime) {
return -1;
if (this.startTime & ((DelayObject) o).startTime) {
public String toString() {
return "{" +
"data='" + data + '\'' +
", startTime=" + startTime +
Delayqueue到底是干嘛的？
An unbounded blocking queue of Delayed elements, in which an element can only be taken when its delay has expired. The head of the queue is that Delayed element whose delay expired furthest in the past.只有queue里面的元素过期了，才能够被取走，head中存放的是最先过期的元素。可以用在连接释放中，可以给新加入的连接一个过期时间，然后到达过期时间后，就会被取走。
Expiration occurs when an element’s getDelay(TimeUnit.NANOSECONDS) method returns a value less than or equal to zero.
3.LinkedBlockingQueue
4.PriorityBlockingQueue
All elements inserted into the PriorityBlockingQueue must implement the parable interface. The elements thus order themselves according to whatever priority you decide in your Comparable implementation.
当然初始化时传入comparator也是可以的。
notice, that in case you obtain an Iterator from a PriorityBlockingQueue, the Iterator does not guarantee to iterate the elements in priority order. 我擦，iterator居然不是按优先级排序的。
5.SynchronousQueue
The SynchronousQueue is a queue that can only contain a single element internally. A thread inseting an element into the queue is blocked until another thread takes that element from the queue. Likewise, if a thread tries to take an element and no element is currently present, that thread is blocked until a thread insert an element into the queue.
简单说，一次只能存在一个元素在queue中。
2.BlockingDeque
它的用处在哪里呢？
deque就是“double ended queue”的简称。
A BlockingDeque could be used if threads are both producing and consuming elements of the same queue. 两边都可以生产或者消费的情况。It could also just be used if the producting thread needs to insert at both ends of the queue, and the consuming thread needs to remove from both ends of the queue.或者生产或者消费需要添加到两头的情况 Here is an illustration of that:
A BlockingDeque - threads can put and take from both ends of the deque.
方法如下：
BlockingDeque Extends BlockingQueue，原来是继承自queue啊。跟deque继承自queue一模一样。
实现了BlockingDeque的类：
1.LinkedBlockingDeque
选一个PriorityBlockingQueue来看看内部是怎么实现的。
PriorityBlockingQueue和PriorityQueue最大的不同有两点：1.线程安全，这是通过锁实现的，2.支持阻塞操作，特性来自于接口BlockingQueue。
优先级一样的元素出队列顺序是没有保证的。PriorityBlockingQueue不接受null值，线程安全。
PriorityBlockingQueue的核心数据结构是数组+堆。默认是最小堆，可通过改变比较器实现最大堆。
其内部有两把锁，lock是公用锁，用来保证增删改等操作的线程安全性。allocationSpinLock是自旋锁，通过CAS更新，用于扩容。
入队列和出队列
add和put都调用了offer方法，如下：
涉及扩容和从位置ｋ将x元素上浮和下沉。
在最小堆中，添加元素后还要上浮，删除元素是将最小值与末尾值交换，删除末尾值，然后将最上层的值下沉。
public boolean add(E e) {
return offer(e);
public void put(E e) {
public boolean offer(E e) {
if (e == null)
throw new NullPointerException();
final ReentrantLock lock = this.
lock.lock();
while ((n = size) &= (cap = (array = queue).length))
tryGrow(array, cap);
Comparator&? super E& cmp =
if (cmp == null)
siftUpComparable(n, e, array);
siftUpUsingComparator(n, e, array, cmp);
size = n + 1;
notEmpty.signal();
} finally {
lock.unlock();
return true;
private static &T& void siftUpComparable(int k, T x, Object[] array) {
Comparable&? super T& key = (Comparable&? super T&)
while (k & 0) {
int parent = (k - 1) &&& 1;
Object e = array[parent];
if (pareTo((T) e) &= 0)
array[k] =
array[k] =
public E remove() {
E x = poll();
if (x != null)
throw new NoSuchElementException();
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
return dequeue();
} finally {
lock.unlock();
private E dequeue() {
int n = size - 1;
if (n & 0)
return null;
Object[] array =
E result = (E) array[0];
E x = (E) array[n];
array[n] = null;
Comparator&? super E& cmp =
if (cmp == null)
siftDownComparable(0, x, array, n);
siftDownUsingComparator(0, x, array, n, cmp);
private static &T& void siftDownComparable(int k, T x, Object[] array,
if (n & 0) {
Comparable&? super T& key = (Comparable&? super T&)x;
int half = n &&& 1;
while (k & half) {
int child = (k && 1) + 1;
Object c = array[child];
int right = child + 1;
if (right & n &&
((Comparable&? super T&) c).compareTo((T) array[right]) & 0)
c = array[child = right];
if (pareTo((T) c) &= 0)
array[k] =
array[k] =
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
while ( (result = dequeue()) == null)
notEmpty.await();
} finally {
lock.unlock();
因为该queue是无界的，意味着增加元素没有阻塞，而删除肯定有啦，size最小值为０嘛。
看Condition.awiat方法的解释：
Causes the current thread to wait until it is signalled or
interrupted
通过put操作的notEmpty.signal()和take操作的notEmpty.await(),控制了线程阻塞，而不用死循环遍历询问。
private void tryGrow(Object[] array, int oldCap) {
lock.unlock();
Object[] newArray = null;
if (allocationSpinLock == 0 &&
pareAndSwapInt(this, allocationSpinLockOffset,
int newCap = oldCap + ((oldCap & 64) ?
(oldCap + 2) :
(oldCap && 1));
if (newCap - MAX_ARRAY_SIZE & 0) {
int minCap = oldCap + 1;
if (minCap & 0 || minCap & MAX_ARRAY_SIZE)
throw new OutOfMemoryError();
newCap = MAX_ARRAY_SIZE;
if (newCap & oldCap && queue == array)
newArray = new Object[newCap];
} finally {
allocationSpinLock = 0;
if (newArray == null)
Thread.yield();
lock.lock();
if (newArray != null && queue == array) {
queue = newA
System.arraycopy(array, 0, newArray, 0, oldCap);
说明：当数组空间满，进行扩容。如果容量小于64则两倍扩容，否则1.5倍扩容。这里有数组的复制动作。这里使用了CAS更新allocationSpinLock，确保只有一个线程会执行扩容操作。这里要弄明白为什么要先放弃锁在获取锁？这是因为有更好的并发特性，在扩容的过程中不影响其他线程并发执行take和poll操作，甚至是offer操作，通过自旋锁使线程安全的进行扩容工作。
public E element() {
E x = peek();
if (x != null)
throw new NoSuchElementException();
public E peek() {
final ReentrantLock lock = this.lock;
lock.lock();
return (size == 0) ? null : (E) queue[0];
} finally {
lock.unlock();
为什么读要加锁呢？
读加锁是因为防止读到正在插入或删除而导致队列元素还在调整的错误数据。
PriorityBlockingQueue是一种按照优先级顺序线程安全的阻塞的队列实现。遍历的结果也没有顺序，可以通过Arrays.sort(queue.toArray(),comparator)达到有序遍历的目的。可以在多线程环境下替代PriorityQueue。
3.ConcurrentMap
１．ConcurrentHashMap
4.ConcurrentNavigableMap
The java.util.concurrent.ConcurrentNavigableMap class is a java.util.NavigableMap with support for concurrent access, and which has concurrent access enabled for its submaps. The “submaps” are the maps returned by various methods like headMap(), subMap() and tailMap().对于他的submap的访问是线程安全的。
几个方法：
１.headMap()
The headMap(T toKey) method returns a view of the map containing the keys which are strictly less than the given key.严格小于
If you make changes to the original map, these changes are reflected in the head map. 重点！如果更改原map，那么submap也会被改变。
ConcurrentNavigableMap map = new ConcurrentSkipListMap();
map.put("1", "one");
map.put("2", "two");
map.put("3", "three");
ConcurrentNavigableMap headMap = map.headMap("2");
将会返回“１”对应的结果。
2.tailMap()
The tailMap(T fromKey) method returns a view of the map containing the keys which are greater than or equal to the given fromKey. 大于等于
如果更改原map，那么submap也会被改变。
3.subMap()
The subMap() method returns a view of the original map which contains all keys from (including), to (excluding) two keys given as parameters to the method.
如果更改原map，那么submap也会被改变。
5.CountDownLatch
A java.util.concurrent.CountDownLatch is a concurrency construct that allows one or more threads to wait for a given set of operations to complete.
A CountDownLatch is initialized with a given count. This count is decremented by calls to the countDown() method. Threads waiting for this count to reach zero can call one of the await() methods. Calling await() blocks the thread until the count reaches zero.
内部会有一个count，每次调用countDown方法都会减小，到０时等待的线程就可以工作了。
CountDownLatch latch = new CountDownLatch(3);
= new Waiter(latch);
Decrementer decrementer = new Decrementer(latch);
new Thread(waiter)
new Thread(decrementer).start();
Thread.sleep(4000);
public class Waiter implements Runnable{
CountDownLatch latch = null;
public Waiter(CountDownLatch latch) {
this.latch =
public void run() {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
System.out.println("Waiter Released");
Decrementer:
public class Decrementer implements Runnable {
CountDownLatch latch = null;
public Decrementer(CountDownLatch latch) {
this.latch =
public void run() {
Thread.sleep(1000);
this.latch.countDown();
Thread.sleep(1000);
this.latch.countDown();
Thread.sleep(1000);
this.latch.countDown();
} catch (InterruptedException e) {
e.printStackTrace();
6.CyclicBarrier
The java.util.concurrent.CyclicBarrier class is a synchronization mechanism that can synchronize threads progressing through some algorithm. In other words, it is a barrier that all threads must wait at, until all threads reach it, before any of the threads can continue. Here is a diagram illustrating that:
Two threads waiting for each other at CyclicBarriers.
主要步骤：
1.Creating a CyclicBarrier
When you create a CyclicBarrier you specify how many threads are to wait at it, before releasing them. Here is how you create a CyclicBarrier: CyclicBarrier barrier = new CyclicBarrier(2);
2.Waiting at a CyclicBarrier
3.CyclicBarrier Action
The CyclicBarrier supports a barrier action, which is a Runnable that is executed once the last thread arrives. You pass the Runnable barrier action to the CyclicBarrier in its constructor, like this:
就是大家都到了的情况下，执行什么动作。
barrierAction = ... ;
CyclicBarrier barrier
= new CyclicBarrier(2, barrierAction);
4.CyclicBarrier Example
Runnable barrier1Action = new Runnable() {
public void run() {
System.out.println("BarrierAction 1 executed ");
Runnable barrier2Action = new Runnable() {
public void run() {
System.out.println("BarrierAction 2 executed ");
CyclicBarrier barrier1 = new CyclicBarrier(2, barrier1Action);
CyclicBarrier barrier2 = new CyclicBarrier(2, barrier2Action);
CyclicBarrierRunnable barrierRunnable1 =
new CyclicBarrierRunnable(barrier1, barrier2);
CyclicBarrierRunnable barrierRunnable2 =
new CyclicBarrierRunnable(barrier1, barrier2);
new Thread(barrierRunnable1).start();
new Thread(barrierRunnable2).start();
CyclicBarrierRunnable：
public class CyclicBarrierRunnable implements Runnable{
CyclicBarrier barrier1 = null;
CyclicBarrier barrier2 = null;
public CyclicBarrierRunnable(
CyclicBarrier barrier1,
CyclicBarrier barrier2) {
this.barrier1 = barrier1;
this.barrier2 = barrier2;
public void run() {
Thread.sleep(1000);
System.out.println(Thread.currentThread().getName() +
" waiting at barrier 1");
this.barrier1.await();
Thread.sleep(1000);
System.out.println(Thread.currentThread().getName() +
" waiting at barrier 2");
this.barrier2.await();
System.out.println(Thread.currentThread().getName() +
" done!");
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
Thread-0 waiting at barrier 1
Thread-1 waiting at barrier 1
BarrierAction 1 executed
Thread-1 waiting at barrier 2
Thread-0 waiting at barrier 2
BarrierAction 2 executed
Thread-0 done!
Thread-1 done!
7.Exchanger
The java.util.concurrent.Exchanger class represents a kind of rendezvous point(约会地点) where two threads can exchange objects. Here is an illustration of this mechanism:
Two threads exchanging objects via an Exchanger.
Exchanger exchanger = new Exchanger();
ExchangerRunnable exchangerRunnable1 =
new ExchangerRunnable(exchanger, "A");
ExchangerRunnable exchangerRunnable2 =
new ExchangerRunnable(exchanger, "B");
new Thread(exchangerRunnable1).start();
new Thread(exchangerRunnable2).start();
ExchangerRunnable：
public class ExchangerRunnable implements Runnable{
Exchanger exchanger = null;
public ExchangerRunnable(Exchanger exchanger, Object object) {
this.exchanger =
this.object =
public void run() {
Object previous = this.
this.object = this.exchanger.exchange(this.object);
System.out.println(
Thread.currentThread().getName() +
" exchanged " + previous + " for " + this.object
} catch (InterruptedException e) {
e.printStackTrace();
Thread-0 exchanged A for B
Thread-1 exchanged B for A
8.Semaphore
先看关于Semaphores的介绍。
Semaphores
A Semaphore is a thread synchronization construct that can be used either to send signals between threads to avoid missed signals, or to guard a critical section like you would with a lock.既可以用来同步，也可以用来锁。
1.Simple Semaphore
public class Semaphore {
private boolean signal = false;
public synchronized void take() {
this.signal = true;
this.notify();
public synchronized void release() throws InterruptedException{
while(!this.signal) wait();
this.signal = false;
2.Using Semaphores for Signaling
最常见的还是线程之间的take和release：
Semaphore semaphore = new Semaphore();
SendingThread sender = new SendingThread(semaphore);
ReceivingThread receiver = new ReceivingThread(semaphore);
receiver.start();
sender.start();
SendingThread：
public class SendingThread {
Semaphore semaphore = null;
public SendingThread(Semaphore semaphore){
this.semaphore =
public void run(){
while(true){
this.semaphore.take();
RecevingThread：
public class RecevingThread {
Semaphore semaphore = null;
public ReceivingThread(Semaphore semaphore){
this.semaphore =
public void run(){
while(true){
this.semaphore.release();
3.Counting Semaphore
不再只有true与false两种选择：
public class CountingSemaphore {
private int signals = 0;
public synchronized void take() {
this.signals++;
this.notify();
public synchronized void release() throws InterruptedException{
while(this.signals == 0) wait();
this.signals--;
4.Bounded Semaphore
增加了界限：
public class BoundedSemaphore {
private int signals = 0;
private int bound
public BoundedSemaphore(int upperBound){
this.bound = upperB
public synchronized void take() throws InterruptedException{
while(this.signals == bound) wait();
this.signals++;
this.notify();
public synchronized void release() throws InterruptedException{
while(this.signals == 0) wait();
this.signals--;
this.notify();
5.Using Semaphores as Locks
作为锁来用
It is possible to use a bounded semaphore as a lock. To do so, 1.set the upper bound to 1, and have the 2.call to take() and release() guard the critical section. Here is an example:
BoundedSemaphore semaphore = new BoundedSemaphore(1);
semaphore.take();
//critical section
} finally {
semaphore.release();
You can also use a bounded semaphore to limit the number of threads allowed into a section of code.
如果将数量设置为5，5 threads would be allowed to enter the critical section at a time.
回到正题，java.util.concurrent.Semaphore是一个counting semaphore。意味着拥有两个主要方法：
1.acquire()
2.release()
Semaphore Usage
鉴于信号量的主要作用是：
1.To guard a critical section against entry by more than N threads at a time.
2.To send signals between two threads.
1.Guarding Critical Sections
Semaphore semaphore = new Semaphore(1);
//critical section
semaphore.acquire();
semaphore.release();
2.Sending Signals Between Threads
If you use a semaphore to send signals between threads, then you would typically have one thread call the acquire() method, and the other thread to call the release() method.
传递信号的通常写法一般是一个线程acquire，另一个线程release。
Thus it is possible to coordinate threads. For instance, if acquire was called after Thread 1 had inserted an object in a shared list, and Thread 2 had called release() just before taking an object from that list, you had essentially created a blocking queue（仔细理解，实际上潜在地创建了一个blockingqueue）. The number of permits available in the semaphore would correspond to the maximum number of elements the blocking queue could hold（permits的个数对应了blockingqueue的最大数量）.
3.Fairness
No guarantees are made about fairness of the threads acquiring permits from the Semaphore. That is, there is no guarantee that the first thread to call acquire() is also the first thread to obtain a permit. If the first thread is blocked waiting for a permit, then a second thread checking for a permit just as a permit is released, may actually obtain the permit ahead of thread 1.
不保证先acquire的一定先拿到release后的permit。
当然，可以强行设置，但会牺牲性能。
Semaphore semaphore = new Semaphore(1, true);
9.ExecutorService
An ExecutorService is thus very similar to a thread pool （差不多）. In fact, the implementation of ExecutorService present in the java.util.concurrent package is a thread pool implementation.
ExecutorService executorService = Executors.newFixedThreadPool(10);
executorService.execute(new Runnable() {
public void run() {
System.out.println("Asynchronous task");
executorService.shutdown();
an anonymous implementation of the Runnable interface is passed to the execute() method. This causes the Runnable to be executed by one of the threads in the ExecutorService.
Task Delegation
A thread delegating a task to an ExecutorService for asynchronous execution.
实现该接口的类：
1.ThreadPoolExecutor
2.ScheduledThreadPoolExecutor
public class Test2 {
public static void main(String[] args){
ExecutorService service1 = Executors.newSingleThreadExecutor();
ExecutorService service2 = Executors.newFixedThreadPool(10);
ExecutorService service3 = Executors.newScheduledThreadPool(10);
几种常用方法：
1.execute(Runnable)
ExecutorService executorService = Executors.newSingleThreadExecutor();
executorService.execute(new Runnable() {
public void run() {
System.out.println("Asynchronous task");
executorService.shutdown();
没有办法拿到执行后的返回结果
2.submit(Runnable)
Future future = executorService.submit(new Runnable() {
public void run() {
System.out.println("Asynchronous task");
future.get();
3.submit(Callable)
Future future = executorService.submit(new Callable(){
public Object call() throws Exception {
System.out.println("Asynchronous Callable");
return "Callable Result";
System.out.println("future.get() = " + future.get());
4.invokeAny()
The invokeAny() method takes a collection of Callable objects, or subinterfaces of Callable. Invoking this method does not return a Future（不反悔Future对象）, but returns the result of one of the Callable objects. You have no guarantee about which of the Callable’s results you get. Just one of the ones that finish.拿到的结果是最先完成的结果
If one of the tasks complete (or throws an exception), the rest of the Callable’s are cancelled. 一人完成，全部cancel
ExecutorService executorService = Executors.newSingleThreadExecutor();
Set&Callable&String&& callables = new HashSet&Callable&String&&();
callables.add(new Callable&String&() {
public String call() throws Exception {
return "Task 1";
callables.add(new Callable&String&() {
public String call() throws Exception {
return "Task 2";
callables.add(new Callable&String&() {
public String call() throws Exception {
return "Task 3";
String result = executorService.invokeAny(callables);
System.out.println("result = " + result);
executorService.shutdown();
5.invokeAll()
The invokeAll() method invokes all of the Callable objects you pass to it in the collection passed as parameter. The invokeAll() returns a list of Future objects via which you can obtain the results of the executions of each Callable.
ExecutorService executorService = Executors.newSingleThreadExecutor();
Set&Callable&String&& callables = new HashSet&Callable&String&&();
callables.add(new Callable&String&() {
public String call() throws Exception {
return "Task 1";
callables.add(new Callable&String&() {
public String call() throws Exception {
return "Task 2";
callables.add(new Callable&String&() {
public String call() throws Exception {
return "Task 3";
List&Future&String&& futures = executorService.invokeAll(callables);
for(Future&String& future : futures){
System.out.println("future.get = " + future.get());
executorService.shutdown();
ThreadPoolExecutor
上面3种方式中的前两个其实内部就是返回的这个对象。
The number of threads in the pool is determined by these variables:
1.corePoolSize
2.maximumPoolSize
If less than corePoolSize threads are created in the the thread pool when a task is delegated to the thread pool, then a new thread is created, even if idle threads exist in the pool.意思是尽量有corePoolSize数量的线程在运行
If the internal queue of tasks is full, and corePoolSize threads or more are running, but less than maximumPoolSize threads are running, then a new thread is created to execute the task.针对于queue full的情况
10.ScheduledExecutorService
实现的类：
1.ScheduledThreadPoolExecutor（上面3种方式中的第三个其实内部就是返回的这个对象。）
主要方法：
1.schedule (Callable task, long delay, TimeUnit timeunit)
2.schedule (Runnable task, long delay, TimeUnit timeunit)
3.scheduleAtFixedRate (Runnable, long initialDelay, long period, TimeUnit timeunit)
4.scheduleWithFixedDelay (Runnable, long initialDelay, long period, TimeUnit timeunit)
先看看Locks的简介吧。
Locks in Java
A lock is a thread synchronization mechanism like synchronized blocks except locks can be more sophisticated than Java’s synchronized blocks.
1.A Simple Lock
public class Counter{
private int count = 0;
public int inc(){
synchronized(this){
当然用Lock可以这样写：
public class Counter{
private Lock lock = new Lock();
private int count = 0;
public int inc(){
lock.lock();
int newCount = ++
lock.unlock();
return newC
那么Lock内部又是怎么实现的呢？一种简单实现：
public class Lock{
private boolean isLocked = false;
public synchronized void lock()
throws InterruptedException{
while(isLocked){
isLocked = true;
public synchronized void unlock(){
isLocked = false;
Notice the while(isLocked) loop, which is also called a “spin lock”.
2.Lock Reentrance
Synchronized blocks in Java are reentrant（就是能够在Java 同步方法中切换）. This means, that if a Java thread enters a synchronized block of code, and thereby take the lock on the monitor object the block is synchronized on, the thread can enter other Java code blocks synchronized on the same monitor object. Here is an example:
public class Reentrant{
public synchronized outer(){
public synchronized inner(){
The lock implementation shown earlier is not reentrant（但是上面这种关于Lock的实现不是可重入的哦！）. If we rewrite the Reentrant class like below, the thread calling outer() will be blocked inside the lock.lock() in the inner() method. 下面将会出错：
public class Reentrant2{
Lock lock = new Lock();
public outer(){
lock.lock();
lock.unlock();
public synchronized inner(){
lock.lock();
lock.unlock();
A thread calling outer() will first lock the Lock instance. Then it will call inner(). Inside the inner() method the thread will again try to lock the Lock instance. This will fail (meaning the thread will be blocked), since the Lock instance was locked already in the outer() method.两次lock操作将会block。
我们做一点改动，就可以实现可重入了，上述Lock的实现之所以不可重入，是因为lock方法内部不能够辨别调用lock的线程和锁住的线程是不是同一个线程。
public class Lock{
boolean isLocked = false;
lockedBy = null;
lockedCount = 0;
public synchronized void lock()
throws InterruptedException{
Thread callingThread = Thread.currentThread();
while(isLocked && lockedBy != callingThread){
isLocked = true;
lockedCount++;
lockedBy = callingT
public synchronized void unlock(){
if(Thread.curentThread() == this.lockedBy){
lockedCount--;
if(lockedCount == 0){
isLocked = false;
如果不加入lockedCount，那么lock多次，但是unlock一次就可以解开。
现在该Lock类就可以重入了。
3.Lock Fairness
好了，回到正题。
简单例子：
Lock lock = new ReentrantLock();
lock.lock();
//critical section
lock.unlock();
实现Lock接口的类：
1.ReentrantLock
lock和synchronized的主要不同：
1.A synchronized block makes no guarantees about the sequence in which threads waiting to entering it are granted access. synchronized块对等待的线程进入的顺序没有保证
2.You cannot pass any parameters to the entry of a synchronized block. Thus, having a timeout trying to get access to a synchronized block is not possible. lock可设置等待时间
3.The synchronized block must be fully contained within a single method. A Lock can have it’s calls to lock() and unlock() in separate methods. lock可在一个方法中lock，另一个方法中unlock。
主要方法：
2.lockInterruptibly()
3.tryLock()
4.tryLock(long timeout, TimeUnit timeUnit)
5.unlock()
The lock() method locks the Lock instance if possible. If the Lock instance is already locked, the thread calling lock() is blocked until the Lock is unlocked. 阻塞式的lock
The tryLock() method attempts to lock the Lock instance immediately. It returns true if the locking succeeds, false if Lock is already locked. This method never blocks. 起一个检查的作用，不会阻塞
The unlock() method unlocks the Lock instance. Typically, a Lock implementation will only allow the thread that has locked the Lock to call this method. Other threads calling this method may result in an unchecked exception (RuntimeException). 只允许锁住该lock的线程调用unlock，否则会异常。
12.ReadWriteLock
单独的一个类，没有继承自Lock接口，注意。
A java.util.concurrent.locks.ReadWriteLock is an advanced thread lock mechanism. It allows multiple threads to read a certain resource, but only one to write it, at a time.能够同时让很多线程读，但只能同时让一个线程写。
The idea is, that multiple threads can read from a shared resource without causing concurrency errors. The concurrency errors first occur when 1.reads and writes to a shared resource occur concurrently, or if 2.multiple writes take place concurrently.错误发生情况
我们先来了解的并发理论。
它比前面的Locks要难一点。
Read / Write Lock Java Implementation
Read Access
If no threads are writing, and no threads have requested write access.读情况
Write Access
If no threads are reading or writing.写独占
如果不将写操作的优先级提高，很容易在读频繁的情况下，出现starvation的情况。这样写就要等到它们都完成了才能轮到它。
自己实现之：
public class ReadWriteLock{
private int readers
private int writers
private int writeRequests = 0;
public synchronized void lockRead() throws InterruptedException{
while(writers & 0 || writeRequests & 0){
readers++;
public synchronized void unlockRead(){
readers--;
notifyAll();
public synchronized void lockWrite() throws InterruptedException{
writeRequests++;
while(readers & 0 || writers & 0){
writeRequests--;
writers++;
public synchronized void unlockWrite() throws InterruptedException{
writers--;
notifyAll();
注意这里要用notifyAll而不是notify，考虑下列一种情况：
Inside the ReadWriteLock there are threads waiting for read access, and threads waiting for write access（读写的waiting都有）. If a thread awakened by notify() was a read access thread, it would be put back to waiting because there are threads waiting for write access（如果用notify，notify读线程，那么它还是会等待，因为有写请求）. However, none of the threads awaiting write access are awakened, so nothing more happens. No threads gain neither read nor write access. By calling noftifyAll() all waiting threads are awakened and check if they can get the desired access. 通过notifyAll来让它们来一起竞争。
notifyAll还有一个好处：
If multiple threads are waiting for read access and none for write access, and unlockWrite() is called, all threads waiting for read access are granted read access at once - not one by one.
仔细看上面的代码，写的非常好！
Read / Write Lock Reentrance
上面的代码是不可重入的，考虑下面情况：
1.Thread 1 gets read access.
2.Thread 2 requests write access but is blocked because there is one reader.
3.Thread 1 re-requests read access (re-enters the lock), but is blocked because there is a write request
如何实现可重入？还是像之前一样吗？看下面：
Read Reentrance
一条规则：A thread is granted read reentrance if it can get read access (no writers or write requests), or if it already has read access (regardless of write requests).
改动如下：
public class ReadWriteLock{
private Map&Thread, Integer& readingThreads =
new HashMap&Thread, Integer&();
private int writers
private int writeRequests
public synchronized void lockRead() throws InterruptedException{
Thread callingThread = Thread.currentThread();
while(! canGrantReadAccess(callingThread)){
readingThreads.put(callingThread,
(getAccessCount(callingThread) + 1));
public synchronized void unlockRead(){
Thread callingThread = Thread.currentThread();
int accessCount = getAccessCount(callingThread);
if(accessCount == 1){ readingThreads.remove(callingThread); }
else { readingThreads.put(callingThread, (accessCount -1)); }
notifyAll();
private boolean canGrantReadAccess(Thread callingThread){
if(writers & 0)
return false;
if(isReader(callingThread) return true;
if(writeRequests & 0)
return false;
return true;
private int getReadAccessCount(Thread callingThread){
Integer accessCount = readingThreads.get(callingThread);
if(accessCount == null) return 0;
return accessCount.intValue();
private boolean isReader(Thread callingThread){
return readingThreads.get(callingThread) != null;
Write Reentrance
规则一：Write reentrance is granted only if the thread has already write access.
public class ReadWriteLock{
private Map&Thread, Integer& readingThreads =
new HashMap&Thread, Integer&();
private int writeAccesses
private int writeRequests
private Thread writingThread = null;
public synchronized void lockWrite() throws InterruptedException{
writeRequests++;
Thread callingThread = Thread.currentThread();
while(! canGrantWriteAccess(callingThread)){
writeRequests--;
writeAccesses++;
writingThread = callingT
public synchronized void unlockWrite() throws InterruptedException{
writeAccesses--;
if(writeAccesses == 0){
writingThread = null;
notifyAll();
private boolean canGrantWriteAccess(Thread callingThread){
if(hasReaders())
return false;
if(writingThread == null)
return true;
if(!isWriter(callingThread)) return false;
return true;
private boolean hasReaders(){
return readingThreads.size() & 0;
private boolean isWriter(Thread callingThread){
return writingThread == callingT
Read to Write Reentrance
Sometimes it is necessary for a thread that have read access to also obtain write access.For this to be allowed the thread must be the only reader.
只改写一下writeLock的重入条件即可：
public class ReadWriteLock{
private Map&Thread, Integer& readingThreads =
new HashMap&Thread, Integer&();
private int writeAccesses
private int writeRequests
private Thread writingThread = null;
public synchronized void lockWrite() throws InterruptedException{
writeRequests++;
Thread callingThread = Thread.currentThread();
while(! canGrantWriteAccess(callingThread)){
writeRequests--;
writeAccesses++;
writingThread = callingT
public synchronized void unlockWrite() throws InterruptedException{
writeAccesses--;
if(writeAccesses == 0){
writingThread = null;
notifyAll();
private boolean canGrantWriteAccess(Thread callingThread){
if(isOnlyReader(callingThread))
return true;
if(hasReaders())
return false;
if(writingThread == null)
return true;
if(!isWriter(callingThread))
return false;
return true;
private boolean hasReaders(){
return readingThreads.size() & 0;
private boolean isWriter(Thread callingThread){
return writingThread == callingT
private boolean isOnlyReader(Thread thread){
return readers == 1 && readingThreads.get(callingThread) != null;
Write to Read Reentrance
Sometimes a thread that has write access needs read access too. A writer should always be granted read access if requested.
只需要改动 canGrantReadAccess方法：
public class ReadWriteLock{
private boolean canGrantReadAccess(Thread callingThread){
if(isWriter(callingThread)) return true;
if(writingThread != null)
return false;
if(isReader(callingThread)
return true;
if(writeRequests & 0)
return false;
return true;
Fully Reentrant ReadWriteLock：
public class ReadWriteLock{
private Map&Thread, Integer& readingThreads =
new HashMap&Thread, Integer&();
private int writeAccesses
private int writeRequests
private Thread writingThread = null;
public synchronized void lockRead() throws InterruptedException{
Thread callingThread = Thread.currentThread();
while(! canGrantReadAccess(callingThread)){
readingThreads.put(callingThread,
(getReadAccessCount(callingThread) + 1));
private boolean canGrantReadAccess(Thread callingThread){
if( isWriter(callingThread) ) return true;
if( hasWriter()
) return false;
if( isReader(callingThread) ) return true;
if( hasWriteRequests()
) return false;
return true;
public synchronized void unlockRead(){
Thread callingThread = Thread.currentThread();
if(!isReader(callingThread)){
throw new IllegalMonitorStateException("Calling Thread does not" +
" hold a read lock on this ReadWriteLock");
int accessCount = getReadAccessCount(callingThread);
if(accessCount == 1){ readingThreads.remove(callingThread); }
else { readingThreads.put(callingThread, (accessCount -1)); }
notifyAll();
public synchronized void lockWrite() throws InterruptedException{
writeRequests++;
Thread callingThread = Thread.currentThread();
while(! canGrantWriteAccess(callingThread)){
writeRequests--;
writeAccesses++;
writingThread = callingT
public synchronized void unlockWrite() throws InterruptedException{
if(!isWriter(Thread.currentThread()){
throw new IllegalMonitorStateException("Calling Thread does not" +
" hold the write lock on this ReadWriteLock");
writeAccesses--;
if(writeAccesses == 0){
writingThread = null;
notifyAll();
private boolean canGrantWriteAccess(Thread callingThread){
if(isOnlyReader(callingThread))
return true;
if(hasReaders())
return false;
if(writingThread == null)
return true;
if(!isWriter(callingThread))
return false;
return true;
private int getReadAccessCount(Thread callingThread){
Integer accessCount = readingThreads.get(callingThread);
if(accessCount == null) return 0;
return accessCount.intValue();
private boolean hasReaders(){
return readingThreads.size() & 0;
private boolean isReader(Thread callingThread){
return readingThreads.get(callingThread) != null;
private boolean isOnlyReader(Thread callingThread){
return readingThreads.size() == 1 &&
readingThreads.get(callingThread) != null;
private boolean hasWriter(){
return writingThread != null;
private boolean isWriter(Thread callingThread){
return writingThread == callingT
private boolean hasWriteRequests(){
return this.writeRequests & 0;
好了，我们回归正题java built-in ReadWriteLock：
这个类实现的原则跟上面是一样的：
1.Read Lock:If no threads have locked the ReadWriteLock for writing,and no thread have requested a write lock (but not yet obtained it).Thus, multiple threads can lock the lock for reading.
2.Write Lock:If no threads are reading or writing.
Thus, only one thread at a time can lock the lock for writing.
该接口的实现：
1.ReentrantReadWriteLock
ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
readWriteLock.readLock().lock();
// multiple readers can enter this section
// if not locked for writing, and not writers waiting
// to lock for writing.
readWriteLock.readLock().unlock();
readWriteLock.writeLock().lock();
// only one writer can enter this section,
// and only if no threads are currently reading.
readWriteLock.writeLock().unlock();
13.一些原子类
1.AtomicBoolean
2.AtomicInteger
3.AtomicLong
4.AtomicReference
5.AtomicStampedReference
6.AtomicIntegerArray
7.AtomicLongArray
8.AtomicReferenceArray
关于这些原子类的具体原理，参考：
Compare and swap is a technique used when designing concurrent algorithms.
What Situations Compare And Swap is Intended to Support
一般我们写程序会用到”check then act” pattern.比如：
class MyLock {
private boolean locked = false;
public boolean lock() {
if(!locked) {
locked = true;
return true;
return false;
但是上述代码在多线程中肯定会出错。为了能在多线程中正确，”check then act” operations必须是原子操作。
Compare And Swap As Atomic Operation
Modern CPUs have built-in support for atomic compare and swap operations.
通过原子类来实现锁：
public static class MyLock {
private AtomicBoolean locked = new AtomicBoolean(false);
public boolean lock() {
return pareAndSet(false, true);
推荐使用这样的原子方法是因为CPU自带这些原子操作，效率高一点。
&&相关文章推荐
参考知识库
* 以上用户言论只代表其个人观点，不代表CSDN网站的观点或立场
访问：27619次
积分：1285
积分：1285
排名：千里之外
原创：101篇
(6)(8)(8)(13)(3)(5)(5)(9)(5)(1)(5)(14)(9)(1)(2)(15)