JAVA并发包备忘录

JAVA并发包的功能主要包含：

1）原子操作类，诸如AtomicInteger、AtomicLong等类；

2）并发集合类，如ArrayBlockingQueue、ConcurrentMap、CopyOnWriteArrayList

3）锁机制，比synchronized性能更优的锁机制，

boolean captured = lock.tryLock();
try {
System.out.println("tryLock:" + captured);
} finally {
if (captured)
lock.unlock();
}

4）线程池，newCachedThreadPool，newFixedThreadPool(5)，newSingleThreadExecutor()等

5）工具类：

       CountDownLatch：在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待

//一个CountDouwnLatch实例是不能重复使用的，也就是说它是一次性的，锁一经被打开就不能再关闭使用了，如果想重复使用，请考虑使用CyclicBarrier。
public class CountDownLatchTest {

// 模拟了100米赛跑，10名选手已经准备就绪，只等裁判一声令下。当所有人都到达终点时，比赛结束。
public static void main(String[] args) throws InterruptedException {

// 开始的倒数锁
final CountDownLatch begin = new CountDownLatch(1);

// 结束的倒数锁
final CountDownLatch end = new CountDownLatch(10);

// 十名选手
final ExecutorService exec = Executors.newFixedThreadPool(10);

for (int index = 0; index < 10; index++) {
final int NO = index + 1;
Runnable run = new Runnable() {
public void run() {
try {
// 如果当前计数为零，则此方法立即返回。
// 等待
begin.await();
Thread.sleep((long) (Math.random() * 10000));
System.out.println("No." + NO + " arrived");
} catch (InterruptedException e) {
} finally {
// 每个选手到达终点时，end就减一
end.countDown();
}
}
};
exec.submit(run);
}
System.out.println("Game Start");
// begin减一，开始游戏
begin.countDown();
// 等待end变为0，即所有选手到达终点
end.await();
System.out.println("Game Over");
exec.shutdown();
}
}

cyclicBarrier：与countDownLoatch相似，只不过countDownLoatch只能执行一次，而cyclicBarrier可以执行多次

/**
* 与countDownLoatch相似，只不过countDownLoatch只能执行一次，而cyclicBarrier可以执行多次
*
* //设置parties、count及barrierCommand属性。 <br/>
* CyclicBarrier(int): <br/>
* <br/>
* //当await的数量到达了设定的数量后，首先执行该Runnable对象。<br/>
* CyclicBarrier(int,Runnable):<br/>
* <br/>
* //通知barrier已完成线程 <br/>
* await(): <br/>
* getNumberWaiting():返回当前正在等待的线程数量。
*
* @author wull
*
*/
public class CyclicBarrierTest {
public static void main(String[] args) {
ExecutorService service = Executors.newCachedThreadPool();
final CyclicBarrier cb = new CyclicBarrier(3); // 三个线程同时到达
for (int i = 0; i < 3; i++) {
Runnable runnable = new Runnable() {
public void run() {
try {
Thread.sleep((long) (Math.random() * 10000));
System.out.println("线程"
+ Thread.currentThread().getName()
+ "即将到达集合地点1，当前已有"
+ (cb.getNumberWaiting() + 1)
+ "个已到达"
+ (cb.getNumberWaiting() == 2 ? "都到齐了，继续走啊"
: "正在等候"));
try {
cb.await();
} catch (BrokenBarrierException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
Thread.sleep((long) (Math.random() * 10000));
System.out.println("线程"
+ Thread.currentThread().getName()
+ "即将到达集合地点2，当前已有"
+ (cb.getNumberWaiting() + 1)
+ "个已到达"
+ (cb.getNumberWaiting() == 2 ? "都到齐了，继续走啊"
: "正在等候"));
try {
cb.await();
} catch (BrokenBarrierException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
Thread.sleep((long) (Math.random() * 10000));
System.out.println("线程"
+ Thread.currentThread().getName()
+ "即将到达集合地点3，当前已有"
+ (cb.getNumberWaiting() + 1)
+ "个已到达"
+ (cb.getNumberWaiting() == 2 ? "都到齐了，继续走啊"
: "正在等候"));
try {
cb.await();
} catch (BrokenBarrierException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
};
service.execute(runnable);
}
service.shutdown();
}
}

Exchanger：提供了一个同步点，在这个同步点，一对线程可以交换数据

/**
* 类java.util.concurrent.Exchanger提供了一个同步点，在这个同步点，一对线程可以交换数据。<br/>
* 每个线程通过exchange()方法的入口提供数据给他的伙伴线程，并接收他的伙伴线程提供的数据，并返回。<br/>
* 当两个线程通过Exchanger交换了对象，这个交换对于两个线程来说都是安全的。 <br/>
*
* @author wull
*
*/
public class ExchangerDemo {
public static void main(String args[]) {
Exchanger<String> exgr = new Exchanger<String>();

new UseString(exgr);
new MakeString(exgr);
}
}

class MakeString implements Runnable {
Exchanger<String> ex;
String str;
MakeString(Exchanger<String> c) {
ex = c;
str = new String();

new Thread(this).start();
}

public void run() {
char ch = 'A';
for (int i = 0; i < 3; i++) {
for (int j = 0; j < 5; j++)
str += (char) ch++;
try {
str = ex.exchange(str);
System.out.println("RECIEVE:"+str);
} catch (InterruptedException exc) {
System.out.println(exc);
}
}
}
}

class UseString implements Runnable {
Exchanger<String> ex;
String str;
UseString(Exchanger<String> c) {
ex = c;
new Thread(this).start();
}

public void run() {
for (int i = 0; i < 3; i++) {
try {
str = ex.exchange(new String());
System.out.println("Got: " + str);
} catch (InterruptedException exc) {
System.out.println(exc);
}
}
}
}

SemaphoreSemaphore：相当于一个锁，这个锁同时可以被N个线程所使用，第N+1个线程处于等待状态

public class SemaphoreDemo {
public static void main(String[] args) {

// 线程池
ExecutorService exec = Executors.newCachedThreadPool();

// 只能5个线程同时访问
final Semaphore semp = new Semaphore(5);

// 模拟20个客户端访问
for (int index = 0; index < 20; index++) {
final int NO = index;
Runnable run = new Runnable() {
public void run() {
try {
// 获取许可
semp.acquire();
System.out.println("Accessing: " + NO);
Thread.sleep((long) (Math.random() * 10000));

// 访问完后，释放
semp.release();

System.out.println("-----------------"
+ semp.availablePermits());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
exec.execute(run);
}

// 退出线程池
exec.shutdown();

}
}

6）Callable和Future

一个有返回值的线程Runnable

 

7）Fork-join框架

这是一个JDK7引入的并行框架，它把流程划分成fork（分解）+join（合并）两个步骤（怎么那么像MapReduce？），传统线程池来实现一个并行任务的时候，经常需要花费大量的时间去等待其他线程执行任务的完成，但是fork-join框架使用work stealing技术缓解了这个问题：

每个工作线程都有一个双端队列，当分给每个任务一个线程去执行的时候，这个任务会放到这个队列的头部；

当这个任务执行完毕，需要和另外一个任务的结果执行合并操作，可是那个任务却没有执行的时候，不会干等，而是把另一个任务放到队列的头部去，让它尽快执行；

当工作线程的队列为空，它会尝试从其他线程的队列尾部偷一个任务过来；

取得的任务可以被进一步分解。

ForkJoinPool.class，ForkJoin框架的任务池，ExecutorService的实现类

ForkJoinTask.class，Future的子类，框架任务的抽象

ForkJoinWorkerThread.class，工作线程

RecursiveTask.class，ForkJoinTask的实现类，compute方法有返回值，下文中有例子

RecursiveAction.class，ForkJoinTask的实现类，compute方法无返回值，只需要覆写compute方法，对于可继续分解的子任务，调用coInvoke方法完成（参数是RecursiveAction子类对象的可变数组）：