1 并发的多面性
1.1 更快的执行
1.2 改进代码设计
2 基本的线程机制
2.1 定义任务
实现Runnable接口，编写run方法
package concurrency;
public class LiftOff implements Runnable {
protected int countDown = 10;
private static int taskCount = 0;
private final int id = taskCount++; // final,初始化后，不被修改
public LiftOff() {
}
public LiftOff(int countDown) {
this.countDown = countDown;
}
public String status() {
return "#" + id + "(" + (countDown > 0 ? countDown : "Liftoff!") + "), ";
}
@Override
public void run() {
while (countDown-- > 0) {
System.out.print(status());
Thread.yield();// 线程调度器
}
}
}
主线程
package concurrency;
public class MainThread {
public static void main(String[] args) {
LiftOff launch = new LiftOff();
launch.run();
}
}
/*
#0(9), #0(8), #0(7), #0(6), #0(5), #0(4), #0(3), #0(2), #0(1), #0(Liftoff!),
*/
2.2 Thread 类
package concurrency;
public class BasicThreads {
public static void main(String[] args) {
Thread t = new Thread(new LiftOff()); // 设置线程 主线程
t.start();// t 线程
System.out.println("waiting for LiftOff"); // 主线程
// main 和 LiftOff.run "同时"执行
}
}
/*
waiting for LiftOff
#0(9), #0(8), #0(7), #0(6), #0(5), #0(4), #0(3), #0(2), #0(1), #0(Liftoff!),
*/
更多线程驱动更多的任务
package concurrency;
public class MoreBasicThreads {
public static void main(String[] args) {
for (int i = 0; i < 5; i++)
new Thread(new LiftOff()).start();
System.out.println("waiting for liftoff");
}
}
/*
#0(9), #1(9), #2(9), #1(8), #0(8), #1(7), #2(8), #3(9), waiting for liftoff
#4(9), #1(6), #0(7), #2(7), #3(8), #4(8), #1(5), #0(6), #4(7), #3(7), #2(6), #3(6), #4(6), #0(5), #1(4), #0(4), #4(5), #3(5), #2(5), #3(4), #4(4), #0(3), #1(3), #0(2), #4(3), #3(3), #2(4), #3(2), #4(2), #0(1), #1(2), #0(Liftoff!), #4(1), #3(1), #3(Liftoff!), #2(3), #4(Liftoff!), #1(1), #2(2), #1(Liftoff!), #2(1), #2(Liftoff!),
*/
2.3 使用Executor
package concurrency;
import chapter9interfaces.filters.Waveform;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class CachedThreadPool {
public static void main(String[] args) {
ExecutorService exec = Executors.newCachedThreadPool();
for (int i = 0; i < 5; i++)
exec.execute(new LiftOff());
exec.shutdown();
}
}
/*
#0(9), #2(9), #1(9), #0(8), #1(8), #0(7), #2(8), #1(7), #2(7), #3(9), #1(6), #2(6), #0(6), #4(9), #0(5), #2(5), #1(5), #3(8), #1(4), #2(4), #0(4), #4(8), #0(3), #2(3), #1(3), #3(7), #1(2), #2(2), #0(2), #4(7), #0(1), #2(1), #1(1), #3(6), #1(Liftoff!), #2(Liftoff!), #0(Liftoff!), #4(6), #3(5), #4(5), #4(4), #3(4), #4(3), #3(3), #4(2), #3(2), #4(1), #3(1), #4(Liftoff!), #3(Liftoff!),
*/
固定线程池，可以一次性预先执行代价高昂的线程分配，限制线程数量。
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class FixedThreadPool {
public static void main(String[] args) {
ExecutorService exec = Executors.newFixedThreadPool(5);
for (int i = 0; i < 5; i++)
exec.execute(new LiftOff());
exec.shutdown();
}
}
/*
#1(9), #4(9), #3(9), #2(9), #0(9), #4(8), #3(8), #1(8), #0(8), #2(8), #3(7), #4(7), #2(7), #3(6), #0(7), #1(7), #0(6), #3(5), #2(6), #4(6), #2(5), #3(4), #0(5), #1(6), #0(4), #3(3), #4(5), #2(4), #4(4), #3(2), #0(3), #1(5), #0(2), #1(4), #3(1), #4(3), #2(3), #4(2), #3(Liftoff!), #1(3), #0(1), #1(2), #4(1), #2(2), #4(Liftoff!), #1(1), #0(Liftoff!), #1(Liftoff!), #2(1), #2(Liftoff!),
*/
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class SingleThreadExecutor {
public static void main(String[] args) {
ExecutorService exec = Executors.newSingleThreadExecutor();
for (int i = 0; i < 5; i++)
exec.execute(new LiftOff());
exec.shutdown();
}
}
/*
#0(9), #0(8), #0(7), #0(6), #0(5), #0(4), #0(3), #0(2), #0(1), #0(Liftoff!), #1(9), #1(8), #1(7), #1(6), #1(5), #1(4), #1(3), #1(2), #1(1), #1(Liftoff!), #2(9), #2(8), #2(7), #2(6), #2(5), #2(4), #2(3), #2(2), #2(1), #2(Liftoff!), #3(9), #3(8), #3(7), #3(6), #3(5), #3(4), #3(3), #3(2), #3(1), #3(Liftoff!), #4(9), #4(8), #4(7), #4(6), #4(5), #4(4), #4(3), #4(2), #4(1), #4(Liftoff!),
*/
2.4 从任务中产生返回值
Runnable 没有返回值
package concurrency;
import java.util.ArrayList;
import java.util.concurrent.*;
class TaskWithResult implements Callable<String>{
private int id;
public TaskWithResult(int id){this.id = id;}
@Override
public String call() {
return "result of TaskWithResult " + id;
}
}
public class CallableDemo {
public static void main(String[] args) {
ExecutorService exec = Executors.newCachedThreadPool();
ArrayList<Future<String>> results = new ArrayList<>(); // 异步执行结果对象
for (int i = 0; i < 10;i++)
results.add(exec.submit(new TaskWithResult(i)));
for (Future<String> fs : results) {
try {
// get() Waits if necessary for the computation to complete,
// and then retrieves its result.
System.out.println(fs.get());
} catch (InterruptedException e) {
System.out.println(e);
} catch (ExecutionException e) {
System.out.println(e);
}finally {
exec.shutdown();
}
}
}
}
/*
result of TaskWithResult 0
result of TaskWithResult 1
result of TaskWithResult 2
result of TaskWithResult 3
result of TaskWithResult 4
result of TaskWithResult 5
result of TaskWithResult 6
result of TaskWithResult 7
result of TaskWithResult 8
result of TaskWithResult 9
*/
2.5 休眠
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class SleepingTask extends LiftOff {
public void run() {
try {
while (countDown-- > 0) {
System.out.print(status());
// Thread.sleep(100);//传统方式 毫秒
TimeUnit.MILLISECONDS.sleep(100);// 时间单位类
}
} catch (InterruptedException e) {
System.out.println("Interrupted");
}
}
public static void main(String[] args) {
ExecutorService exec = Executors.newCachedThreadPool();
for (int i = 0; i < 5; i++)
exec.execute(new SleepingTask());
exec.shutdown();
}
}
/*
#0(9), #2(9), #1(9), #3(9), #4(9), #2(8), #1(8), #4(8), #0(8), #3(8), #2(7), #0(7), #1(7), #4(7), #3(7), #3(6), #4(6), #0(6), #1(6), #2(6), #1(5), #0(5), #3(5), #4(5), #2(5), #0(4), #2(4), #1(4), #3(4), #4(4), #0(3), #4(3), #3(3), #1(3), #2(3), #0(2), #1(2), #2(2), #4(2), #3(2), #0(1), #1(1), #2(1), #3(1), #4(1), #0(Liftoff!), #2(Liftoff!), #1(Liftoff!), #3(Liftoff!), #4(Liftoff!),
*/
2.6 优先级
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class SimplePriorities implements Runnable{
private int countDown = 5;
private volatile double d;
private int priority;
public SimplePriorities(int priority){this.priority = priority;}
public String toString(){
return Thread.currentThread() + ": " + countDown; //"Thread[" + getName() + "," + getPriority() + "," +group.getName() + "]";
}
@Override
public void run() {
Thread.currentThread().setPriority(priority);
while (true){
for (int i = 0;i < 100000; i++){
d += (Math.PI + Math.E)/(double)i;
if (i % 1000 == 0)
Thread.yield();// 线程让步，运算时间足够长，线程调度机制来得及介入
}
System.out.println(this);
if (--countDown == 0) return;;
}
}
public static void main(String[] args) {
ExecutorService exec = Executors.newCachedThreadPool();
for (int i =0; i < 5; i++)
exec.execute(new SimplePriorities(Thread.MIN_PRIORITY));// JDK有10个基本，但与操作系统不能映射但很好。只使用3种
exec.execute(new SimplePriorities(Thread.MAX_PRIORITY));
exec.shutdown();
}
}
/*
Thread[pool-1-thread-3,1,main]: 5
Thread[pool-1-thread-6,10,main]: 5
Thread[pool-1-thread-4,1,main]: 5
Thread[pool-1-thread-5,1,main]: 5
Thread[pool-1-thread-2,1,main]: 5
Thread[pool-1-thread-1,1,main]: 5
Thread[pool-1-thread-4,1,main]: 4
Thread[pool-1-thread-2,1,main]: 4
Thread[pool-1-thread-5,1,main]: 4
Thread[pool-1-thread-6,10,main]: 4
Thread[pool-1-thread-4,1,main]: 3
Thread[pool-1-thread-5,1,main]: 3
Thread[pool-1-thread-3,1,main]: 4
Thread[pool-1-thread-2,1,main]: 3
Thread[pool-1-thread-1,1,main]: 4
Thread[pool-1-thread-6,10,main]: 3
Thread[pool-1-thread-4,1,main]: 2
Thread[pool-1-thread-6,10,main]: 2
Thread[pool-1-thread-5,1,main]: 2
Thread[pool-1-thread-3,1,main]: 3
Thread[pool-1-thread-1,1,main]: 3
Thread[pool-1-thread-4,1,main]: 1
Thread[pool-1-thread-2,1,main]: 2
Thread[pool-1-thread-5,1,main]: 1
Thread[pool-1-thread-6,10,main]: 1
Thread[pool-1-thread-1,1,main]: 2
Thread[pool-1-thread-3,1,main]: 2
Thread[pool-1-thread-2,1,main]: 1
Thread[pool-1-thread-1,1,main]: 1
Thread[pool-1-thread-3,1,main]: 1
*/
2.7 让步
调用yield时，只是建议具有相同优先级但其他线程可以运行。
不能依赖于 yield()，经常被误用。
2.8 后台线程daemon
daemon 不属于程序中不可或缺对部分。
所有的非后台线程结束时，程序就终止了，同时还会杀死进程中的所有后台程序。
也就是说，只要有任何非后台线程还在运行，程序就不会终止。比如，执行main()的就是一个非后台线程。
package concurrency;
import java.util.concurrent.TimeUnit;
public class SimpleDaemons implements Runnable {
@Override
public void run() {
while (true){
try {
TimeUnit.MILLISECONDS.sleep(100);
System.out.println(Thread.currentThread() + " " + this); // 线程名称 + 任务
} catch (InterruptedException e) {
System.out.println("sleep() interrupted");
}
}
}
public static void main(String[] args) throws InterruptedException {
for (int i =0; i< 10; i++){
Thread daemon = new Thread(new SimpleDaemons());
daemon.setDaemon(true); // 启动前，设置守护线程
daemon.start();
}
System.out.println("All daemons started");
TimeUnit.MILLISECONDS.sleep(175);// 设置后，才能看到后台线程启动后的结果。若时间短，则退出主线程，看不到结果
}
}
/*
All daemons started
Thread[Thread-4,5,main] concurrency.SimpleDaemons@251ff751
Thread[Thread-9,5,main] concurrency.SimpleDaemons@24e59eb1
Thread[Thread-1,5,main] concurrency.SimpleDaemons@6c9ffdd2
Thread[Thread-3,5,main] concurrency.SimpleDaemons@449ab273
Thread[Thread-8,5,main] concurrency.SimpleDaemons@52826699
Thread[Thread-5,5,main] concurrency.SimpleDaemons@23981350
Thread[Thread-2,5,main] concurrency.SimpleDaemons@730a3f
Thread[Thread-7,5,main] concurrency.SimpleDaemons@3aa14650
Thread[Thread-0,5,main] concurrency.SimpleDaemons@75892c3d
Thread[Thread-6,5,main] concurrency.SimpleDaemons@34ce7b1e
*/
定制的ThreadFactory,后台状态全部设置为true. 后台线程工厂。
package concurrency;
import java.util.concurrent.ThreadFactory;
public class DaemonThreadFactory implements ThreadFactory {
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r);
t.setDaemon(true);
return t;
}
}
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class DaemonFromFactory implements Runnable {
@Override
public void run() {
try {
while (true) {
TimeUnit.MILLISECONDS.sleep(100);
System.out.println(Thread.currentThread() + " " + this);
}
} catch (InterruptedException e) {
System.out.println("Interrupted");
}
}
public static void main(String[] args) throws InterruptedException {
ExecutorService exec = Executors.newCachedThreadPool(new DaemonThreadFactory());//执行器
for (int i = 0; i< 10; i++)
exec.execute(new DaemonFromFactory());
System.out.println("All daemons started");
TimeUnit.MILLISECONDS.sleep(500);
}
}
/*
All daemons started
Thread[Thread-8,5,main] concurrency.DaemonFromFactory@cdba8b5
Thread[Thread-9,5,main] concurrency.DaemonFromFactory@103290e
Thread[Thread-2,5,main] concurrency.DaemonFromFactory@3640cadc
Thread[Thread-3,5,main] concurrency.DaemonFromFactory@1f99e22a
Thread[Thread-4,5,main] concurrency.DaemonFromFactory@17233e2b
Thread[Thread-0,5,main] concurrency.DaemonFromFactory@8c73ab6
Thread[Thread-7,5,main] concurrency.DaemonFromFactory@2e522ee1
Thread[Thread-6,5,main] concurrency.DaemonFromFactory@148e99e
Thread[Thread-5,5,main] concurrency.DaemonFromFactory@9a3c87d
Thread[Thread-1,5,main] concurrency.DaemonFromFactory@20cc06fb
Thread[Thread-8,5,main] concurrency.DaemonFromFactory@cdba8b5
Thread[Thread-3,5,main] concurrency.DaemonFromFactory@1f99e22a
Thread[Thread-7,5,main] concurrency.DaemonFromFactory@2e522ee1
Thread[Thread-4,5,main] concurrency.DaemonFromFactory@17233e2b
Thread[Thread-9,5,main] concurrency.DaemonFromFactory@103290e
Thread[Thread-5,5,main] concurrency.DaemonFromFactory@9a3c87d
Thread[Thread-1,5,main] concurrency.DaemonFromFactory@20cc06fb
Thread[Thread-6,5,main] concurrency.DaemonFromFactory@148e99e
Thread[Thread-0,5,main] concurrency.DaemonFromFactory@8c73ab6
Thread[Thread-2,5,main] concurrency.DaemonFromFactory@3640cadc
Thread[Thread-3,5,main] concurrency.DaemonFromFactory@1f99e22a
Thread[Thread-6,5,main] concurrency.DaemonFromFactory@148e99e
Thread[Thread-5,5,main] concurrency.DaemonFromFactory@9a3c87d
Thread[Thread-8,5,main] concurrency.DaemonFromFactory@cdba8b5
Thread[Thread-4,5,main] concurrency.DaemonFromFactory@17233e2b
Thread[Thread-7,5,main] concurrency.DaemonFromFactory@2e522ee1
Thread[Thread-9,5,main] concurrency.DaemonFromFactory@103290e
Thread[Thread-0,5,main] concurrency.DaemonFromFactory@8c73ab6
Thread[Thread-2,5,main] concurrency.DaemonFromFactory@3640cadc
Thread[Thread-1,5,main] concurrency.DaemonFromFactory@20cc06fb
Thread[Thread-5,5,main] concurrency.DaemonFromFactory@9a3c87d
Thread[Thread-0,5,main] concurrency.DaemonFromFactory@8c73ab6
Thread[Thread-2,5,main] concurrency.DaemonFromFactory@3640cadc
Thread[Thread-9,5,main] concurrency.DaemonFromFactory@103290e
Thread[Thread-4,5,main] concurrency.DaemonFromFactory@17233e2b
Thread[Thread-3,5,main] concurrency.DaemonFromFactory@1f99e22a
Thread[Thread-8,5,main] concurrency.DaemonFromFactory@cdba8b5
Thread[Thread-7,5,main] concurrency.DaemonFromFactory@2e522ee1
Thread[Thread-6,5,main] concurrency.DaemonFromFactory@148e99e
Thread[Thread-1,5,main] concurrency.DaemonFromFactory@20cc06fb
*/
每个静态对ExectutorService创建方法都被重载为接受一个ThreadFactory对象，而这个对象被用来创建新的线程
package concurrency;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class DaemonThreadPoolExecutor extends ThreadPoolExecutor {
//(int corePoolSize,
// int maximumPoolSize,
// long keepAliveTime,
// TimeUnit unit,
// BlockingQueue<Runnable> workQueue)
public DaemonThreadPoolExecutor() {
super(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(),
new DaemonThreadFactory());
}
}
可以调用isDaemon 确定线程是否是一个后台线程
Daemon线程被设置成后台模式，然后派生出许多子线程，这些线程没有被显示地设置为后台模式，不过它们的确是后台线程。
package concurrency;
import java.util.concurrent.TimeUnit;
class Daemon implements Runnable {
private Thread[] t = new Thread[10];
@Override
public void run() {
for (int i = 0; i < t.length; i++) {
t[i] = new Thread(new DaemonSpawn());
t[i].start();
System.out.println("DaemonSpawn " + i + " Started, ");
}
for (int i = 0; i < t.length; i++) {
System.out.println("t[" + i + "].isDaemon() = " + t[i].isDaemon() + ", ");
}
while (true)
Thread.yield();
}
}
class DaemonSpawn implements Runnable {
@Override
public void run() {
while (true)
Thread.yield();
}
}
public class Daemons {
public static void main(String[] args) throws InterruptedException {
Thread d = new Thread(new Daemon());
d.setDaemon(true);
d.start();
System.out.println("d.isDaemon() = " + d.isDaemon() + ",");
//允许Daemon线程完成
TimeUnit.SECONDS.sleep(1);
}
}
/*
d.isDaemon() = true,
DaemonSpawn 0 Started,
DaemonSpawn 1 Started,
DaemonSpawn 2 Started,
DaemonSpawn 3 Started,
DaemonSpawn 4 Started,
DaemonSpawn 5 Started,
DaemonSpawn 6 Started,
DaemonSpawn 7 Started,
DaemonSpawn 8 Started,
DaemonSpawn 9 Started,
t[0].isDaemon() = true,
t[1].isDaemon() = true,
t[2].isDaemon() = true,
t[3].isDaemon() = true,
t[4].isDaemon() = true,
t[5].isDaemon() = true,
t[6].isDaemon() = true,
t[7].isDaemon() = true,
t[8].isDaemon() = true,
t[9].isDaemon() = true,
*/
后台进程在不执行finally子句的情况下，会终止run
package concurrency;
import java.util.concurrent.TimeUnit;
class ADaemon implements Runnable{
@Override
public void run() {
System.out.println("Starting ADaemon");
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
System.out.println("Exiting via InterruptedExcption");
}finally {
System.out.println("This should always run?"); // 如果
}
}
}
public class DaemonsDontRunFinally {
public static void main(String[] args) {
Thread t = new Thread(new ADaemon());
t.setDaemon(true); // 如果注释掉，则 执行finally子句
t.start();
}
}
/*
Starting ADaemon
//This should always run?
*/
2.9 编码的变体
继承线程，重写run
package concurrency;
public class SimpleThread extends Thread {
private int countDown = 5;
private static int threadCount = 0;
public SimpleThread(){
// Store the thread name;
super(Integer.toString(++threadCount));
start();
}
public String toString(){
return "#" + getName() + "(" + countDown + "),";
}
public void run(){
while(true){
System.out.println(this);
if (--countDown == 0)
return;
}
}
public static void main(String[] args) {
for (int i = 0; i < 5; i++)
new SimpleThread();
}
}
/*
#2(5),
#5(5),
#2(4),
#3(5),
#4(5),
#1(5),
#4(4),
#3(4),
#2(3),
#5(4),
#2(2),
#3(3),
#4(3),
#1(4),
#4(2),
#3(2),
#2(1),
#5(3),
#3(1),
#4(1),
#1(3),
#5(2),
#1(2),
#5(1),
#1(1),
*/
自管理的Runnable
package concurrency;
public class SelfManaged implements Runnable {
private int countDown = 5;
private Thread t = new Thread(this);
public SelfManaged() {
t.start();
}
public String toString() {
return Thread.currentThread().getName() + "(" + countDown + "),";
}
@Override
public void run() {
while (true) {
System.out.println(this);
if (--countDown == 0)
return;
}
}
public static void main(String[] args) {
for (int i = 0; i < 5; i++)
new SelfManaged();
}
}
/*
Thread-1(5),
Thread-2(5),
Thread-3(5),
Thread-0(5),
Thread-3(4),
Thread-4(5),
Thread-4(4),
Thread-4(3),
Thread-2(4),
Thread-1(4),
Thread-2(3),
Thread-4(2),
Thread-3(3),
Thread-4(1),
Thread-0(4),
Thread-3(2),
Thread-2(2),
Thread-3(1),
Thread-1(3),
Thread-2(1),
Thread-0(3),
Thread-1(2),
Thread-0(2),
Thread-0(1),
Thread-1(1),
*/
在构造器中启动线程可能会有问题，因为另一个任务可能会在构造器结束之前开始执行，以为着该任务能够访问处于不稳定状态的对象。
这是优选Executor而不是显示的创建Thread对象的另一个原因。
通过内部类来将线程代码隐藏在类中。
package concurrency;
import java.util.concurrent.TimeUnit;
class InnerThread1 {
//扩展自Thread的named内部类，并且在构造器中创建了这个内部类的一个实例
//如果内部类具有你在其他方法中需要访问的特殊能力（新方法），那么这么做将会很有意义。
//但在大多数时候，创建线程的原因只是为了使用Thread的能力，因此不必创建named内部类
private int countDown = 5;
private Inner inner;
private class Inner extends Thread {
Inner(String name) {
super(name);
start();
}
public void run() {
try {
while (true) {
System.out.println(this);
if (--countDown == 0)
return;
sleep(10);
}
} catch (InterruptedException e) {
System.out.println("Interrupted");
}
}
public String toString() {
return getName() + " : " + countDown;
}
}
public InnerThread1(String name){
inner = new Inner(name);
}
}
class InnerThread2{
// 在构造器中创建一个匿名的Thread子类，并且将其向上转型为Thread 引用t
private int countDown = 5;
private Thread t;
public InnerThread2(String name){
t = new Thread(name){
@Override
public void run(){
try {
while (true) {
System.out.println(this);
if (--countDown == 0) return;
sleep(10);
}
}catch (InterruptedException e){
System.out.println("Sleep() interrupted.");
}
}
@Override
public String toString(){
return getName() + " : " + countDown;
}
};
t.start();
}
}
class InnerRunnable1{
private int countDown = 5;
private Inner inner;
private class Inner implements Runnable{
Thread t;
Inner(String name){
t = new Thread(this,name);
t.start();
}
@Override
public void run() {
try{
while (true){
System.out.println(this);
if (--countDown == 0) return;
TimeUnit.MILLISECONDS.sleep(10);
}
}catch (InterruptedException e){
System.out.println("sleep() interrupted.");
}
}
public String toString(){
return t.getName() + " : " + countDown;
}
}
public InnerRunnable1(String name){
inner = new Inner(name);
}
}
class InnerRunnable2 {
private int countDown = 5;
private Thread t;
public InnerRunnable2(String name){
t = new Thread(new Runnable() {
@Override
public void run() {
try{
while (true){
System.out.println(this);
if (--countDown == 0) return;
TimeUnit.MILLISECONDS.sleep(10);
}
}catch (InterruptedException e){
System.out.println("sleep() interrupted.");
}
}
public String toString(){
return Thread.currentThread().getName() + " : " + countDown;
}
},name);
t.start();
}
}
class ThreadMethod{
//方法内部创建线程。准备好线程，调用方法，与该类的构造器内部启动线程相比，是更加有用而适合的方式。
private int countDown = 5;
private Thread t;
private String name;
public ThreadMethod(String name){this.name = name;}
public void runTask(){
if (t == null){
t = new Thread(name){
public void run(){
try{
while (true){
System.out.println(this);
if (--countDown ==0)return;
sleep(10);
}
}catch (InterruptedException e){
System.out.println("sleep() interrupted.");
}
}
};
t.start();
}
}
}
public class ThreadVariations {
public static void main(String[] args) {
new InnerThread1("InnerThread1");
new InnerThread2("InnerThread2");
new InnerRunnable1("InnerRunnable1");
new InnerRunnable2("InnerRunnable2");
new ThreadMethod("ThreadMethod").runTask();
}
}
/*
InnerThread1 : 5
InnerThread2 : 5
InnerRunnable1 : 5
InnerRunnable2 : 5
Thread[ThreadMethod,5,main]
InnerThread1 : 4
InnerThread2 : 4
InnerRunnable2 : 4
Thread[ThreadMethod,5,main]
InnerRunnable1 : 4
InnerThread1 : 3
InnerThread2 : 3
InnerRunnable2 : 3
InnerRunnable1 : 3
Thread[ThreadMethod,5,main]
InnerThread1 : 2
InnerThread2 : 2
InnerRunnable2 : 2
InnerRunnable1 : 2
Thread[ThreadMethod,5,main]
InnerThread1 : 1
InnerThread2 : 1
InnerRunnable2 : 1
Thread[ThreadMethod,5,main]
InnerRunnable1 : 1
*/
2.10 术语
创建任务，并通过某种方式将一个线程附着到任务上，以使得这个线程可以驱动任务。
线程类自身不执行任何操作，它只是驱动赋予它对任务。
Runnable接口对名字选择很糟糕，TASK是个好名字。
Java的线程机制来自于C的低级的p线程方式。
2.11 加入一个线程
t.join(),则当前线程被挂起，直到目标线程t结束后才恢复（t.isAlive（）返回为假）
package concurrency;
class Sleeper extends Thread {
private int duration;
public Sleeper(String name, int sleepTime) {
super(name);
duration = sleepTime;
start();
}
@Override
public void run() {
try {
sleep(duration);
} catch (InterruptedException e) {
System.out.println(getName() + " was interrupted. " + " isInterrupted():" + isInterrupted());//异常被捕获时候将清理这个标识，恒为false
return;
}
System.out.println(getName() + " has awakened.");
}
}
class Joiner extends Thread {
private Sleeper sleeper;
public Joiner(String name,Sleeper sleeper){
super(name);
this.sleeper = sleeper;
start();
}
@Override
public void run(){
try {
sleeper.join();
} catch (InterruptedException e) {
System.out.println("Interrupted.");
}
System.out.println(getName() + " join completed.");
}
}
public class Joining {
public static void main(String[] args) {
Sleeper sleeper = new Sleeper("Sleepy",1500),
grumpy = new Sleeper("Grumpy",1500);
Joiner dopey = new Joiner("Dopey",sleeper),
doc = new Joiner("Doc",grumpy);
grumpy.interrupt();
}
}
/*
Grumpy was interrupted. isInterrupted():false
Doc join completed.
Sleepy has awakened.
Dopey join completed.
如果 grumpy.interrupt(); 注释掉
Sleepy has awakened.
Grumpy has awakened.
Dopey join completed.
Doc join completed.
*/
2.12 创建一个有响应的用户界面
package concurrency;
class UnresponsiveUI {//关注于运算，不能读取控制台输入
private volatile double d = 1;
public UnresponsiveUI() throws Exception {
while (d > 0)
d = d + (Math.PI + Math.E) / d;
System.in.read();//永远都不会执行
}
}
public class ResponsiveUi extends Thread {// 把运算放在任务里单独允许，可以进行运算都同时监听控制台输入
private static volatile double d = 1;
public ResponsiveUi() {
setDaemon(true);
start();
}
public void run() {
while (true) {
d = d + (Math.PI + Math.E) / d;
}
}
public static void main(String[] args) throws Exception {
// new UnresponsiveUI();
new ResponsiveUi();
System.in.read();
System.out.println(d);
}
}
2.13 线程组
线程组是不成功的尝试，忽略，没有官方承认。
“继续错误的代价由别人承担，而承认错误的代价由自己承担”
2.14 捕获异常
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ExceptionThread implements Runnable {
@Override
public void run() {
throw new RuntimeException();
}
public static void main(String[] args) {
ExecutorService exec = Executors.newCachedThreadPool();
exec.execute(new ExceptionThread());
}
}
/*
Exception in thread "pool-1-thread-1" java.lang.RuntimeException
at concurrency.ExceptionThread.run(ExceptionThread.java:9)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
*/
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class NaiveExceptionHandling {
public static void main(String[] args) {
try {
ExecutorService exec = Executors.newCachedThreadPool();
exec.execute(new ExceptionThread());
}catch (RuntimeException e){ // 不能捕获逃逸的异常
System.out.println("异常已处理");
}
}
}
解决该问题，需要修改Executor产生线程的方式。
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
class ExceptionThread2 implements Runnable{
@Override
public void run() {
Thread t = Thread.currentThread();
System.out.println("Run() by " + t);//Run() by Thread[Thread-0,5,main]
System.out.println("eh = " + t.getUncaughtExceptionHandler());//eh = concurrency.MyUncaughtExceptionHandler@198e2867
throw new RuntimeException();
}
}
class MyUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler{
@Override
public void uncaughtException(Thread t, Throwable e) {
System.out.println(t + "捕获" + e);//捕获java.lang.RuntimeException
}
}
class HandlerThreadFactory implements ThreadFactory{
@Override
public Thread newThread(Runnable r) {
System.out.println(this + " Creating new Thread"); //concurrency.HandlerThreadFactory@8807e25 Creating new Thread
Thread t = new Thread(r);
System.out.println("created " + t);//created Thread[Thread-0,5,main]
t.setUncaughtExceptionHandler(new MyUncaughtExceptionHandler());
System.out.println("eh = " + t.getUncaughtExceptionHandler());//eh = concurrency.MyUncaughtExceptionHandler@198e2867
return t;
}
}
public class CaptureUncaughtException {
public static void main(String[] args) {
ExecutorService exec = Executors.newCachedThreadPool(
new HandlerThreadFactory());
exec.execute(new ExceptionThread2());
}
}
/*
concurrency.HandlerThreadFactory@7440e464 Creating new Thread
created Thread[Thread-0,5,main]
eh = concurrency.MyUncaughtExceptionHandler@49476842
Run() by Thread[Thread-0,5,main]
eh = concurrency.MyUncaughtExceptionHandler@49476842
concurrency.HandlerThreadFactory@7440e464 Creating new Thread
created Thread[Thread-1,5,main]
eh = concurrency.MyUncaughtExceptionHandler@58b6b39
Thread[Thread-0,5,main]捕获java.lang.RuntimeException
*/
上面示例，是逐个设置处理器。
在代码中处处使用相同的异常处理器，简单的方式如下：
Thread类中设置一个静态域，并将这个处理器设置为默认的未捕获异常处理器。
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class SettingDefaultHandler {
public static void main(String[] args) {
Thread.setDefaultUncaughtExceptionHandler(new MyUncaughtExceptionHandler());
ExecutorService exec = Executors.newCachedThreadPool();
exec.execute(new ExceptionThread());
}
}
/*
Thread[pool-1-thread-1,5,main]捕获java.lang.RuntimeException
*/
系统检查 线程专有版本，若没有，检查线程组 是否有专有的uncaughtException方法，再没有，调用DefaultUncaughtExceptionHandler。
3 共享受限资源
两个实体试图同时使用同一个资源
3.1 不正确地访问资源
一个任务产生偶数
其他任务检查偶数的有效性
IntGenerator抽象类，没有实现Generator接口，因为它必须产生一个int,而泛型不支持基本类型参数
package concurrency;
public abstract class IntGenerator {
private volatile boolean canceled = false; // boolean atomic 原子性，可视性
public abstract int next();
public void cancel(){canceled = true;}
public boolean isCanceled(){return canceled;}
}
测试类
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class EvenChecker implements Runnable{
private IntGenerator generator;
private final int id;
public EvenChecker(IntGenerator g, int ident){
generator = g;
id = ident;
}
@Override
public void run() {
while (!generator.isCanceled()){
int val = generator.next();
if (val % 2 !=0 ){
System.out.println(val + " not even!");
generator.cancel();
}
}
}
public static void test(IntGenerator gp,int count){ // 静态测试方法
System.out.println("Press Control-C to exit");
ExecutorService exec = Executors.newCachedThreadPool();
for (int i = 0; i< count; i++)
exec.execute(new EvenChecker(gp,i));
exec.shutdown();
}
public static void test(IntGenerator gp){// 静态测试方法，线程数量默认为10，只传生成器
test(gp,10);
}
}
package concurrency;
public class EvenGenerator extends IntGenerator {
private int currentEvenValue = 0;
@Override
public int next() {
++currentEvenValue; // Danger!
// Thread.yield(); // 线程让步，更快看到效果
++currentEvenValue;
return currentEvenValue;
}
public static void main(String[] args) {
EvenChecker.test(new EvenGenerator());
}
}
/*
Press Control-C to exit
91 not even!
93 not even!
97 not even!
*/
在java中，递增也不是原子性的操作，如果不保护操作，即使单一的递增也不是安全的。
3.2 解决共享资源竞争
同步控制EvenGenerator
currentEvenValue 域设置为private
package concurrency;
public class SynchronizedEvenGenerator extends IntGenerator {
private int currentEvenValue = 0;
@Override
public synchronized int next() {
++currentEvenValue; // Danger!
Thread.yield(); // 线程让步，更快看到效果
++currentEvenValue;
return currentEvenValue;
}
public static void main(String[] args) {
EvenChecker.test(new EvenGenerator());
}
}
使用显式的Lock对象
package concurrency;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class MutexEvenGenerator extends IntGenerator {
private int currentEvenValue = 0;
private Lock lock = new ReentrantLock(); //
@Override
public int next() {
lock.lock(); //锁
try{
++currentEvenValue;
Thread.yield();
++currentEvenValue;
return currentEvenValue;//返回数据
}finally {
lock.unlock();//解锁
}
}
public static void main(String[] args) {
EvenChecker.test(new MutexEvenGenerator());
}
}
Synchronized 代码量，用户出错可能性低 没有机会去做异常的清理工作，
显式Lock可以try finally维护正确状态。拥有锁获取和释放的可操作性，可中断的获取锁以及超时获取锁等Synchronized关键字所不具备的同步特性。
package concurrency;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
public class AttemptLocking {
private ReentrantLock lock = new ReentrantLock();// 重入锁：支持重进入的锁，表示该锁能够支持一个线程对资源的重复加锁。该锁还支持获取锁时的公平和非公平性的选择。
public void untimed(){
boolean captured = lock.tryLock();
try {
System.out.println("tryLock(): " + captured);
}finally {
if (captured)
lock.unlock();
}
}
public void timed(){
boolean captured = false;
try{
captured = lock.tryLock(2, TimeUnit.SECONDS);
}catch (InterruptedException e){
throw new RuntimeException(e);
}
try{
System.out.println("tryLock(2, TimeUnit.SECONDS): " + captured);
}finally {
if (captured)
lock.unlock();
}
}
public static void main(String[] args) throws InterruptedException {
final AttemptLocking al = new AttemptLocking();
al.untimed();
al.timed();
new Thread(){
{setDaemon(true);}
public void run(){
al.lock.lock();
System.out.println("acquired");
}
}.start();
Thread.sleep(1000); //后台线程启动的比较晚，会看不到效果
Thread.yield();
al.untimed();
al.timed();
}
}
/*
tryLock(): true
tryLock(2, TimeUnit.SECONDS): true
acquired
tryLock(): false
tryLock(2, TimeUnit.SECONDS): false
*/
3.3 原子性与易变性
原子性可以应用于除long和double之外的所有基本类型之上的“简单操作”（JVM将64位 读取和写入当作 两个分离的32位操作来操作执行。字撕裂）
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class AtomicityTest implements Runnable{
private int i = 0;
public int getValue(){return i;} // return i 是原子性操作，但缺少同步使得其数值可以处于不稳定但中间状态被读取。
private synchronized void evenIncrement(){
i++;
i++;
}
@Override
public void run() {
while (true)
evenIncrement();
}
public static void main(String[] args) throws InterruptedException {
ExecutorService exec = Executors.newCachedThreadPool();
AtomicityTest at = new AtomicityTest();
exec.execute(at);
// Thread.sleep(1000);
while (true){
int val = at.getValue();
if (val % 2 != 0) {
System.out.println(val);
System.exit(0);
}
}
}
}
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
class CircularSet{
private int[] array;
private int len;
private int index = 0;
public CircularSet(int size){
array = new int[size];
len = size;
for (int i = 0; i < size; i++)
array[i] = -1;
}
public synchronized void add(int i){
array[index] = i;
index = ++index % len;
}
public synchronized boolean contains(int val){
for (int i = 0; i < len; i++)
if (array[i] == val) return true;
return false;
}
}
public class SerialNumberChecker {
private static final int SIZE = 10;
private static CircularSet serials = new CircularSet(1000);
private static ExecutorService exec = Executors.newCachedThreadPool();
static class SerialChecker implements Runnable{
@Override
public void run() {
while (true){
int serial = SerialNumberGenerator.nextSerialNumber();
if (serials.contains(serial)){
System.out.println("Duplicate: " + serial);
System.exit(0);
}
serials.add(serial);
}
}
}
public static void main(String[] args) throws InterruptedException {
for (int i = 0; i < SIZE; i++)
exec.execute(new SerialChecker());
TimeUnit.SECONDS.sleep(4);
System.out.println("No duplicates detected.");
System.exit(0);
}
}
/*
Duplicate: 2937
修改 synchronized int nextSerialNumber()
No duplicates detected.
3.4 原子类
package concurrency;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
public class AtomicIntegerTest implements Runnable{
private AtomicInteger i = new AtomicInteger(0);
public int getValue(){return i.get();
}
private void evenIncrement(){i.addAndGet(2);}//Atomically adds the given value to the current value.
public void run(){
while (true)
evenIncrement();
}
public static void main(String[] args) {
new Timer().schedule(new TimerTask() {
@Override
public void run() {
System.err.println("Aborting");
System.exit(0);
}
},5000);// 定时器 5秒后停止
ExecutorService exec = Executors.newCachedThreadPool();
AtomicIntegerTest ait = new AtomicIntegerTest();
exec.execute(ait);
while (true){
int val = ait.getValue();
if (val % 2 != 0){
System.out.println(val);
System.exit(0);
}
}
}
}
/*
Aborting
*/
package concurrency;
import java.util.concurrent.atomic.AtomicInteger;
public class AtomicEvenGenerator extends IntGenerator {
private AtomicInteger currentEvenValue = new AtomicInteger(0);
@Override
public int next() {
return currentEvenValue.addAndGet(2);
}
public static void main(String[] args) {
EvenChecker.test(new AtomicEvenGenerator());
}
}
3.5 临界区
critical section 又叫 同步控制块。进入此段代码前，必须得到syncObject对象的锁
不对整个方法进行同步控制，可以使多个任务访问对象但时间性能得到显著提高。
package concurrency;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
class Pair {//非线程安全
private int x, y;
public Pair(int x, int y) {
this.x = x;
this.y = y;
}
public Pair() {
this(0, 0);
}
public int getX() {
return x;
}
public int getY() {
return y;
}
public void incrementX() {
x++;
}
public void incrementY() {
y++;
}
public String toString() {
return "x " + x + ",y: " + y;
}
public class PairValuesNotEqualException extends RuntimeException {
public PairValuesNotEqualException() {
super("Pair value not equal: " + Pair.this);
}
}
public void checkState() {
if (x != y)
throw new PairValuesNotEqualException();
}
}
abstract class PairManager { //放入线程安全的类中
AtomicInteger checkCounter = new AtomicInteger(0); // 检查的计数器
protected Pair p = new Pair();
private List<Pair> storage = Collections.synchronizedList(new ArrayList<Pair>());// 同步list
public synchronized Pair getPair() {
//拷贝一份，保证原数据 safe
return new Pair(p.getX(), p.getY());
}
//假设它是一个需要时间消耗的操作
protected void store(Pair p) {
storage.add(p);
try {
TimeUnit.MILLISECONDS.sleep(50);
} catch (InterruptedException ingore) {
}
}
public abstract void increment();
}
class PairManager1 extends PairManager {
@Override
public synchronized void increment() {// 方法同步
p.incrementX();
p.incrementY();
store(getPair());
}
}
class PairManager2 extends PairManager {
@Override
public void increment() {
Pair temp;
synchronized (this) {// 临界区
p.incrementX();
p.incrementY();
temp = getPair();
}
store(temp);
}
}
class PairManipulator implements Runnable{
private PairManager pm;
public PairManipulator(PairManager pm){this.pm = pm;}
@Override
public void run() {
while (true)
pm.increment();
}
public String toString(){
return "Pair: " + pm.getPair() + " checkCounter = " + pm.checkCounter.get();// 计数数值 取出
}
}
class PairChecker implements Runnable{
private PairManager pm;
public PairChecker(PairManager pm){this.pm = pm;}
@Override
public void run() {
while (true){
pm.checkCounter.incrementAndGet();
pm.getPair().checkState();
}
}
}
public class CriticalSection {
static void testSpproaches(PairManager pman1,PairManager pman2){
ExecutorService exec = Executors.newCachedThreadPool();
PairManipulator pm1 = new PairManipulator(pman1),
pm2 = new PairManipulator(pman2);
PairChecker pcheck1 = new PairChecker(pman1);
PairChecker pcheck2 = new PairChecker(pman2);
exec.execute(pm1);
exec.execute(pm2);
exec.execute(pcheck1);
exec.execute(pcheck2);
try{
TimeUnit.MILLISECONDS.sleep(500);
}catch (InterruptedException e){
System.out.println("Sleep interrupted");
}
System.out.println("pm1: " + pm1 + "
pm2: " + pm2);
System.exit(0);
}
public static void main(String[] args) {
PairManager pman1 = new PairManager1(),
pman2=new PairManager2();
testSpproaches(pman1,pman2);
}
}
/*
pm1: Pair: x 39,y: 39 checkCounter = 3
pm2: Pair: x 39,y: 39 checkCounter = 116136397
*/
使用锁，这个例子会有异常：
package concurrency;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
class ExplictPairManager1 extends PairManager{
private Lock lock = new ReentrantLock();
@Override
public synchronized void increment() {
lock.lock();
try{
p.incrementX();
p.incrementX();
store(p);
}finally {
lock.unlock();
}
}
}
class ExplicttPairManager2 extends PairManager{
private Lock lock = new ReentrantLock();
@Override
public void increment() {
Pair temp;
lock.lock();
try{
p.incrementX();
p.incrementY();
temp = getPair();
}finally {
lock.unlock();
}
store(temp);
}
}
public class ExplicitCriticalSection {
public static void main(String[] args) {
PairManager pman1 = new ExplictPairManager1(),
pman2 = new ExplicttPairManager2();
CriticalSection.testSpproaches(pman1,pman2);
}
}
/*
Exception in thread "pool-1-thread-4" concurrency.Pair$PairValuesNotEqualException: Pair value not equal: x 2,y: 1
at concurrency.Pair.checkState(CriticalSection.java:51)
at concurrency.PairChecker.run(CriticalSection.java:123)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Exception in thread "pool-1-thread-3" concurrency.Pair$PairValuesNotEqualException: Pair value not equal: x 44,y: 0
at concurrency.Pair.checkState(CriticalSection.java:51)
at concurrency.PairChecker.run(CriticalSection.java:123)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
pm1: Pair: x 254,y: 0 checkCounter = 1
pm2: Pair: x 128,y: 128 checkCounter = 2171782
*/
3.6 在其他对象上同步
两个任务可以同时进入同一个对象，这个对象上的方法，是在不同的锁上同步
互相让步运行的效果
package concurrency;
class DualSynch{
private Object syncObject = new Object();
public synchronized void f(){
for (int i = 0;i < 5; i++){
System.out.println("f()");
Thread.yield();
}
}
public void g(){
synchronized (syncObject){
for (int i = 0; i < 5; i++){
System.out.println("g()");
Thread.yield();
}
}
}
}
public class SyncObject {
public static void main(String[] args) {
final DualSynch ds = new DualSynch();
new Thread(){
public void run(){
ds.f();
}
}.start();
ds.g();
}
}
/*
g()
f()
f()
g()
g()
f()
g()
f()
g()
f()
*/
3.7 线程本地存储
单独的线程都被分配了自己的存储，每个都跟踪自己的计数器，即便只有一个ThreadLocalVariableHolder
package concurrency;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
class Accessor implements Runnable{
private final int id;
public Accessor(int idn){this.id = idn;}
@Override
public void run() {
while (!Thread.currentThread().isInterrupted()){
ThreadLocalVariableHolder.increment();
System.out.println(this);
Thread.yield();
}
}
public String toString(){
return "#" + id + ": " + ThreadLocalVariableHolder.get();
}
}
public class ThreadLocalVariableHolder {
private static ThreadLocal<Integer> value = new ThreadLocal<Integer>(){
private Random rand = new Random(47);
protected synchronized Integer initialValue(){return rand.nextInt(10000);}
};
public static void increment(){
value.set(value.get() + 1) ;
}
public static int get(){return value.get();}
public static void main(String[] args) throws InterruptedException {
ExecutorService exec = Executors.newCachedThreadPool();
for (int i = 0; i < 5; i ++)
exec.execute(new Accessor(i));
TimeUnit.MILLISECONDS.sleep(20);
exec.shutdownNow();
}
}
/*
#0: 9259
#3: 1862
#0: 9260
#2: 6694
#1: 556
#2: 6695
#0: 9261
#3: 1863
#0: 9262
#4: 962
#0: 9263
*/
4 终结任务</h2
4 装饰性花园</h2
演示终止问题，而且还是一个资源共享的示例。
通过cancel
package concurrency;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
class Count{// 花园参观者的主计数值
private int count = 0;
private Random rand = new Random(47);
public synchronized int increment(){
int temp = count;
if (rand.nextBoolean())// 一半的时间让步
Thread.yield();
return (count = ++temp);
}
public synchronized int value(){return count;}
}
class Entrance implements Runnable{
private static Count count = new Count();
private static List<Entrance> entrances = new ArrayList<>();
private int number = 0;// 通过本入口进入的参观者的数量
private final int id;
private static volatile boolean canceled = false;
public static void cancel(){canceled = true;}
public Entrance(int id){
this.id = id;
entrances.add(this);
}
@Override
public void run() {
while (!canceled){
synchronized (this){
++ number;
}
System.out.println(this + " Total: " + count.increment());
try{
TimeUnit.MILLISECONDS.sleep(100);
}catch (InterruptedException e){
System.out.println("Sleep interrupted");
}
}
System.out.println("Stopping " + this);
}
public synchronized int getValue(){
return number;
}
public String toString(){
return "Entrance " + id + ": " + getValue();
}
public static int getTotalCount(){
return count.value();
}
public static int sumEntrances(){
int sum = 0;
for (Entrance entrance: entrances)
sum += entrance.getValue();
return sum;
}
}
public class OrnamentalGarden {
public static void main(String[] args) throws InterruptedException {
ExecutorService exec = Executors.newCachedThreadPool();
for (int i = 0;i < 5; i++)
exec.execute(new Entrance(i));
TimeUnit.SECONDS.sleep(3);
Entrance.cancel();
exec.shutdown();
if (!exec.awaitTermination(250,TimeUnit.MILLISECONDS))
System.out.println("Some tasks were not terminated!");
System.out.println("Total: " + Entrance.getTotalCount());
System.out.println("Sum of Entrances: " + Entrance.sumEntrances());
}
}
/*
Entrance 0: 1 Total: 1
Entrance 4: 1 Total: 5
Entrance 3: 1 Total: 4
Entrance 2: 1 Total: 3
Entrance 1: 1 Total: 2
Entrance 0: 2 Total: 6
Entrance 4: 2 Total: 7
Entrance 3: 2 Total: 8
Entrance 2: 2 Total: 10
Entrance 4: 29 Total: 141
Entrance 2: 29 Total: 142
Entrance 1: 29 Total: 143
Entrance 3: 29 Total: 144
Entrance 0: 29 Total: 145
Entrance 4: 30 Total: 146
Entrance 2: 30 Total: 147
Entrance 1: 30 Total: 148
Entrance 3: 30 Total: 149
Entrance 0: 30 Total: 150
Stopping Entrance 4: 30
Stopping Entrance 1: 30
Stopping Entrance 2: 30
Stopping Entrance 0: 30
Stopping Entrance 3: 30
Total: 150
Sum of Entrances: 150
*/
4.2 在阻塞时终结</h2
线程状态
/**
* Thread state for a thread which has not yet started.
*/
NEW, //还没有调用start方法
/**
* Thread state for a runnable thread. A thread in the runnable
* state is executing in the Java virtual machine but it may
* be waiting for other resources from the operating system
* such as processor.
*/
RUNNABLE,//JVM可以运行的状态，可以是 在等操作系统资源的状态
/**
* Thread state for a thread blocked waiting for a monitor lock.
* A thread in the blocked state is waiting for a monitor lock
* to enter a synchronized block/method or
* reenter a synchronized block/method after calling
* {@link Object#wait() Object.wait}.
*/
BLOCKED,//监视器锁 状态 synchronized
/**
* Thread state for a waiting thread.
* A thread is in the waiting state due to calling one of the
* following methods:
* <ul>
* <li>{@link Object#wait() Object.wait} with no timeout</li>
* <li>{@link #join() Thread.join} with no timeout</li>
* <li>{@link LockSupport#park() LockSupport.park}</li>
* </ul>
*
* <p>A thread in the waiting state is waiting for another thread to
* perform a particular action.
*
* For example, a thread that has called <tt>Object.wait()</tt>
* on an object is waiting for another thread to call
* <tt>Object.notify()</tt> or <tt>Object.notifyAll()</tt> on
* that object. A thread that has called <tt>Thread.join()</tt>
* is waiting for a specified thread to terminate.
*/
WAITING, //该状态是因为调用了以下方法：Object.wait/Thread.join/LockSupport.park
/**
* Thread state for a waiting thread with a specified waiting time.
* A thread is in the timed waiting state due to calling one of
* the following methods with a specified positive waiting time:
* <ul>
* <li>{@link #sleep Thread.sleep}</li>
* <li>{@link Object#wait(long) Object.wait} with timeout</li>
* <li>{@link #join(long) Thread.join} with timeout</li>
* <li>{@link LockSupport#parkNanos LockSupport.parkNanos}</li>
* <li>{@link LockSupport#parkUntil LockSupport.parkUntil}</li>
* </ul>
*/
TIMED_WAITING,//定时等待 Thread.sleep/Object.wait/Thread.join/LockSupport.parkNanos/LockSupport.parkUntil
/**
* Thread state for a terminated thread.
* The thread has completed execution.
*/
TERMINATED; //（终止）：线程执行完成。
进入阻塞状态，可能原因如下：
- sleep进入休眠状态
- wait使任务刮起，直到线程得到notify或notifyAll消息，或signal,signalAll
- 任务在等待某个输入/输出完成
- 在某个对象上调用其同步控制方法，但对象锁不可用，因为另一个任务获取了锁。
4.3 中断
interrupt
只能中断sleep，不能中断IO操作的线程 或 synchronized 的锁
package concurrency;
import java.io.IOException;
import java.io.InputStream;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
class SleepBlocked implements Runnable{ // 可中断的阻塞示例
@Override
public void run() {
try{
TimeUnit.SECONDS.sleep(100);
}catch (InterruptedException e){ // 被中断
System.out.println("Exiting SleepBlocked.run()");
}
}
}
class IOBlocked implements Runnable{// IO 不可中断的阻塞示例
private InputStream in;
public IOBlocked(InputStream is){in = is;}
@Override
public void run() {
try {
System.out.println("Waiting for read()");
in.read();
}catch (IOException e){
if (Thread.currentThread().isInterrupted()){
System.out.println("Interrupted from blocked I/O");
}else {
throw new RuntimeException(e);
}
}
System.out.println("Exiting IOBlocked.run()");
}
}
class SynchronizedBlocked implements Runnable{// 同步 不可中断的阻塞示例
public synchronized void f(){
while (true) //从不释放锁
Thread.yield();
}
public SynchronizedBlocked(){
new Thread(){
@Override
public void run(){
f();// 先获得锁
}
}.start();
}
@Override
public void run() {// 试图调用f(),并阻塞等
System.out.println("Trying to call f()");
f();
System.out.println("Exiting SynchronizedBlocked.run()");
}
}
public class Interrupting {
private static ExecutorService exec = Executors.newCachedThreadPool();
static void test(Runnable r) throws InterruptedException {
System.out.println("=======");
Future<?> f = exec.submit(r); // 任务传给执行器，拿到等待结果对象
TimeUnit.MILLISECONDS.sleep(100);
System.out.println("Interrupting " + r.getClass().getName());
f.cancel(true); // 如果运行，则中断进行取消
System.out.println("Interrupt sent to " + r.getClass().getName());
}
public static void main(String[] args) throws InterruptedException {
test(new SleepBlocked());
test(new IOBlocked(System.in));
test(new SynchronizedBlocked());
TimeUnit.SECONDS.sleep(3);
System.out.println("Aborting with System.exit(0)");
System.exit(0);
}
}
/*
=======
Interrupting concurrency.SleepBlocked
Interrupt sent to concurrency.SleepBlocked
Exiting SleepBlocked.run()
=======
Waiting for read()
Interrupting concurrency.IOBlocked
Interrupt sent to concurrency.IOBlocked
=======
Trying to call f()
Interrupting concurrency.SynchronizedBlocked
Interrupt sent to concurrency.SynchronizedBlocked
Aborting with System.exit(0)
*/
略显笨拙但确实行之有效的解决方案：关闭任务在其上发生阻塞的底层资源
package concurrency;
import java.io.IOException;
import java.io.InputStream;
import java.net.*;
import java.util.concurrent.*;
public class CloseResource {
public static void main(String[] args) throws InterruptedException, IOException {
ExecutorService exec = Executors.newCachedThreadPool();
ServerSocket server = new ServerSocket(8080);
InputStream socketInput = new Socket("localhost",8080).getInputStream();
exec.execute(new IOBlocked(socketInput));
exec.execute(new IOBlocked(System.in));
TimeUnit.MILLISECONDS.sleep(100);
System.out.println("Shutting down all threads");
exec.shutdownNow();
TimeUnit.SECONDS.sleep(1);
System.out.println("Closing " + socketInput.getClass().getName());
socketInput.close();
TimeUnit.SECONDS.sleep(1);
System.out.println("Closing " + System.in.getClass().getName());
System.in.close();
}
}
/*
Waiting for read()
Waiting for read()
Shutting down all threads
Closing java.net.SocketInputStream
Interrupted from blocked I/O // 出现中断
Exiting IOBlocked.run()
Closing java.io.BufferedInputStream
Exiting IOBlocked.run()
*/
更人性化的NIO中断
package concurrency;
import java.io.IOException;
import java.net.*;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.ClosedByInterruptException;
import java.nio.channels.SocketChannel;
import java.util.concurrent.*;
class NIOBlocked implements Runnable{
private final SocketChannel sc;
public NIOBlocked(SocketChannel sc){this.sc = sc;}
@Override
public void run() {
try {
System.out.println("Waiting for read() in " + this);
sc.read(ByteBuffer.allocate(1));
}catch (ClosedByInterruptException e) { // 中断关闭异常
System.out.println("ClosedByInterruptException");
}catch (AsynchronousCloseException e){ // 异步闭异常
System.out.println("AsynchronousCloseException");
}catch (IOException e){
throw new RuntimeException(e);
}
System.out.println("Exiting NIOBlocked.run() " + this);
}
}
public class NIOInterruption {
public static void main(String[] args) throws IOException, InterruptedException {
ExecutorService exec = Executors.newCachedThreadPool();
ServerSocket server = new ServerSocket(8081);
InetSocketAddress isa = new InetSocketAddress("localhost",8080);
SocketChannel sc1 = SocketChannel.open(isa);
SocketChannel sc2 = SocketChannel.open(isa);
Future<?> f =exec.submit(new NIOBlocked(sc1));
exec.execute(new NIOBlocked(sc2));
exec.shutdown();
TimeUnit.SECONDS.sleep(1);
//通过cancel产生中断
f.cancel(true);
TimeUnit.SECONDS.sleep(1);
// 通过关闭通道
sc2.close();
}
}
/*
Waiting for read() in concurrency.NIOBlocked@3d33b971
Waiting for read() in concurrency.NIOBlocked@3a237724
ClosedByInterruptException
Exiting NIOBlocked.run() concurrency.NIOBlocked@3d33b971
AsynchronousCloseException
Exiting NIOBlocked.run() concurrency.NIOBlocked@3a237724
*/
被互斥阻塞
同一个互斥可以被同一个任务多次获得。
一个任务应该能调用同一个对象中的其他synchronized方法，而这个任务已经持有锁
package concurrency;
public class MultiLock {
public synchronized void f1(int count){
if (count-- > 0){
System.out.println("f1() calling f2() with count " + count);
f2(count);
}
}
public synchronized void f2(int count){
if (count-- > 0){
System.out.println("f2() calling f1() with count " + count);
f1(count);
}
}
public static void main(String[] args) {
final MultiLock multiLock = new MultiLock();
new Thread(){
@Override
public void run(){
multiLock.f1(10);
}
}.start();
}
}
/*
f1() calling f2() with count 9
f2() calling f1() with count 8
f1() calling f2() with count 7
f2() calling f1() with count 6
f1() calling f2() with count 5
f2() calling f1() with count 4
f1() calling f2() with count 3
f2() calling f1() with count 2
f1() calling f2() with count 1
*/
ReentrantLock上阻塞的任务 具备 可以中断的能力，这与synchronized方法或临界区上阻塞的任务完全不同
package concurrency;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
class BlockedMutex{
private Lock lock = new ReentrantLock();
public BlockedMutex(){
lock.lock();// 从不释放这个锁
}
public void f(){
try {
lock.lockInterruptibly(); //是可中断的获取锁方式。
System.out.println("lock acquired in f() ");
} catch (InterruptedException e) {
System.out.println("Interrupted from lock acquisition in f() ");
}
}
}
class Blocked2 implements Runnable{
BlockedMutex blocked = new BlockedMutex();
@Override
public void run() {
System.out.println("Waiting for f() in BlockedMutex");
blocked.f();
System.out.println("Broken out of blocked call");
}
}
public class Interrupting2 {
public static void main(String[] args) throws InterruptedException {
Thread t = new Thread(new Blocked2());
t.start();
TimeUnit.SECONDS.sleep(1);
System.out.println("Issuing t.interrupt()");
t.interrupt();
}
}
/*
Waiting for f() in BlockedMutex
Issuing t.interrupt()
Interrupted from lock acquisition in f()
Broken out of blocked call
*/
4.4 检查中断
package concurrency;
import java.util.concurrent.TimeUnit;
class NeedsCleanup {
private final int id;
public NeedsCleanup(int ident) {
id = ident;
System.out.println("NeedsCleanup " + id);
}
public void cleanup() {
System.out.println("Cleaning up " + id);
}
}
class Blocked3 implements Runnable {
private volatile double d = 0.0;
@Override
public void run() {
try {
while (!Thread.interrupted()) {
// point 1
NeedsCleanup n1 = new NeedsCleanup(1);
try {
System.out.println("Sleeping");
TimeUnit.SECONDS.sleep(1);
// point 2
NeedsCleanup n2 = new NeedsCleanup(2);
try {
System.out.println("Calculating");
for (int i = 1; i < 2500000; i++)
d = d + (Math.PI + Math.E) / d;
System.out.println("Finished time - consuming operation");
} finally {
n2.cleanup();
}
} finally {
n1.cleanup();
}
}
System.out.println("Exiting via while() test");
} catch (InterruptedException e) {
System.out.println("Exiting via InterruptedException");
}
}
}
public class InterruptingIdiom {
public static void main(String[] args) throws InterruptedException {
// if (args.length != 1) {
// System.out.println("Usage: java InterruptingIdiom delay-in-mS");
// System.exit(1);
// }
Thread t = new Thread(new Blocked3());
t.start();
TimeUnit.MILLISECONDS.sleep(1000);
t.interrupt();
}
}
5 线程之间的协作
5.1 wait() 和 notifyAll()
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
class Car{
private boolean waxOn = false; // 涂蜡-抛光处理的状态 waxing-polishing
public synchronized void waxed(){
waxOn = true; //Ready to buff 打蜡完成
notifyAll();
}
public synchronized void buffed(){ //抛光
waxOn = false; // Ready for another coat of wax
notifyAll();
}
public synchronized void waitForWaxing() throws InterruptedException {
while (waxOn == false)//抛光完成后，等待打蜡
wait();
}
public synchronized void waitForBuffing() throws InterruptedException {
while (waxOn == true)//打蜡完成后，等待抛光
wait();
}
}
class WaxOn implements Runnable{// 打蜡任务
private Car car;
public WaxOn(Car c){car = c;}//两个线程操作一个车
@Override
public void run() {
try {
while (!Thread.interrupted()){
System.out.println("Wax On! ");
TimeUnit.MILLISECONDS.sleep(200);
car.waxed();
car.waitForBuffing();
}
}catch (InterruptedException e){
System.out.println("Exiting via interrupt");
}
System.out.println("Ending Wax On task");
}
}
class WaxOff implements Runnable{
private Car car;
public WaxOff(Car c){car =c;}
@Override
public void run() {
try{
while (!Thread.interrupted()){
car.waitForWaxing();// 先打蜡后抛光，等待抛光
System.out.println("Wax Off! ");
TimeUnit.MILLISECONDS.sleep(200);
car.buffed();
}
}catch (InterruptedException e){
System.out.println("Exiting via interrupt");
}
System.out.println("Ending Wax Off task");
}
}
public class WaxOMatic {
public static void main(String[] args) throws InterruptedException {
Car car = new Car();
ExecutorService exec = Executors.newCachedThreadPool();
exec.execute(new WaxOff(car));
exec.execute(new WaxOn(car));
TimeUnit.SECONDS.sleep(5);
exec.shutdownNow();
}
}
/*
Wax On!
Wax Off!
Wax On!
Wax Off!
Wax On!
Wax Off!
Wax On!
Wax Off!
Wax On!
Wax Off!
Wax On!
Wax Off!
Wax On!
Wax Off!
Wax On!
Wax Off!
Wax On!
Wax Off!
Wax On!
Wax Off!
Wax On!
Wax Off!
Wax On!
Wax Off!
Wax On!
Exiting via interrupt
Ending Wax On task
Exiting via interrupt
Ending Wax Off task
*/
错失的信号
有缺陷的方式：
T1:
synchronized(sharedMonitor) {
<setup condition for T2>
sharedMonitor.notify();
}
T2:
while(someCondition) {
// Point 1
synchronized(sharedMonitor) {
sharedMonitor.wait();
}
}
正确的方式
synchronized(sharedMonitor) {
while(someCondition)
sharedMonitor.wait();
}
5.2 notify()和 notifyAll()
notifyAll 因某个特定锁而被调用时，只有等待这个锁的任务才能被唤醒。
package concurrency;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
class Blocker{
synchronized void waitingCall(){
try{
while (!Thread.interrupted()) {
wait();
System.out.println(Thread.currentThread() + " ");
}
}catch (InterruptedException e){
// ok to exit this way
}
}
synchronized void prod(){notify();}
synchronized void prodAll(){notifyAll();}
}
class Task implements Runnable{
static Blocker blocker = new Blocker();
@Override
public void run() {
blocker.waitingCall();
}
}
class Task2 implements Runnable{
static Blocker block = new Blocker();
@Override
public void run() {
block.waitingCall();
}
}
public class NotifyVsNotifyAll {
public static void main(String[] args) throws InterruptedException {
ExecutorService exec = Executors.newCachedThreadPool();
for (int i = 0; i < 5; i++)
exec.execute(new Task()); //操作 同一个静态Blocker对象
exec.execute(new Task2());// 操作 第二个Blocker对象
Timer timer = new Timer();
timer.scheduleAtFixedRate(new TimerTask() {// 定时器安排
boolean prod = true;
@Override
public void run() {
if (prod){
System.out.println("
notify() ");
Task.blocker.prod(); //唤醒
prod = false;
}else {
System.out.println("
notifyAll() ");
Task.blocker.prodAll(); // 唤醒全部
prod = true;
}
}
},400,400); //400毫秒
TimeUnit.SECONDS.sleep(5);
timer.cancel();
System.out.println("
Timer canceled ");
TimeUnit.MILLISECONDS.sleep(500);
System.out.println("Task2.blocker.prodAll() ");
Task2.block.prodAll();
TimeUnit.MILLISECONDS.sleep(500);
System.out.println("
Shutting dowm");
exec.shutdownNow(); // 中断所有任务
}
}
/*
notify()
Thread[pool-1-thread-1,5,main]
notifyAll()
Thread[pool-1-thread-1,5,main]
Thread[pool-1-thread-5,5,main]
Thread[pool-1-thread-4,5,main]
Thread[pool-1-thread-3,5,main]
Thread[pool-1-thread-2,5,main]
notify()
Thread[pool-1-thread-1,5,main]
notifyAll()
Thread[pool-1-thread-1,5,main]
Thread[pool-1-thread-2,5,main]
Thread[pool-1-thread-3,5,main]
Thread[pool-1-thread-4,5,main]
Thread[pool-1-thread-5,5,main]
notify()
Thread[pool-1-thread-1,5,main]
notifyAll()
Thread[pool-1-thread-1,5,main]
Thread[pool-1-thread-5,5,main]
Thread[pool-1-thread-4,5,main]
Thread[pool-1-thread-3,5,main]
Thread[pool-1-thread-2,5,main]
notify()
Thread[pool-1-thread-1,5,main]
notifyAll()
Thread[pool-1-thread-1,5,main]
Thread[pool-1-thread-2,5,main]
Thread[pool-1-thread-3,5,main]
Thread[pool-1-thread-4,5,main]
Thread[pool-1-thread-5,5,main]
notify()
Thread[pool-1-thread-1,5,main]
notifyAll()
Thread[pool-1-thread-1,5,main]
Thread[pool-1-thread-5,5,main]
Thread[pool-1-thread-4,5,main]
Thread[pool-1-thread-3,5,main]
Thread[pool-1-thread-2,5,main]
notify()
Thread[pool-1-thread-1,5,main]
notifyAll()
Thread[pool-1-thread-1,5,main]
Thread[pool-1-thread-2,5,main]
Thread[pool-1-thread-3,5,main]
Thread[pool-1-thread-4,5,main]
Thread[pool-1-thread-5,5,main]
Timer canceled
Task2.blocker.prodAll()
Thread[pool-1-thread-6,5,main]
Shutting dowm
*/
5.3 生产者与消费者
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
class Meal{
private final int orderNum; //序号
public Meal(int orderNum){this.orderNum = orderNum;}
public String toString(){return "Meal " + orderNum;}
}
class WaitPerson implements Runnable{
private Restaurant restaurant;
public WaitPerson(Restaurant r){restaurant =r;}
@Override
public void run() {
try {
while (!Thread.interrupted()) {
synchronized (this) {
while (restaurant.meal == null)
wait();
}
System.out.println("Waitperson got " + restaurant.meal);
synchronized (restaurant.chef){
restaurant.meal = null;
restaurant.chef.notifyAll();
}
}
}catch (InterruptedException e){
System.out.println("WaitPerson interrupted");
}
}
}
class Chef implements Runnable{
private Restaurant restaurant;
private int count = 0;
public Chef(Restaurant r){restaurant = r;}
@Override
public void run() {
try {
while (!Thread.interrupted()){
synchronized (this){
while (restaurant.meal != null)
wait();
}
if (++count ==10){
System.out.println("Out of food,closing");
restaurant.exec.shutdownNow();//发送 interrupt( )
}
System.out.print("Order up! ");
synchronized (restaurant.waitPerson){
restaurant.meal = new Meal(count);
restaurant.waitPerson.notifyAll();
}
TimeUnit.MILLISECONDS.sleep(100);//在sleep时，抛出InterruptedException，如果注释掉，则通过Thread.interrupted退出循环，不抛出异常
}
}catch (InterruptedException e){//这是一个 可中断的组设操作时候，这个中断只能抛出 InterruptedException
System.out.println("Chef interrupted");
}
}
}
public class Restaurant {
Meal meal;
ExecutorService exec = Executors.newCachedThreadPool();
WaitPerson waitPerson = new WaitPerson(this);
Chef chef = new Chef(this);
public Restaurant(){
exec.execute(chef);
exec.execute(waitPerson);
}
public static void main(String[] args) {
new Restaurant();
}
}
/*
Order up!
Waitperson got Meal 1
Order up!
Waitperson got Meal 2
Order up!
Waitperson got Meal 3
Order up!
Waitperson got Meal 4
Order up!
Waitperson got Meal 5
Order up!
Waitperson got Meal 6
Order up!
Waitperson got Meal 7
Order up!
Waitperson got Meal 8
Order up!
Waitperson got Meal 9
Out of food,closing
Order up! //这后面，需要重点理解
WaitPerson interrupted
Chef interrupted
*/
使用显式的Lock和Condition对象
package concurrency.waxomatic2;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
class Car{
private Lock lock = new ReentrantLock();
private Condition condition = lock.newCondition();//由Lock对象创建出来，Condition依赖于Lock对象
private boolean waxOn = false;
public void waxed(){
lock.lock();
try{
waxOn = true;
condition.signalAll();
}finally {
lock.unlock();
}
}
public void buffed(){
lock.lock();
try{
waxOn = false;
condition.signalAll();
}finally {
lock.unlock();
}
}
public void waitForWaxing() throws InterruptedException {
lock.lock();
try{
while (waxOn == false)
condition.await();//当前线程会释放锁，并在此等待，
// 其他线程调用singal方法后，通知当前线程后，当前线程才从awit方法中返回，并返回前已经获得了锁。
}finally {
lock.unlock();
}
}
public void waitForBuffing() throws InterruptedException {
lock.lock();
try{
while (waxOn == true)
condition.await();
}finally {
lock.unlock();
}
}
}
class WaxOn implements Runnable{
private Car car;
public WaxOn(Car c){car = c;}
@Override
public void run() {
try{
while (!Thread.interrupted()){
System.out.println("Wax on!");
TimeUnit.MILLISECONDS.sleep(200);
car.waxed();
car.waitForBuffing();
}
}catch (InterruptedException e){
System.out.println("Exiting via interrupt");
}
System.out.println("Ending Wax On task");
}
}
class WaxOff implements Runnable{
private Car car;
public WaxOff(Car c){car =c;}
@Override
public void run() {
try{
while (!Thread.interrupted()){
car.waitForWaxing();
System.out.println("Wax Off!");
TimeUnit.MILLISECONDS.sleep(200);
car.buffed();
}
}catch (InterruptedException e){
System.out.println("Exiting via interrupt");
}
System.out.println("Ending Wax Off task");
}
}
public class WaxOMatic2 {
public static void main(String[] args) throws InterruptedException {
Car car = new Car();
ExecutorService exec = Executors.newCachedThreadPool();
exec.execute(new WaxOff(car));
exec.execute(new WaxOn(car));
TimeUnit.SECONDS.sleep(5);
exec.shutdownNow();
}
}
/*
Wax on!
Wax Off!
Wax on!
Wax Off!
Wax on!
Wax Off!
Wax on!
Wax Off!
Wax on!
Wax Off!
Wax on!
Wax Off!
Wax on!
Wax Off!
Wax on!
Wax Off!
Wax on!
Wax Off!
Wax on!
Wax Off!
Wax on!
Wax Off!
Wax on!
Wax Off!
Wax on!
Exiting via interrupt
Ending Wax On task
Exiting via interrupt
Ending Wax Off task
*/
Lock和Condition对象是更加困难的多线程问题中才是必需的。
5.4 生产者-消费者 与 队列
package concurrency;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.SynchronousQueue;
class LiftOffRunner implements Runnable{
private BlockingQueue<LiftOff> rockets;
public LiftOffRunner(BlockingQueue<LiftOff> queue){
rockets = queue;
}
public void add(LiftOff lo){
try{
rockets.put(lo);
}catch (InterruptedException e){
System.out.println("Interrupted during put()");
}
}
@Override
public void run() {
try{
while (!Thread.interrupted()){
LiftOff rocked = rockets.take();
rocked.run();// use this thread 将对象中队列中推出，显式调用run()使用自己的线程运行，而不是为每个任务启动新线程
}
}catch (InterruptedException e){
System.out.println("Waking from take()");
}
System.out.println("Exiting LiftOffRunner");
}
}
public class TestBlockingQueues {
static void getkey(){
try{
new BufferedReader(new InputStreamReader(System.in)).readLine();
}catch (IOException e){
throw new RuntimeException(e);
}
}
static void getkey(String message){
System.out.println(message);
getkey();
}
static void test(String msg, BlockingQueue<LiftOff> queue){
System.out.println(msg);
LiftOffRunner runner = new LiftOffRunner(queue);
Thread t = new Thread(runner);
t.start();
for (int i = 0;i < 5; i++)
runner.add(new LiftOff(5));
getkey("Press 'Enter' (" + msg +")");
t.interrupt();
System.out.println("Finished " + msg + " test");
}
public static void main(String[] args) {
test("LinkedBlockingQueue",new LinkedBlockingQueue<LiftOff>());//无届队列 unbounded queue
test("ArrayBlockingQueue",new ArrayBlockingQueue<LiftOff>(3));//固定尺寸
test("SynchronousQueue",new SynchronousQueue<LiftOff>());
}
}
/*
LinkedBlockingQueue
Press 'Enter' (LinkedBlockingQueue)
#0(4), #0(3), #0(2), #0(1), #0(Liftoff!), #1(4), #1(3), #1(2), #1(1), #1(Liftoff!), #2(4), #2(3), #2(2), #2(1), #2(Liftoff!), #3(4), #3(3), #3(2), #3(1), #3(Liftoff!), #4(4), #4(3), #4(2), #4(1), #4(Liftoff!),
Finished LinkedBlockingQueue test
Waking from take()
Exiting LiftOffRunner
ArrayBlockingQueue
#5(4), #5(3), #5(2), #5(1), #5(Liftoff!), #6(4), Press 'Enter' (ArrayBlockingQueue)
#6(3), #6(2), #6(1), #6(Liftoff!), #7(4), #7(3), #7(2), #7(1), #7(Liftoff!), #8(4), #8(3), #8(2), #8(1), #8(Liftoff!), #9(4), #9(3), #9(2), #9(1), #9(Liftoff!),
Finished ArrayBlockingQueue test
Waking from take()
Exiting LiftOffRunner
SynchronousQueue
#10(4), #10(3), #10(2), #10(1), #10(Liftoff!), #11(4), #11(3), #11(2), #11(1), #11(Liftoff!), #12(4), #12(3), #12(2), #12(1), #12(Liftoff!), #13(4), #13(3), #13(2), #13(1), #13(Liftoff!), Press 'Enter' (SynchronousQueue)
#14(4), #14(3), #14(2), #14(1), #14(Liftoff!),
Finished SynchronousQueue test
Waking from take()
Exiting LiftOffRunner
*/
吐司BlockingQueues
package concurrency;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
class Toast{
public enum Status{DRY,BUTTERED,JAMMED}
private Status status = Status.DRY;
private final int id;
public Toast(int idn){id = idn;}
public void butter(){status = Status.BUTTERED;}
public void jam(){status = Status.JAMMED;}
public Status getStatus(){return status;}
public int getId(){return id;}
public String toString(){
return "Toast " + id + ": " + status;
}
}
class ToastQueue extends LinkedBlockingQueue<Toast>{}
class Toaster implements Runnable{// 制做吐司
private ToastQueue toastQueue;
private int count = 0;
private Random rand = new Random(47);
public Toaster(ToastQueue tq){toastQueue = tq;}
@Override
public void run() {
try{
while (!Thread.interrupted()){
TimeUnit.MILLISECONDS.sleep(100 + rand.nextInt(500));
Toast t = new Toast(count++);
System.out.println(t);
toastQueue.put(t);
}
}catch (InterruptedException e){
System.out.println("Toaster interrupted");
}
System.out.println("Toaster off");
}
}
class Butterer implements Runnable{//给吐司抹黄油
private ToastQueue dryQueue,butteredQueue;
public Butterer(ToastQueue dry,ToastQueue buttered){
dryQueue = dry;
butteredQueue = buttered;
}
@Override
public void run() {
try{
while (!Thread.interrupted()){
Toast t = dryQueue.take();
t.butter();
System.out.println(t);
butteredQueue.put(t);
}
}catch (InterruptedException e){
System.out.println("Butter interrupted");
}
System.out.println("Butterer off");
}
}
class Jammer implements Runnable{//给抹过黄油的吐司上涂果酱
private ToastQueue butteredQueue,finishedQueue;
public Jammer(ToastQueue buttered, ToastQueue finished){
butteredQueue = buttered;
finishedQueue = finished;
}
@Override
public void run() {
try{
while (!Thread.interrupted()){
Toast t = butteredQueue.take();
t.jam();
System.out.println(t);
finishedQueue.put(t);
}
}catch (InterruptedException e){
System.out.println("Jammer interrupted");
}
System.out.println("jammer off");
}
}
class Eater implements Runnable{
private ToastQueue finishedQueue;
private int counter = 0;
public Eater(ToastQueue finished){
finishedQueue = finished;
}
@Override
public void run() {
try{
while (!Thread.interrupted()){
Toast t = finishedQueue.take();
if (t.getId()!= counter++ || t.getStatus() != Toast.Status.JAMMED){
System.out.println(">>> Error: " + t);
System.exit(1);
}else
System.out.println("Chomp!" + t);
}
}catch (InterruptedException e){
System.out.println("Eater interrupted");
}
System.out.println("Eater off");
}
}
public class ToastOMatic {
public static void main(String[] args) throws InterruptedException {
ToastQueue dryQueue = new ToastQueue(),
butterredQueue = new ToastQueue(),
finishedQueue = new ToastQueue();
ExecutorService exec = Executors.newCachedThreadPool();
exec.execute(new Toaster(dryQueue));
exec.execute(new Butterer(dryQueue,butterredQueue));
exec.execute(new Jammer(butterredQueue,finishedQueue));
exec.execute(new Eater(finishedQueue));
TimeUnit.SECONDS.sleep(5);
exec.shutdownNow();
}
}
/*
Toast 0: DRY
Toast 0: BUTTERED
Toast 0: JAMMED
Chomp!Toast 0: JAMMED
Toast 1: DRY
Toast 1: BUTTERED
Toast 1: JAMMED
Chomp!Toast 1: JAMMED
Toast 2: DRY
Toast 2: BUTTERED
Toast 2: JAMMED
Chomp!Toast 2: JAMMED
Toast 3: DRY
Toast 3: BUTTERED
Toast 3: JAMMED
Chomp!Toast 3: JAMMED
Toast 4: DRY
Toast 4: BUTTERED
Toast 4: JAMMED
Chomp!Toast 4: JAMMED
Toast 5: DRY
Toast 5: BUTTERED
Toast 5: JAMMED
Chomp!Toast 5: JAMMED
Toast 6: DRY
Toast 6: BUTTERED
Toast 6: JAMMED
Chomp!Toast 6: JAMMED
Toast 7: DRY
Toast 7: BUTTERED
Toast 7: JAMMED
Chomp!Toast 7: JAMMED
Toast 8: DRY
Toast 8: BUTTERED
Toast 8: JAMMED
Chomp!Toast 8: JAMMED
Toast 9: DRY
Toast 9: BUTTERED
Toast 9: JAMMED
Chomp!Toast 9: JAMMED
Toast 10: DRY
Toast 10: BUTTERED
Toast 10: JAMMED
Chomp!Toast 10: JAMMED
Toast 11: DRY
Toast 11: BUTTERED
Toast 11: JAMMED
Chomp!Toast 11: JAMMED
Toast 12: DRY
Toast 12: BUTTERED
Toast 12: JAMMED
Chomp!Toast 12: JAMMED
Toast 13: DRY
Toast 13: BUTTERED
Toast 13: JAMMED
Chomp!Toast 13: JAMMED
Toast 14: DRY
Toast 14: BUTTERED
Toast 14: JAMMED
Chomp!Toast 14: JAMMED
Eater interrupted
Eater off
Toaster interrupted
Toaster off
Butter interrupted
Butterer off
Jammer interrupted
jammer off
*/
5.5 任务间使用管道进行输入输出
管道基本上是个阻塞队列，在引入BlockingQueue之前的多个java版本中存在
package concurrency;
import java.io.IOException;
import java.io.PipedReader;
import java.io.PipedWriter;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
class Sender implements Runnable{
private Random rand = new Random(47);
private PipedWriter out = new PipedWriter();
public PipedWriter getPipedWriter(){return out;}
@Override
public void run() {
try{
while (true)
for (char c ='A';c <= 'z'; c++){
out.write(c);
TimeUnit.MILLISECONDS.sleep(rand.nextInt(500));
}
}catch (IOException e){
System.out.println(e + " Sender write exception");
}catch (InterruptedException e){
System.out.println(e + " Sender sleep interrupted");
}
}
}
class Receiver implements Runnable{
private PipedReader in;
public Receiver(Sender sender) throws IOException {
in = new PipedReader(sender.getPipedWriter());
}
@Override
public void run() {
try{
while (true)
System.out.println("Read: " + (char)in.read() + ", ");// 当没有更多数据时候，自动阻塞
}catch (IOException e){
System.out.println(e + " Receiver read exception");
}
}
}
public class PipedIO {
public static void main(String[] args) throws IOException, InterruptedException {
Sender sender = new Sender();
Receiver receiver = new Receiver(sender);
ExecutorService exec = Executors.newCachedThreadPool();
exec.execute(sender);
exec.execute(receiver);
TimeUnit.SECONDS.sleep(4);
exec.shutdownNow();
}
}
/*
Read: A,
Read: B,
Read: C,
Read: D,
Read: E,
Read: F,
Read: G,
Read: H,
Read: I,
Read: J,
Read: K,
Read: L,
Read: M,
java.lang.InterruptedException: sleep interrupted Sender sleep interrupted
java.io.InterruptedIOException Receiver read exception
*/
6 死锁
package concurrency;
import java.util.Random;
import java.util.concurrent.TimeUnit;
public class Philosopher implements Runnable{
private Chopstick left;
private Chopstick right;
private final int id;
private final int ponderFactor;
private Random rand = new Random(47);
private void pause() throws InterruptedException {
if (ponderFactor == 0) return;
TimeUnit.MILLISECONDS.sleep(rand.nextInt(ponderFactor*250));
}
public Philosopher(Chopstick left,Chopstick right,int ident,int ponder){
this.left = left;
this.right = right;
id = ident;
ponderFactor =ponder;
}
public void run(){
try{
while (!Thread.interrupted()){
System.out.println(this + " " + "thinking");
pause();//thinking
// Philosoper becomes hungry
System.out.println(this + " " + "grabbing right");
right.take();
System.out.println(this + " " + "grbbing left");
left.take();
System.out.println(this + " " + "eating");
pause();
right.drop();
left.drop();
}
} catch (InterruptedException e) {
System.out.println(this + " " + "exiting via interrupt");
}
}
public String toString(){return "Philosopher " + id;}
}
package concurrency;
import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class DeadlockingDiningPhilosophers {
public static void main(String[] args) throws IOException, InterruptedException {
int ponder = 5;
if (args.length > 0)
ponder = Integer.parseInt(args[0]);
int size = 5;
if (args.length > 1)
size = Integer.parseInt(args[1]);// Fake it until you make it
ExecutorService exec = Executors.newCachedThreadPool();
Chopstick[] sticks = new Chopstick[size];
for (int i = 0;i < size; i++)
sticks[i] = new Chopstick();
for (int i = 0; i< size; i++)
exec.execute(new Philosopher(sticks[i],sticks[(i+1)%size],i,ponder));
if (args.length == 3 && args[2].equals("timeout"))
TimeUnit.SECONDS.sleep(5);
else {
System.out.println("Press 'Enter' to quit");
System.in.read();
}
exec.shutdownNow();
}
}
四个条件同时满足，就会发生死锁
- 互斥
- 至少有一个任务持有资源，等待获取当前被其他任务持有的资源
- 资源不能被任务抢占
- 必须有循环等待
防止死锁，破坏第4个条件
package concurrency;
import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class FixedDiningPhilosophers {
public static void main(String[] args) throws InterruptedException, IOException {
int ponder =5;
if (args.length> 0)
ponder = Integer.parseInt(args[0]);
int size = 5;
if (args.length > 1)
size = Integer.parseInt(args[1]);
ExecutorService exec = Executors.newCachedThreadPool();
Chopstick[] sticks = new Chopstick[size];
for (int i = 0; i < size; i++)
sticks[i] = new Chopstick();
for (int i = 0; i < size; i++)
if (i < (size -1))
exec.execute(new Philosopher(sticks[i],sticks[i+1],i,ponder));
else
exec.execute(new Philosopher(sticks[0],sticks[i],i,ponder));
if (args.length == 3 && args[2].equals("timeout"))
TimeUnit.SECONDS.sleep(5);
else {
System.out.println("Press 'Enter' to quit");
System.in.read();
}
exec.shutdownNow();
}
}
7 新类库中的构建
7.1 CountDownLatch
允许一个或多个线程等待 其他线程完成操作
package concurrency;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
//执行任务中的一部分
class TaskPortion implements Runnable{
private static int counter = 0;
private final int id = counter++;
private static Random rand = new Random(47);
private final CountDownLatch latch;
TaskPortion(CountDownLatch latch){this.latch = latch;}
@Override
public void run() {
try{
doWork();
latch.countDown();
}catch (InterruptedException ex){
// exit
}
}
public void doWork() throws InterruptedException {
TimeUnit.MILLISECONDS.sleep(rand.nextInt(2000));
System.out.println(this + " completed");
}
public String toString(){
return String.format("%1$-3d ",id);
}
}
class WaitingTask implements Runnable{
private static int counter = 0;
private final int id = counter++;
private final CountDownLatch latch;
WaitingTask(CountDownLatch latch){this.latch = latch;}
@Override
public void run() {
try {
latch.await();
System.out.println("Latch barrier passed for " + this);
}catch (InterruptedException ex){
System.out.println(this + " interrupted");
}
}
public String toString(){
return String.format("WaitingTask %1$-3d ",id);
}
}
public class CountDownLatchDemo {
static final int SIZE = 100;
public static void main(String[] args) {
ExecutorService exec = Executors.newCachedThreadPool();
// 所有 公用 同一个 倒计时锁
CountDownLatch latch = new CountDownLatch(SIZE);
for (int i =0;i < 10; i++)
exec.execute(new WaitingTask(latch)); // 等待任务
for (int i = 0; i < SIZE; i++)
exec.execute(new TaskPortion(latch)); // 执行任务
System.out.println("Launched all tasks");
exec.shutdown();
}
}
/*
Launched all tasks
43 completed
99 completed
36 completed
95 completed
94 completed
11 completed
21 completed
77 completed
7 completed
9 completed
75 completed
79 completed
10 completed
40 completed
96 completed
34 completed
63 completed
23 completed
29 completed
38 completed
55 completed
90 completed
88 completed
28 completed
5 completed
50 completed
8 completed
12 completed
1 completed
27 completed
98 completed
13 completed
72 completed
71 completed
2 completed
45 completed
91 completed
31 completed
14 completed
17 completed
6 completed
97 completed
35 completed
69 completed
20 completed
32 completed
4 completed
68 completed
37 completed
47 completed
87 completed
70 completed
84 completed
86 completed
66 completed
54 completed
42 completed
41 completed
46 completed
74 completed
57 completed
65 completed
0 completed
80 completed
19 completed
60 completed
15 completed
89 completed
51 completed
25 completed
53 completed
62 completed
58 completed
92 completed
76 completed
22 completed
18 completed
56 completed
85 completed
61 completed
30 completed
59 completed
24 completed
26 completed
67 completed
48 completed
39 completed
33 completed
3 completed
52 completed
93 completed
81 completed
78 completed
73 completed
44 completed
82 completed
49 completed
64 completed
83 completed
16 completed
Latch barrier passed for WaitingTask 0
Latch barrier passed for WaitingTask 4
Latch barrier passed for WaitingTask 3
Latch barrier passed for WaitingTask 7
Latch barrier passed for WaitingTask 1
Latch barrier passed for WaitingTask 2
Latch barrier passed for WaitingTask 9
Latch barrier passed for WaitingTask 8
Latch barrier passed for WaitingTask 6
Latch barrier passed for WaitingTask 5
*/
类库的线程安全
了解哪些是线程安全，JDK文档并不指出。比如Random对象，可以通过移除static限定符解决
7.2 CyclicBarrier
可循环使用的屏障（同步点）
package concurrency;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;
class Horse implements Runnable{
private static int counter = 0;
private final int id = counter++;
private int strides = 0;
private static Random rand = new Random(47);
private static CyclicBarrier barrier;
public Horse(CyclicBarrier b){barrier = b;}
public synchronized int getStrides(){return strides;}
@Override
public void run() {
try{
while (!Thread.interrupted()){
synchronized (this){
strides += rand.nextInt(3);// 0,1,2
}
barrier.await();// 循环阻碍,告诉CyclicBarrier，我已经到达，当前线程被阻塞
}
}catch (InterruptedException e){
// A legitimate way to exit
}catch (BrokenBarrierException e){
throw new RuntimeException(e);
}
}
public String toString(){return "Horse " + id + " ";}
public String tracks(){
StringBuilder s = new StringBuilder();
for (int i = 0; i < getStrides();i++)
s.append("*");
s.append(id);
return s.toString();
}
}
public class HorseRace {
static final int FINISH_LINE =75;//终点线
private List<Horse> horses = new ArrayList<>();
private ExecutorService exec = Executors.newCachedThreadPool();
private CyclicBarrier barrier;
public HorseRace(int nHorses,final int pause){
barrier = new CyclicBarrier(nHorses, new Runnable() {// 阻碍时候，需要执行的任务
@Override
public void run() {
StringBuilder s = new StringBuilder();
for (int i =0; i < FINISH_LINE; i++)
s.append("=");
System.out.println(s);
for (Horse horse:horses)
System.out.println(horse.tracks());
for (Horse horse:horses)
if (horse.getStrides() >= FINISH_LINE) {// 有马赢了，退出
System.out.println(horse + "won!");
exec.shutdownNow();
return;
}
try{
TimeUnit.MILLISECONDS.sleep(pause);
}catch (InterruptedException e){
System.out.println("barrier-action sleep interrupted");
}
}
});
for (int i =0; i < nHorses; i++){
Horse horse = new Horse(barrier);
horses.add(horse);
exec.execute(horse);
}
}
public static void main(String[] args) {
int nHorses = 7;
int pause = 200;
if (args.length > 0){
int n = new Integer(args[0]);
nHorses = n > 0 ? n : nHorses;
}
if (args.length>1){
int p = new Integer(args[1]);
pause = p > -1 ? p : pause;
}
new HorseRace(nHorses,pause);
}
}
/*
===========================================================================
*************************************************************************0
*******************************************************1
****************************************************************2
****************************************************************************3
**************************************************************************4
***********************************************************************5
************************************************************6
Horse 3 won!
*/
7.3 DelayQueue
延迟队列
package concurrency;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;
class DelayedTask implements Runnable, Delayed {
private static int counter = 0;
private final int id = counter++;
private final int delta;//延迟时间
private final long trigger;//long 触发时间
protected static List<DelayedTask> sequence = new ArrayList<>();
public DelayedTask(int delayInMilliseconds) {//毫秒单位
delta = delayInMilliseconds;
trigger = System.nanoTime() + TimeUnit.NANOSECONDS.convert(delta, TimeUnit.MILLISECONDS);
sequence.add(this);
}
@Override
public int compareTo(Delayed arg) {
DelayedTask that = (DelayedTask) arg;
if (trigger < that.trigger) return -1;
if (trigger > that.trigger) return 1;
return 0;
}
@Override
public void run() {
System.out.println(this + " ");
}
@Override
public long getDelay(TimeUnit unit) {
return unit.convert(trigger - System.nanoTime(), TimeUnit.NANOSECONDS);
}
public String toString() {
return String.format("[%1$-4d]", delta) + " Task " + id;
}
public String summary() { // 摘要
return "(" + id + ":" + delta + ")";
}
public static class EndSentinel extends DelayedTask {
private ExecutorService exec;
public EndSentinel(int delay, ExecutorService e) {
super(delay);
exec = e;
}
public void run() {
for (DelayedTask pt : sequence) {
System.out.print(pt.summary() + " ");
}
System.out.println();
System.out.println(this + " Calling shutdownNown()");
exec.shutdownNow();
}
}
}
class DelayedTaskConsumer implements Runnable {
private DelayQueue<DelayedTask> q;
public DelayedTaskConsumer(DelayQueue<DelayedTask> q) {
this.q = q;
}
@Override
public void run() {
try {
while (!Thread.interrupted())
q.take().run();
}catch (InterruptedException e){
// exit
}
System.out.println("Finished DelayedTaskConsumer");
}
}
public class DelayQueueDemo {
public static void main(String[] args) {
Random rand = new Random(47);
ExecutorService exec = Executors.newCachedThreadPool();
DelayQueue<DelayedTask> queue = new DelayQueue<>();
for (int i = 0; i < 20; i++)
queue.put(new DelayedTask(rand.nextInt(5000)));
queue.add(new DelayedTask.EndSentinel(5000, exec));
exec.execute(new DelayedTaskConsumer(queue));
}
}
/*
[128 ] Task 11 [200 ] Task 7 [429 ] Task 5 [520 ] Task 18 [555 ] Task 1 [961 ] Task 4 [998 ] Task 16 [1207] Task 9 [1693] Task 2 [1809] Task 14 [1861] Task 3 [2278] Task 15 [3288] Task 10 [3551] Task 12 [4258] Task 0 [4258] Task 19 [4522] Task 8 [4589] Task 13 [4861] Task 17 [4868] Task 6 (0:4258) (1:555) (2:1693) (3:1861) (4:961) (5:429) (6:4868) (7:200) (8:4522) (9:1207) (10:3288) (11:128) (12:3551) (13:4589) (14:1809) (15:2278) (16:998) (17:4861) (18:520) (19:4258) (20:5000)
[5000] Task 20 Calling shutdownNown()
Finished DelayedTaskConsumer
*/
7.4 PriorityBlockingQueue
优先级阻塞队列
package concurrency;
import java.util.*;
import java.util.concurrent.*;
class PrioritizedTask implements Runnable, Comparable<PrioritizedTask> {
private Random rand = new Random(47);
private static int counter = 0;
private final int id = counter++;
private final int priority;
protected static List<PrioritizedTask> sequence = new ArrayList<>();
public PrioritizedTask(int priority) {
this.priority = priority;
sequence.add(this);
}
@Override
public int compareTo(PrioritizedTask arg) {
return priority < arg.priority ? -1 : (priority > arg.priority ? -1 : 0);
}
@Override
public void run() {
try {
TimeUnit.MILLISECONDS.sleep(rand.nextInt(250));
} catch (InterruptedException e) {
// exit
}
System.out.println(this);
}
public String toString() {
return String.format("[%1$-3d]", priority) + " Task " + id;
}
public String summary() {
return "(" + id + ":" + priority + ")";
}
public static class EndSentinel extends PrioritizedTask {
private ExecutorService exec;
public EndSentinel(ExecutorService e) {
super(-1);// 优先级最低
exec = e;
}
public void run() {
int count = 0;
for (PrioritizedTask pt : sequence) {
System.out.print(pt.summary());
if (++count % 5 == 0)
System.out.println();
}
System.out.println();
System.out.println(this + " Calling shutdownNow()");
exec.shutdownNow();
}
}
}
class PrioritizedTaskProducer implements Runnable {
private Random rand = new Random(47);
private Queue<Runnable> queue;
private ExecutorService exec;
PrioritizedTaskProducer(PriorityBlockingQueue<Runnable> q, ExecutorService e) {
queue = q;
exec = e; // 用于 EndSentinel
}
@Override
public void run() {
//无界队列，永不阻塞
// 随机优先级 快速填充
for (int i = 0; i < 20; i++) {
queue.add(new PrioritizedTask(rand.nextInt(10)));
Thread.yield();
}
try {
//Trickle in highest-priority jobs:
for (int i = 0; i < 10; i++) {
TimeUnit.MILLISECONDS.sleep(250);
queue.add(new PrioritizedTask(10));
}
// Add jobs, lowest priority first:
for (int i = 0; i < 10; i++) {
queue.add(new PrioritizedTask(i));
}
// A sentinel to stop all the tasks:
queue.add(new PrioritizedTask.EndSentinel(exec));
} catch (InterruptedException e) {
// exit
System.out.println("Finished PrioritizedTaskProducer");
}
}
}
class PrioritizedTaskConsumer implements Runnable {
private PriorityBlockingQueue<Runnable> q;
public PrioritizedTaskConsumer(PriorityBlockingQueue<Runnable> q) {
this.q = q;
}
@Override
public void run() {
try {
while (!Thread.interrupted())
q.take().run();// 优先级大，先运行
} catch (InterruptedException e) {
// exit
}
System.out.println("Finished PrioritizedTaskConsumer");
}
}
public class PriorityBlockingQueueDemo {
public static void main(String[] args) {
Random rand = new Random(47);
ExecutorService exec = Executors.newCachedThreadPool();
PriorityBlockingQueue<Runnable> queue = new PriorityBlockingQueue<>();
exec.execute(new PrioritizedTaskProducer(queue, exec));
exec.execute(new PrioritizedTaskConsumer(queue));
}
}
/*
[7 ] Task 9
[0 ] Task 18
[8 ] Task 19
[1 ] Task 4
[8 ] Task 0
[8 ] Task 16
[8 ] Task 15
[1 ] Task 3
[9 ] Task 13
[5 ] Task 1
[2 ] Task 8
[0 ] Task 7
[8 ] Task 11
[8 ] Task 6
[8 ] Task 10
[9 ] Task 5
[3 ] Task 2
[9 ] Task 14
[1 ] Task 12
[1 ] Task 17
[10 ] Task 20
[10 ] Task 21
[10 ] Task 22
[10 ] Task 23
[10 ] Task 24
[10 ] Task 25
[10 ] Task 26
[10 ] Task 27
[10 ] Task 28
[2 ] Task 32
(0:8)(1:5)(2:3)(3:1)(4:1)
(5:9)(6:8)(7:0)(8:2)(9:7)
(10:8)(11:8)(12:1)(13:9)(14:9)
(15:8)(16:8)(17:1)(18:0)(19:8)
(20:10)(21:10)(22:10)(23:10)(24:10)
(25:10)(26:10)(27:10)(28:10)(29:10)
(30:0)(31:1)(32:2)(33:3)(34:4)
(35:5)(36:6)(37:7)(38:8)(39:9)
(40:-1)
[-1 ] Task 40 Calling shutdownNow()
Finished PrioritizedTaskConsumer
*/
7.5 使用ScheduledExecutor的温室控制器
预定执行器
package concurrency;
import java.util.*;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class GreenhouseScheduler {
private volatile boolean light =false;
private volatile boolean water = false;
private String thermostat = "Day";
public synchronized String getThermostat(){
return thermostat;
}
public synchronized void setThermostat(String value){
thermostat = value;
}
ScheduledThreadPoolExecutor sheduler = new ScheduledThreadPoolExecutor(10);
public void schedule(Runnable event,long delay){
sheduler.schedule(event,delay, TimeUnit.MILLISECONDS);//执行一次
}
public void repeat(Runnable event,long initailDelay,long period){
sheduler.scheduleAtFixedRate(event,initailDelay,period,TimeUnit.MILLISECONDS);//重复执行
}
class LightOn implements Runnable{
@Override
public void run() {
// Put hardware control code here to
// physically turn on the light.
System.out.println("Turning on lights");
light = true;
}
}
class LightOff implements Runnable{
@Override
public void run() {
System.out.println("Turning off Lights");
light = true;
}
}
class WaterOn implements Runnable{
@Override
public void run() {
System.out.println("Turning greenhouse water on");
water = true;
}
}
class WaterOff implements Runnable{
@Override
public void run() {
System.out.println("Turning greenhouse water off");
water = false;
}
}
class ThermostatNight implements Runnable{
@Override
public void run() {
System.out.println("Thermostat to Night setting");
setThermostat("Night");
}
}
class ThermostatDay implements Runnable{
@Override
public void run() {
System.out.println("Thermostat to Day setting");
setThermostat("Day");
}
}
class Bell implements Runnable{
@Override
public void run() {
System.out.println("Bing!");
}
}
class Terminate implements Runnable{
@Override
public void run() {
System.out.println("Terminating");
sheduler.shutdownNow();
// Must start a separate task to do this job,
// since the scheduler has been shut down:
new Thread(){
public void run(){
for (DataPoint d: data)
System.out.println(d);
}
}.start();
}
}
static class DataPoint{
final Calendar time;
final float temperature;
final float humidity;//湿度
public DataPoint(Calendar d,float temp,float hum){
time = d;
temperature = temp;
humidity = hum;
}
public String toString(){
return time.getTime() + String.format(" temperature: %1$.1f humidity: %2$.2f",temperature,humidity);
}
}
private Calendar lastTime = Calendar.getInstance();//当前时间
{
// Adjust date to the half hour
lastTime.set(Calendar.MINUTE,30);
lastTime.set(Calendar.SECOND,00);
}
private float lastTemp = 65.0f;
private int tempDirection = +1;
private float lastHumidity = 50.0f;
private int humidityDirection = +1;
private Random rand = new Random(47);
List<DataPoint> data = Collections.synchronizedList(new ArrayList<>());
class CollectData implements Runnable{
@Override
public void run() {
System.out.println("Collecting data");
synchronized (GreenhouseScheduler.this){
// Pretend the interval is longer than it is:
lastTime.set(Calendar.MINUTE,lastTime.get(Calendar.MINUTE) + 30);
// One in 5 chances of reversing the direction:
if (rand.nextInt(5) == 4)
tempDirection = - tempDirection;
// Store previous value:
lastTemp = lastTemp + tempDirection * (1.0f + rand.nextFloat());
if (rand.nextInt(5) ==4)
humidityDirection = - humidityDirection;
lastHumidity = lastHumidity + humidityDirection * rand.nextFloat();
// Calendar must be cloned, otherwise all
// DataPoints hold references to the same lastTime.
// For a basic object like Calendar, clone() is OK.
data.add(new DataPoint((Calendar)lastTime.clone(),lastTemp,lastHumidity));
}
}
}
public static void main(String[] args) {
GreenhouseScheduler gh = new GreenhouseScheduler();
gh.schedule(gh.new Terminate(),5000);// 5秒钟结束
gh.repeat(gh.new Bell(),0,1000);//每隔1s
gh.repeat(gh.new ThermostatNight(),0,2000);
gh.repeat(gh.new LightOn(),0,200);
gh.repeat(gh.new LightOff(),0,400);
gh.repeat(gh.new WaterOn(),0,600);
gh.repeat(gh.new WaterOff(),0,800);
gh.repeat(gh.new ThermostatDay(),0,1400);
gh.repeat(gh.new CollectData(),500,500);//500毫秒后开始收集数据，每隔500毫秒收集
}
}
/*
Bing!
Thermostat to Night setting
Turning on lights
Turning off Lights
Turning greenhouse water on
Turning greenhouse water off
Thermostat to Day setting
Turning on lights
Turning on lights
Turning off Lights
Collecting data
Turning on lights
Turning greenhouse water on
Turning on lights
Turning off Lights
Turning greenhouse water off
Bing!
Turning on lights
Collecting data
Turning on lights
Turning off Lights
Turning greenhouse water on
Turning on lights
Thermostat to Day setting
Collecting data
Turning on lights
Turning greenhouse water off
Turning off Lights
Turning on lights
Turning greenhouse water on
Bing!
Thermostat to Night setting
Turning on lights
Turning off Lights
Collecting data
Turning on lights
Turning on lights
Turning off Lights
Turning greenhouse water on
Turning greenhouse water off
Collecting data
Turning on lights
Turning on lights
Turning off Lights
Thermostat to Day setting
Bing!
Turning greenhouse water on
Collecting data
Turning on lights
Turning on lights
Turning off Lights
Turning greenhouse water off
Turning on lights
Collecting data
Turning on lights
Turning off Lights
Turning greenhouse water on
Turning on lights
Bing!
Thermostat to Night setting
Turning on lights
Turning off Lights
Collecting data
Turning greenhouse water off
Turning on lights
Turning greenhouse water on
Thermostat to Day setting
Turning on lights
Turning off Lights
Collecting data
Turning on lights
Turning on lights
Turning off Lights
Turning greenhouse water off
Turning greenhouse water on
Terminating
Turning on lights
Bing!
Fri Jan 24 15:00:00 CST 2020 temperature: 66.4 humidity: 50.05
Fri Jan 24 15:30:00 CST 2020 temperature: 68.0 humidity: 50.47
Fri Jan 24 16:00:00 CST 2020 temperature: 69.7 humidity: 51.42
Fri Jan 24 16:30:00 CST 2020 temperature: 70.8 humidity: 50.87
Fri Jan 24 17:00:00 CST 2020 temperature: 72.0 humidity: 50.32
Fri Jan 24 17:30:00 CST 2020 temperature: 73.2 humidity: 49.92
Fri Jan 24 18:00:00 CST 2020 temperature: 71.9 humidity: 49.81
Fri Jan 24 18:30:00 CST 2020 temperature: 70.1 humidity: 50.25
Fri Jan 24 19:00:00 CST 2020 temperature: 68.9 humidity: 51.00
*/
7.6 Semaphore
Semaphore（信号量）是用来控制同时访问特定资源的线程数量，通过协调各个线程，以保证合理的使用公共资源。
package concurrency;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Semaphore;
public class Pool<T> {
private int size;
private List<T> items = new ArrayList<>();
private volatile boolean[] checkedOut;
private Semaphore available;
public Pool(Class<T> classObject,int size){
this.size = size;
checkedOut = new boolean[size];
available = new Semaphore(size,true);// true 按先进先出顺序
// Load pool with objects that can be checked out:
for (int i = 0; i < size; ++i)
try{
// Assumes a default constructor:
items.add(classObject.newInstance());
}catch (Exception e){
throw new RuntimeException(e);
}
}
public T checkOut() throws InterruptedException { //使用对象时，签出它们
available.acquire();//没拿到，阻塞；拿到，执行getItem
return getItem();
}
public void checkIn(T x){//使用完毕，签回它们
if (releaseItem(x))
available.release();
}
private synchronized T getItem(){
for (int i =0 ; i < size; ++i)
if (!checkedOut[i]){
checkedOut[i] = true;
return items.get(i);
}
return null; // Semaphore prevents reaching here
}
private synchronized boolean releaseItem(T item){
int index = items.indexOf(item);
if(index == -1)return false;// Not in the list 不在容器
if (checkedOut[index]){ // 如果被签出，则允许释放
checkedOut[index] = false;
return true;
}
return false;// Wasn’t checked out
}
}
package concurrency;
public class Fat { // 放池管理的对象
private volatile double d;// Prevent optimization
private static int counter = 0;
private final int id = counter++;
public Fat() {
// Expensive, interruptible operation:
for (int i = 1; i < 10000; i++) {
d += (Math.PI + Math.E) / (double) i;
}
}
public void operation() {
System.out.println(this);
}
public String toString() {
return "Fat id: " + id;
}
}
package concurrency;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
// A task to check a resource out of a pool:
class CheckoutTask<T> implements Runnable{
private static int counter = 0;
private final int id = counter++;
private Pool<T> pool;
public CheckoutTask(Pool<T> pool){
this.pool = pool;
}
@Override
public void run() {
try {
T item = pool.checkOut();
System.out.println(this + "checked out " + item);
TimeUnit.SECONDS.sleep(1);
System.out.println(this + "checking in " + item);
pool.checkIn(item);
}catch (InterruptedException e){
//terminate
}
}
public String toString(){
return "CheckoutTask " + id + " ";
}
}
public class SemaphoreDemo {
final static int SIZE = 25;
public static void main(String[] args) throws InterruptedException {
final Pool<Fat> pool = new Pool<>(Fat.class,SIZE);
ExecutorService exec = Executors.newCachedThreadPool();
for (int i = 0; i < SIZE; i++)
exec.execute(new CheckoutTask<>(pool));
System.out.println("All checkoutTasks created-----------------------------");
List<Fat> list = new ArrayList<>();
for (int i = 0; i < SIZE; i++){
Fat f = pool.checkOut();
System.out.println(i + ": main() thread checked out ");
f.operation();
list.add(f);
}
Future<?> blocked = exec.submit(new Runnable() {
@Override
public void run() {
try{
// Semaphore prevents additional checkout,
// so call is blocked:
pool.checkOut();
}catch (InterruptedException e){
System.out.println("checkOut() Interrupted");
}
}
});
TimeUnit.SECONDS.sleep(2);
blocked.cancel(true);
System.out.println("Checking in objecdts in " + list);
for (Fat f: list)
pool.checkIn(f);
for (Fat f: list)
pool.checkIn(f); // Second checkIn ignored
exec.shutdown();
}
}
7.6 Exchanger
package art;
import concurrency.Fat;
import net.mindview.util.BasicGenerator;
import net.mindview.util.Generator;
import java.util.List;
import java.util.concurrent.*;
class ExchangerProducer<T> implements Runnable{
private Generator<T> generator;
private Exchanger<List<T>> exchanger;
private List<T> holder;
ExchangerProducer(Exchanger<List<T>> exchg, Generator<T> gen,List<T> holder){
exchanger = exchg;
generator = gen;
this.holder = holder;
}
@Override
public void run() {
try{
while (!Thread.interrupted()){
for (int i =0; i < ExchangerDemo.size; i++) {
holder.add(generator.next());
System.out.println("producer " + i);
}
// Exchange full for empty:
holder = exchanger.exchange(holder);
}
}catch (InterruptedException e){
// OK to terminate this way
}
}
}
class ExchangerConsumer<T> implements Runnable{
private Exchanger<List<T>> exchanger;
private List<T> holder;
private volatile T value;
ExchangerConsumer(Exchanger<List<T>> ex, List<T> holder){
exchanger = ex;
this.holder = holder;
}
@Override
public void run() {
try{
while (!Thread.interrupted()){
holder = exchanger.exchange(holder);
for (T x: holder){
value = x; //Fetch out value
System.out.println(value);
holder.remove(x);// OK for CopyOnWriteArrayList
}
}
}catch (InterruptedException e){
// ok to terminate this way
}
System.out.println("Final value: " + value);
}
}
public class ExchangerDemo {
static int size = 10;
static int delay = 5;
public static void main(String[] args) throws InterruptedException {
if (args.length > 0)
size = new Integer(args[0]);
if (args.length > 1)
delay = new Integer(args[1]);
ExecutorService exec = Executors.newCachedThreadPool();
Exchanger<List<Fat>> xc = new Exchanger<>();
List<Fat> producerList = new CopyOnWriteArrayList<>(),//一文解读CopyOnWriteArrayList https://www.nowcoder.com/discuss/232479?type=1
consumerList = new CopyOnWriteArrayList<>();
exec.execute(new ExchangerProducer<Fat>(xc, BasicGenerator.create(Fat.class),producerList));
exec.execute(new ExchangerConsumer<Fat>(xc,consumerList));
TimeUnit.MILLISECONDS.sleep(delay);
exec.shutdownNow();
}
}
/*
producer 0
producer 1
producer 2
producer 3
producer 4
producer 5
producer 6
producer 7
producer 8
producer 9
Fat id: 0
Fat id: 1
producer 0
Fat id: 2
Fat id: 3
Fat id: 4
Fat id: 5
Fat id: 6
producer 1
Fat id: 7
Fat id: 8
Fat id: 9
producer 2
producer 3
producer 4
Final value: Fat id: 9
producer 5
producer 6
producer 7
producer 8
producer 9
*/
8 仿真
8.1 银行出纳员仿真
package concurrency;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;
class Customer{
private final int serviceTime;
public Customer(int tm){serviceTime = tm;}
public int getServiceTime(){return serviceTime;}
public String toString(){
return "[" + serviceTime + "]";
}
}
class CustomerLine extends ArrayBlockingQueue<Customer>{
// ArrayBlockingQueue 数组实现的有界阻塞队列, 此队列按照先进先出（FIFO）的原则
// 对元素进行排序。
public CustomerLine(int maxLineSize){
super(maxLineSize);//capacity： 表示数组的长度，也就是队列的长度 fair：表示是否为公平的阻塞队列，默认情况下构造的是非公平的阻塞队列。
}
public String toString(){
if (this.size() == 0) return "[Empty]";
StringBuilder result = new StringBuilder();
for (Customer customer:this)
result.append(customer);
return result.toString();
}
}
class CustomerGenerator implements Runnable{
private CustomerLine customers;
private static Random rand = new Random(47);
public CustomerGenerator(CustomerLine cq){
customers = cq;
}
@Override
public void run() {
try{
while (!Thread.interrupted()){
TimeUnit.MILLISECONDS.sleep(rand.nextInt(300));
customers.put(new Customer(rand.nextInt(1000)));
// put(e) ：当阻塞队列满了以后，生产者继续通过 put添加元素，
// 队列会一直阻塞生产者线程，直到队列可用
}
}catch (InterruptedException e){
System.out.println("CustomerGenerator interrupted");
}
System.out.println("CustomerGenerator terminating");
}
}
class Teller implements Runnable,Comparable<Teller>{
private static int counter = 0;
private final int id = counter++;
// Customers served during this shift:
private int customerServed = 0;// 服务的客户数量
private CustomerLine customers;
private boolean servingCustomerLine = true;
public Teller(CustomerLine cq){customers = cq;}
// Used by priority queue:
@Override
public synchronized int compareTo(Teller other) {
return Integer.compare(customerServed, other.customerServed);
}
@Override
public void run() {
try{
while (!Thread.interrupted()){
Customer customer = customers.take();
//take()：基于阻塞的方式获取队列中的元素，如果队列为空，则 take 方法会一直阻塞，直到队列中有新的数据可以消费
TimeUnit.MILLISECONDS.sleep(customer.getServiceTime());
synchronized (this){
customerServed++;
while (!servingCustomerLine)
wait();
}
}
}catch (InterruptedException e){
System.out.println(this + "interrupted");
}
System.out.println(this + " terminating");
}
public synchronized void doSomethingElse(){
customerServed = 0;
servingCustomerLine = false;
}
public synchronized void serveCustomerLine(){
assert !servingCustomerLine:"already serving: " + this;
servingCustomerLine = true;
notifyAll();
}
public String toString(){
return "Teller " + id + " ";
}
public String shortString(){return "T" + id;}
}
class TellerManager implements Runnable{
private ExecutorService exec;
private CustomerLine customers;
private PriorityQueue<Teller> workingTellers = new PriorityQueue<>();
private Queue<Teller> tellersDoingOtherThings = new LinkedList<>();
private int adjustmentPeriod;//毫秒
private static Random rand = new Random(47);
public TellerManager(ExecutorService e,CustomerLine customers,int adjustmentPeriod){
exec = e;
this.customers = customers;
this.adjustmentPeriod = adjustmentPeriod;
// Start with a single teller:
Teller teller= new Teller(customers);
exec.execute(teller);
workingTellers.add(teller);
}
public void adjustTellerNumer(){
// This is actually a control system. By adjusting
// the numbers, you can reveal stability issues in
// the control mechanism.
// If line is too long, add another teller:
if (customers.size() / workingTellers.size() > 2){
// If tellers are on break or doing
// another job, bring one back:
if (tellersDoingOtherThings.size() > 0){
Teller teller = tellersDoingOtherThings.remove();
teller.serveCustomerLine();
workingTellers.offer(teller);
return;
}
// Else create (hire) a new teller
Teller teller = new Teller(customers);
exec.execute(teller);
workingTellers.add(teller);
return;
}
// If line is short enough, remove a teller:
if (workingTellers.size() > 1 &&
customers.size() / workingTellers.size() < 2)
reassignOneTeller();
// If there is no line, we only need one teller:
if (customers.size() == 0)
while (workingTellers.size() > 1)
reassignOneTeller();
}
// Give a teller a different job or a break:
private void reassignOneTeller(){
Teller teller = workingTellers.poll();
teller.doSomethingElse();
tellersDoingOtherThings.offer(teller);
}
@Override
public void run() {
try{
while (!Thread.interrupted()){
TimeUnit.MILLISECONDS.sleep(adjustmentPeriod);
adjustTellerNumer();
System.out.print(customers + " { ");
for (Teller teller: workingTellers)
System.out.print(teller.shortString() + " ");
System.out.print("}");
System.out.print(" tellersDoingOtherThings：" + " { ");
for (Teller teller: tellersDoingOtherThings)
System.out.print(teller.shortString() + " ");
System.out.println("}");
}
}catch (InterruptedException e){
System.out.println(this + "interrupted");
}
System.out.println(this + "terminating");
}
public String toString(){
return "TellerManager ";
}
}
public class BankTellerSimulation {
static final int MAX_LINE_SIZE = 50;
static final int ADJUSTMENT_RERIOD = 1000;
public static void main(String[] args) throws InterruptedException, IOException {
ExecutorService exec = Executors.newCachedThreadPool();
// If line is too long, customers will leave:
CustomerLine customers = new CustomerLine(MAX_LINE_SIZE);
exec.execute(new CustomerGenerator(customers));
// Manager will add and remove tellers as necessary:
exec.execute(new TellerManager(exec,customers,ADJUSTMENT_RERIOD));
if (args.length > 0)// Optional argument
TimeUnit.SECONDS.sleep(new Integer(args[0]));
else {
System.out.println("Press 'Enter' to quit" );
System.in.read();
}
exec.shutdownNow();
}
}
/*
Press 'Enter' to quit
[200][207] { T1 T0 } tellersDoingOtherThings： { }
[861][258][140] { T0 } tellersDoingOtherThings： { T1 }
[140][322][383][575][342][804][826][896][984] { T1 T0 } tellersDoingOtherThings： { }
[826][896][984][810][141][12][689][992][976][368][395] { T2 T0 T1 } tellersDoingOtherThings： { }
[12][689][992][976][368][395][354][222][687][634][317][242] { T3 T2 T1 T0 } tellersDoingOtherThings： { }
[354][222][687][634][317][242][698][899][665][909][209][158][273][894] { T4 T3 T1 T0 T2 } tellersDoingOtherThings： { }
[273][894][984][533][8][328][779][882] { T2 T3 T1 T0 } tellersDoingOtherThings： { T4 }
[882][37][871][17][828][696][994][419][738][434][106][718][965] { T4 T2 T1 T0 T3 } tellersDoingOtherThings： { }
[434][106][718][965][416][217][677][146][552][244][111][934][843][585] { T4 T2 T1 T0 T3 } tellersDoingOtherThings： { }
[111][934][843][585][966][615][913][921][213][409] { T4 T2 T1 T0 T3 } tellersDoingOtherThings： { }
[921][213][409][983][158][225][326][851][198][505] { T4 T2 T1 T0 T3 } tellersDoingOtherThings： { }
[933][110][371][529] { T3 T2 T1 T0 } tellersDoingOtherThings： { T4 }
*/
8.2 饭店仿真
package concurrency.restaurant2;
import enumerated.menu.*;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;
// This is given to the waiter, who gives it to the chef:
class Order{// (A data-transfer object)
private static int counter = 0;
private final int id = counter++;
private final Customer customer;
private final WaitPerson waitPerson;
private final Food food;
public Order(Customer cust,WaitPerson wp,Food f){
customer =cust;
waitPerson = wp;
food = f;
}
public Food item(){return food;}
public Customer getCustomer(){return customer;}
public WaitPerson getWaitPerson(){return waitPerson;}
public String toString(){return "Order: " + id + " item: " + food + " for: " + customer +
" served by: " + waitPerson;}
}
// This is what comes back from the chef:
class Plate{
private final Order order;
private final Food food;
public Plate(Order ord,Food f){
order = ord;
food =f;
}
public Order getOrder(){return order;}
public Food getFood(){return food;}
public String toString(){return food.toString();}
}
class Customer implements Runnable{
private static int counter = 0;
private final int id = counter++;
private final WaitPerson waitPerson;
// Only one course at a time can be received:
private SynchronousQueue<Plate> placeSetting = new SynchronousQueue<>();
public Customer(WaitPerson w){waitPerson = w;}
public void deliver(Plate p) throws InterruptedException {
placeSetting.put(p);
}
@Override
public void run() {
for (Course course:Course.values()){
Food food = course.randomSelection();
try{
waitPerson.placeOrder(this,food);
// Blocks until course has been delivered:
System.out.println(this + "eating " + placeSetting.take());
}catch (InterruptedException e){
System.out.println(this + "waiting for " + course + " interrupted" );
break;
}
}
System.out.println(this + "finished meal,leaving");
}
public String toString(){
return "Customer " + id + " ";
}
}
class WaitPerson implements Runnable{
private static int counter =0;
private final int id = counter++;
private final Restaurant restaurant;
BlockingQueue<Plate> filledOrders = new LinkedBlockingQueue<>();
public WaitPerson(Restaurant rest){restaurant = rest;}
public void placeOrder(Customer cust,Food food){
try{
// Shouldn’t actually block because this is
// a LinkedBlockingQueue with no size limit:
restaurant.orders.put(new Order(cust,this,food));
}catch (InterruptedException e){
System.out.println(this + " placeOrder interrupted");
}
}
@Override
public void run() {
try{
while (!Thread.interrupted()){
// Blocks until a course is ready
Plate plate = filledOrders.take();
System.out.println(this + "received " + plate + " delivering to " + plate.getOrder().getCustomer());
plate.getOrder().getCustomer().deliver(plate);
}
}catch (InterruptedException e){
System.out.println(this + " interrupted");
}
System.out.println(this + " off duty");
}
public String toString(){
return "WaitPerson " + id + " ";
}
}
class Chef implements Runnable{
private static int counter = 0;
private final int id = counter++;
private final Restaurant restaurant;
private static Random rand = new Random(47);
public Chef(Restaurant rest){restaurant = rest;}
@Override
public void run() {
try{
while (!Thread.interrupted()){
// Blocks until an order appears:
Order order = restaurant.orders.take();
Food requestedItem = order.item();
// Time to prepare order:
TimeUnit.MILLISECONDS.sleep(rand.nextInt(500));
Plate plate = new Plate(order,requestedItem);
order.getWaitPerson().filledOrders.put(plate);
}
}catch (InterruptedException e){
System.out.println(this + " interrupted");
}
System.out.println(this + " off duty");
}
public String toString(){return "Chef " + id + " ";}
}
class Restaurant implements Runnable{
private List<WaitPerson> waitPersons = new ArrayList<>();
private List<Chef> chefs = new ArrayList<>();
private ExecutorService exec;
private static Random rand = new Random(47);
BlockingQueue<Order> orders = new LinkedBlockingQueue<>();
public Restaurant(ExecutorService e,int nWaitPersons,int nChefs){
exec = e;
for (int i = 0; i< nWaitPersons; i++){
WaitPerson waitPerson = new WaitPerson(this);
waitPersons.add(waitPerson);
exec.execute(waitPerson);
}
for (int i = 0; i < nChefs; i++){
Chef chef = new Chef(this);
chefs.add(chef);
exec.execute(chef);
}
}
@Override
public void run() {
try{
while (!Thread.interrupted()){
// A new customer arrives; assign a WaitPerson:
WaitPerson wp = waitPersons.get(rand.nextInt(waitPersons.size()));
Customer c = new Customer(wp);
exec.execute(c);
TimeUnit.MILLISECONDS.sleep(100);
}
}catch (InterruptedException e){
System.out.println("Restaurant interrupted");
}
System.out.println("Restaurant closing");
}
}
public class RestaurantWithQueues {
public static void main(String[] args) throws IOException, InterruptedException {
ExecutorService exec = Executors.newCachedThreadPool();
Restaurant restaurant = new Restaurant(exec,5,2);
exec.execute(restaurant);
if (args.length > 0)
TimeUnit.SECONDS.sleep(new Integer(args[0]));
else {
System.out.println("Press 'Enter' to quit");
System.in.read();
}
exec.shutdownNow();
}
}
/*
Press 'Enter' to quit
WaitPerson 0 received SPRING_ROLLS delivering to Customer 1
Customer 1 eating SPRING_ROLLS
WaitPerson 3 received SPRING_ROLLS delivering to Customer 0
Customer 0 eating SPRING_ROLLS
WaitPerson 0 received BURRITO delivering to Customer 1
Customer 1 eating BURRITO
WaitPerson 3 received SPRING_ROLLS delivering to Customer 2
Customer 2 eating SPRING_ROLLS
WaitPerson 3 received BURRITO delivering to Customer 0
Customer 0 eating BURRITO
WaitPerson 1 received SPRING_ROLLS delivering to Customer 3
Customer 3 eating SPRING_ROLLS
WaitPerson 0 received FRUIT delivering to Customer 1
Customer 1 eating FRUIT
WaitPerson 4 received SALAD delivering to Customer 5
*/
8.3 分发工作
package concurrency;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.*;
class Car {
private final int id;
private boolean
engine = false, driveTrain = false, wheels = false;
public Car(int idn) {
id = idn;
}
//Empty Car object
public Car() {
id = -1;
}
public synchronized int getId() {
return id;
}
public synchronized void addEngine() {
engine = true;
} // 安装发动机
public synchronized void addDriveTrain() { // 安装车型
driveTrain = true;
}
public synchronized void addWheels() {
wheels = true;
} // 安装轮子
public synchronized String toString() {
return "Car " + id + " [" + " engine: " + engine
+ " driverTrain: " + driveTrain
+ " wheels: " + wheels + " ]";
}
}
class CarQueue extends LinkedBlockingQueue<Car> {
}
class ChassisBuilder implements Runnable { // chassis 底盘
private CarQueue carQueue;
private int counter = 0;
public ChassisBuilder(CarQueue cq) {
carQueue = cq;
}
@Override
public void run() {
try {
while (!Thread.interrupted()) {
TimeUnit.MILLISECONDS.sleep(500);
// Make chassis:
Car c = new Car(counter++);
System.out.println("ChassisBuilder created " + c);
// Insert into queue
carQueue.put(c);
}
} catch (InterruptedException e) {
System.out.println("Interrunpted: ");
}
System.out.println("ChassisBuilder off");
}
}
class Assembler implements Runnable {
private CarQueue chassisQueue, finishingQueue;
private Car car;
private CyclicBarrier barrier = new CyclicBarrier(4);
private RobotPool robotPool;
public Assembler(CarQueue cq, CarQueue fq, RobotPool rp) {
chassisQueue = cq;
finishingQueue = fq;
robotPool = rp;
}
public Car car() {
return car;
}
public CyclicBarrier barrier() {
return barrier;
}
@Override
public void run() {
try {
while (!Thread.interrupted()) {
//Blocks until chassis is avialable:
car = chassisQueue.take();
// Hire robots to perform work:
robotPool.hire(EngineRobot.class, this);
robotPool.hire(DriveTrainRobot.class, this);
robotPool.hire(WheelRobot.class, this);
barrier.await();// Unitl the robots finish
// Put car into finishingQueue for further work
finishingQueue.put(car);
}
} catch (InterruptedException e) {
System.out.println("Exiting Assembler via interrupt");
} catch (BrokenBarrierException e) {
// This one we want to know about
throw new RuntimeException(e);
}
System.out.println("Assembler off");
}
}
class Reporter implements Runnable {
private CarQueue carQueue;
public Reporter(CarQueue cq) {
carQueue = cq;
}
public void run() {
try {
while (!Thread.interrupted()) {
System.out.println(carQueue.take());
}
} catch (InterruptedException e) {
System.out.println("Exiting Reporter via interrupt");
}
System.out.println("Reporter off");
}
}
abstract class Robot implements Runnable {
private RobotPool pool;
public Robot(RobotPool p) {
pool = p;
}
protected Assembler assembler;
public Robot assignAssembler(Assembler assembler) {
this.assembler = assembler;
return this;
}
private boolean engage = false; // 参与
public synchronized void engage() {
engage = true;
notifyAll();
}
// The part of run() that’s different for each robot:
abstract protected void performService();
public void run() {
try {
powerDown(); // Wait until needed
while (!Thread.interrupted()) {
performService();
assembler.barrier().await(); //Synchronize
// We’re done with that job...
powerDown();
}
} catch (InterruptedException e) {
System.out.println("Exiting " + this + " via interrupt");
} catch (BrokenBarrierException e) {
throw new RuntimeException(e);
}
System.out.println(this + " off");
}
private synchronized void powerDown() throws InterruptedException {
engage = false;
assembler = null; // Disconnect from the Assembler
// Put ourselves back in the available pool:
pool.release(this);
while (engage == false) // power down
wait();
}
public String toString() {
return getClass().getName();
}
}
class EngineRobot extends Robot {
public EngineRobot(RobotPool pool) {
super(pool);
}
protected void performService() {
System.out.println(this + " installing engine");
assembler.car().addEngine();
}
}
class DriveTrainRobot extends Robot {
public DriveTrainRobot(RobotPool pool) {
super(pool);
}
protected void performService() {
System.out.println(this + " installing DriveTrain");
assembler.car().addDriveTrain();
}
}
class WheelRobot extends Robot {
public WheelRobot(RobotPool pool) {
super(pool);
}
protected void performService() {
System.out.println(this + " installing Wheels");
assembler.car().addWheels();
}
}
class RobotPool {
// Quietly prevents identical entries:
private Set<Robot> pool = new HashSet<Robot>();
public synchronized void add(Robot r) {
pool.add(r);
notifyAll();
}
public synchronized void hire(Class<? extends Robot> robotType, Assembler d)
throws InterruptedException {
for (Robot r : pool)
if (r.getClass().equals(robotType)) {
pool.remove(r);
r.assignAssembler(d);
r.engage(); // Power it up to do the task
return;
}
wait(); // None available
hire(robotType, d); // Try again,recursively
}
public synchronized void release(Robot r) {
add(r);
}
}
public class CarBuilder {
public static void main(String[] args) throws InterruptedException {
CarQueue chassisQueue = new CarQueue(),
finishingQueue = new CarQueue();
ExecutorService exec = Executors.newCachedThreadPool();
RobotPool robotPool = new RobotPool();
exec.execute(new EngineRobot(robotPool));
exec.execute(new DriveTrainRobot(robotPool));
exec.execute(new WheelRobot(robotPool));
exec.execute(new Assembler(
chassisQueue, finishingQueue, robotPool));
exec.execute(new Reporter(finishingQueue));
// Start everything running by producing chassis:
exec.execute(new ChassisBuilder(chassisQueue));
TimeUnit.SECONDS.sleep(7);
exec.shutdownNow();
}
}
/*
/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/bin/java "-javaagent:/Applications/IntelliJ IDEA.app/Contents/lib/idea_rt.jar=51284:/Applications/IntelliJ IDEA.app/Contents/bin" -Dfile.encoding=UTF-8 -classpath /Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/charsets.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/deploy.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/ext/cldrdata.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/ext/dnsns.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/ext/jaccess.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/ext/jfxrt.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/ext/localedata.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/ext/nashorn.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/ext/sunec.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/ext/sunjce_provider.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/ext/sunpkcs11.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/ext/zipfs.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/javaws.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/jce.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/jfr.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/jfxswt.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/jsse.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/management-agent.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/plugin.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/resources.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/rt.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/lib/ant-javafx.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/lib/dt.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/lib/javafx-mx.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/lib/jconsole.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/lib/packager.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/lib/sa-jdi.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/lib/tools.jar:/Users/erin/JavaProject/thinking_in_java_example/target/classes:/Users/erin/JavaProject/thinking_in_java_example/net.mindview.util.jar:/Users/erin/Documents/TestDoc/pets.jar:/Users/erin/.m2/repository/org/testng/testng/6.10/testng-6.10.jar:/Users/erin/.m2/repository/com/beust/jcommander/1.48/jcommander-1.48.jar:/Users/erin/.m2/repository/xom/xom/1.3.2/xom-1.3.2.jar:/Users/erin/.m2/repository/xml-apis/xml-apis/1.3.03/xml-apis-1.3.03.jar:/Users/erin/.m2/repository/xerces/xercesImpl/2.8.0/xercesImpl-2.8.0.jar:/Users/erin/.m2/repository/xalan/xalan/2.7.0/xalan-2.7.0.jar concurrency.CarBuilder
ChassisBuilder created Car 0 [ engine: false driverTrain: false wheels: false ]
concurrency.EngineRobot installing engine
concurrency.DriveTrainRobot installing DriveTrain
concurrency.WheelRobot installing Wheels
Car 0 [ engine: true driverTrain: true wheels: true ]
ChassisBuilder created Car 1 [ engine: false driverTrain: false wheels: false ]
concurrency.EngineRobot installing engine
concurrency.WheelRobot installing Wheels
concurrency.DriveTrainRobot installing DriveTrain
Car 1 [ engine: true driverTrain: true wheels: true ]
ChassisBuilder created Car 2 [ engine: false driverTrain: false wheels: false ]
concurrency.EngineRobot installing engine
concurrency.DriveTrainRobot installing DriveTrain
concurrency.WheelRobot installing Wheels
Car 2 [ engine: true driverTrain: true wheels: true ]
ChassisBuilder created Car 3 [ engine: false driverTrain: false wheels: false ]
concurrency.EngineRobot installing engine
concurrency.DriveTrainRobot installing DriveTrain
concurrency.WheelRobot installing Wheels
Car 3 [ engine: true driverTrain: true wheels: true ]
ChassisBuilder created Car 4 [ engine: false driverTrain: false wheels: false ]
concurrency.EngineRobot installing engine
concurrency.WheelRobot installing Wheels
concurrency.DriveTrainRobot installing DriveTrain
Car 4 [ engine: true driverTrain: true wheels: true ]
ChassisBuilder created Car 5 [ engine: false driverTrain: false wheels: false ]
concurrency.DriveTrainRobot installing DriveTrain
concurrency.EngineRobot installing engine
concurrency.WheelRobot installing Wheels
Car 5 [ engine: true driverTrain: true wheels: true ]
ChassisBuilder created Car 6 [ engine: false driverTrain: false wheels: false ]
concurrency.EngineRobot installing engine
concurrency.WheelRobot installing Wheels
concurrency.DriveTrainRobot installing DriveTrain
Car 6 [ engine: true driverTrain: true wheels: true ]
ChassisBuilder created Car 7 [ engine: false driverTrain: false wheels: false ]
concurrency.EngineRobot installing engine
concurrency.DriveTrainRobot installing DriveTrain
concurrency.WheelRobot installing Wheels
Car 7 [ engine: true driverTrain: true wheels: true ]
ChassisBuilder created Car 8 [ engine: false driverTrain: false wheels: false ]
concurrency.EngineRobot installing engine
concurrency.DriveTrainRobot installing DriveTrain
concurrency.WheelRobot installing Wheels
Car 8 [ engine: true driverTrain: true wheels: true ]
ChassisBuilder created Car 9 [ engine: false driverTrain: false wheels: false ]
concurrency.DriveTrainRobot installing DriveTrain
concurrency.WheelRobot installing Wheels
concurrency.EngineRobot installing engine
Car 9 [ engine: true driverTrain: true wheels: true ]
ChassisBuilder created Car 10 [ engine: false driverTrain: false wheels: false ]
concurrency.DriveTrainRobot installing DriveTrain
concurrency.WheelRobot installing Wheels
concurrency.EngineRobot installing engine
Car 10 [ engine: true driverTrain: true wheels: true ]
ChassisBuilder created Car 11 [ engine: false driverTrain: false wheels: false ]
concurrency.EngineRobot installing engine
concurrency.WheelRobot installing Wheels
concurrency.DriveTrainRobot installing DriveTrain
Car 11 [ engine: true driverTrain: true wheels: true ]
ChassisBuilder created Car 12 [ engine: false driverTrain: false wheels: false ]
concurrency.EngineRobot installing engine
concurrency.WheelRobot installing Wheels
concurrency.DriveTrainRobot installing DriveTrain
Car 12 [ engine: true driverTrain: true wheels: true ]
Exiting concurrency.DriveTrainRobot via interrupt
Exiting concurrency.EngineRobot via interrupt
concurrency.EngineRobot off
Exiting concurrency.WheelRobot via interrupt
concurrency.WheelRobot off
Interrunpted:
ChassisBuilder off
Exiting Reporter via interrupt
Reporter off
Exiting Assembler via interrupt
concurrency.DriveTrainRobot off
Assembler off
Process finished with exit code 0
*/
9 性能调优
9.1 比较各类互斥技术
示例是一种比较天真的比较方法
package concurrency;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
abstract class Incrementable{
protected long counter = 0;
public abstract void increment();
}
class SynconizingTest extends Incrementable{
@Override
public void increment() {
++counter;
}
}
class LockingTest extends Incrementable{
private Lock lock = new ReentrantLock();
@Override
public void increment() {
lock.lock();
try{
++counter;
}finally {
lock.unlock();
}
}
}
public class SimpleMicroBenchmark {
static long test(Incrementable incr){
long start = System.nanoTime();
for (long i = 0; i < 10000000L; i++)
incr.increment();
return System.nanoTime() - start;
}
public static void main(String[] args) {
long syncTime = test(new SynconizingTest());
long lockTime = test(new LockingTest());
System.out.printf("synchronized: %1$10d
",syncTime);
System.out.printf("Lock: %1$10d
",lockTime);
System.out.printf("Lock/synchronized = %1$.3f",(double)lockTime/(double)syncTime);
}
}
/*
synchronized: 6896974
Lock: 150561207
Lock/synchronized = 21.830
*/
- 互斥存在竞争情况下，才能看到真正的性能差异。因此必须有多个任务尝试访问互斥代码区。上面示例中，每个互斥都是单个main线程隔离的情况下测试。
- 编译器synchronized，会特殊优化，可能会注意到这个线程是单线程。甚至会识别counter被递增次数固定数量。因此预先计算结果。需要防止编译器去预测结果的可能性。
package concurrency;
import java.util.Random;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.util.concurrent.locks.*;
abstract class Accumulator{ //蓄电池
public static long cycles = 50000L;
// Number of Modifiers and Readers during each test:
private static final int N = 4;
public static ExecutorService exec = Executors.newFixedThreadPool(N*2);
private static CyclicBarrier barrier = new CyclicBarrier(N*2 + 1);
protected volatile int index = 0;
protected volatile long value = 0;
protected long duration = 0;
protected String id = "error";
protected final static int SIZE = 100000;
protected static int[] preLoaded = new int[SIZE];
static {
// Load the array of random numbers: 预加载，可以减小在主循环调用 rand.nextInt()所造成的影响
Random rand = new Random(47);
for (int i =0; i < SIZE; i++)
preLoaded[i] = rand.nextInt();
}
//模板方法设计模式，不同代码隔离实现
public abstract void accumulate(); //累加
public abstract long read();//读 避免优化器识别这些值从来不会被使用
private class Modifier implements Runnable{
public void run(){
for (long i = 0; i < cycles; i++)
accumulate();
try{
barrier.await();
}catch (Exception e){
throw new RuntimeException(e);
}
}
}
private class Reader implements Runnable{
private volatile long value;
public void run(){
for (long i = 0; i < cycles; i++)
value = read();
try{
barrier.await();
}catch (Exception e){
throw new RuntimeException(e);
}
}
}
public void timedTest(){
long start = System.nanoTime();
for (int i = 0; i < N; i++){
exec.execute(new Modifier());
exec.execute(new Reader());
}
try{
barrier.await(); // N*2 + 1
}catch (Exception e){
throw new RuntimeException(e);
}
duration = System.nanoTime() - start;
System.out.printf("%-13s: %13d
",id,duration);
}
public static void report(Accumulator acc1,Accumulator acc2){
System.out.printf("%-22s: %.2f
",acc1.id + "/" + acc2.id,(double)acc1.duration/(double)acc2.duration);
}
}
class BaseLine extends Accumulator{
{id = "BaseLine";}
public void accumulate(){
value += preLoaded[(index++)% SIZE]; // 改进书上版
}
public long read(){return value;}
}
class SynchronizedTest extends Accumulator{
{id = "synchronized";}
public synchronized void accumulate(){
value += preLoaded[index++];
if (index >= SIZE) index = 0;
}
public synchronized long read(){
return value;
}
}
class LockTest extends Accumulator{
{id = "Lock";}
private Lock lock = new ReentrantLock();
public void accumulate(){
lock.lock();
try{
value += preLoaded[index++];
if (index >= SIZE) index = 0;
}finally {
lock.unlock();
}
}
@Override
public long read() {
lock.lock();
try{
return value;
}finally {
lock.unlock();
}
}
}
class AtomicTest extends Accumulator{
{ id = "Atomic";}
private AtomicInteger index = new AtomicInteger(0);
private AtomicLong value = new AtomicLong(0);
public void accumulate(){
// Oops! Relying on more than one Atomic at
// a time doesn’t work. But it still gives us
// a performance indicator:
int i = index.getAndIncrement();
if (index.get() < SIZE){ // 改进书上版
value.getAndAdd(preLoaded[i]);
if (++i >= SIZE)
index.set(0);
}
}
@Override
public long read() {
return value.get();
}
}
public class SynchronizationComparisons {
static BaseLine baseLine = new BaseLine();
static SynchronizedTest synch = new SynchronizedTest();
static LockTest lock = new LockTest();
static AtomicTest atomic = new AtomicTest();
static void test(){
System.out.println("============================");
System.out.printf("%-12s : %13d
","Cycles",Accumulator.cycles);
// baseLine.timedTest();
synch.timedTest();
lock.timedTest();
atomic.timedTest();
Accumulator.report(synch,baseLine);
Accumulator.report(lock,baseLine);
Accumulator.report(atomic,baseLine);
Accumulator.report(synch,lock);
Accumulator.report(synch,atomic);
Accumulator.report(lock,atomic);
}
public static void main(String[] args) {
int iterations = 5; //Default
if (args.length > 0) // Optionally change iterations
iterations = new Integer(args[0]);
// The first time fills the thread pool:
System.out.println("Warmup");
baseLine.timedTest();//本项测试包含了初始线程读创建，测试结果丢弃
// Now the initial test doesn’t include the cost
// of starting the threads for the first time.
// Produce multiple data points:
for (int i = 0; i < iterations; i++){
test();
Accumulator.cycles *= 2;
}
Accumulator.exec.shutdown();
}
}
/*
Warmup
BaseLine : 11340184
============================
Cycles : 50000
synchronized : 35240080
Lock : 25491654
Atomic : 21505532
synchronized/BaseLine : 3.11
Lock/BaseLine : 2.25
Atomic/BaseLine : 1.90
synchronized/Lock : 1.38
synchronized/Atomic : 1.64
Lock/Atomic : 1.19
============================
Cycles : 100000
synchronized : 51603796
Lock : 34416901
Atomic : 10885385
synchronized/BaseLine : 4.55
Lock/BaseLine : 3.03
Atomic/BaseLine : 0.96
synchronized/Lock : 1.50
synchronized/Atomic : 4.74
Lock/Atomic : 3.16
============================
Cycles : 200000
synchronized : 101364921
Lock : 66832700
Atomic : 20464159
synchronized/BaseLine : 8.94
Lock/BaseLine : 5.89
Atomic/BaseLine : 1.80
synchronized/Lock : 1.52
synchronized/Atomic : 4.95
Lock/Atomic : 3.27
============================
Cycles : 400000
synchronized : 191781967
Lock : 133360086
Atomic : 39387523
synchronized/BaseLine : 16.91
Lock/BaseLine : 11.76
Atomic/BaseLine : 3.47
synchronized/Lock : 1.44
synchronized/Atomic : 4.87
Lock/Atomic : 3.39
============================
Cycles : 800000
synchronized : 335372500
Lock : 232259964
Atomic : 83264806
synchronized/BaseLine : 29.57
Lock/BaseLine : 20.48
Atomic/BaseLine : 7.34
synchronized/Lock : 1.44
synchronized/Atomic : 4.03
Lock/Atomic : 2.79
*/
- 使用Lock通常比synchronized高效
- synchronized 可读性高，以该关键字入手，只有在性能调优时才替换为Lock
- Atomic在非常简单的情况下才有用。以更加传统的互斥方式入手，只有性能方面有明确指示时，替换为Atomic
9.2 免锁容器
写时复制器
CopyOnWriteArrayList，Java并发包中提供的一个并发容器，线程安全，读操作无锁，写操作则通过创建底层数组的新副本来实现，是一种读写分离的并发策略
CopyOnWriteArraySet。
优点：
读操作性能很高，因为无需任何同步措施，比较适用于读多写少的并发场景。
Java的list在遍历时，若中途有别的线程对list容器进行修改，则会抛出ConcurrentModificationException异常。
而CopyOnWriteArrayList由于其"读写分离"的思想，遍历和修改操作分别作用在不同的list容器，所以在使用迭代器进行遍历时候，也就不会抛出ConcurrentModificationException异常了
缺点：
一是内存占用问题，毕竟每次执行写操作都要将原容器拷贝一份，数据量大时，对内存压力较大，可能会引起频繁GC；
二是无法保证实时性，Vector对于读写操作均加锁同步，可以保证读和写的强一致性。而CopyOnWriteArrayList由于其实现策略的原因，写和读分别作用在新老不同容器上，在写操作执行过程中，读不会阻塞但读取到的却是老容器的数据。
CopyOnWriteArraySet 将使用CopyOnWriteArrayList来实现其免锁行为
ConcurrentHashMap和 ConcurrentLinkedQueue 使用了类似的技术,允许并发的读取和写入.但是容器中只有部分内容而不是整个容器可以被复制和修改.
ConcurrentHashMap不会抛出 ConcurrentModificationException.
乐观锁
package concurrency;
import net.mindview.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public abstract class Tester<C> {
static int testReps = 10;
static int testCycles = 1000;
static int containerSize = 1000;
abstract C containerInitializer(); // 泛型参数C 表示容器类型
abstract void startReadersAndWriters();// 启动读取者 和 写入者任务，读取和修改待测试容器
C testContainer;
String testId;
int nReaders;
int nWriters;
volatile long readResult = 0;
volatile long readTime = 0;
volatile long writeTime = 0;
CountDownLatch endLatch;
static ExecutorService exec = Executors.newCachedThreadPool();
Integer[] writeData;
Tester(String testId, int nReaders, int nWriters) {
this.testId = testId + " " + nReaders + "r " + nWriters + " w";
this.nReaders = nReaders;
this.nWriters = nWriters;
writeData = Generated.array(Integer.class, new RandomGenerator.Integer(), containerSize);
for (int i = 0; i < testReps; i++) {
runTest();
readTime = 0;
writeTime = 0;
}
}
void runTest() {
endLatch = new CountDownLatch(nReaders + nWriters);
testContainer = containerInitializer(); // 初始化容器
startReadersAndWriters(); // 读取和修改待测试容器
try {
endLatch.await();
} catch (InterruptedException e) {
System.out.println("endLatch interrupted");
}
System.out.printf("%-27s %14d %14d
", testId, readTime, writeTime);
if (readTime != 0 && writeTime != 0)
System.out.printf("%-27s %14d
", "readTime + writeTime =", readTime + writeTime);
}
abstract class TestTask implements Runnable{
abstract void test();
abstract void putResults();
long duration;
@Override
public void run(){
long startTime = System.nanoTime();
test();
duration = System.nanoTime() - startTime;
synchronized (Tester.this){
putResults();
}
endLatch.countDown();
}
}
public static void initMain(String[] args){
if (args.length > 0)
testReps = new Integer(args[0]);
if (args.length > 1)
testCycles = new Integer(args[1]);
if (args.length > 2)
containerSize = new Integer(args[2]);
System.out.printf("%-27s %14s %14s
","Type","Read time","Write time");
}
}
package concurrency;
import net.mindview.util.CountingIntegerList;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
abstract class ListTest extends Tester<List<Integer>> {
ListTest(String testId, int nReaders, int nWriters) {
super(testId, nReaders, nWriters);
}
class Reader extends TestTask {
long result = 0;
@Override
void test() {
for (long i = 0; i < testCycles; i++)
for (int index = 0; index < containerSize; index++)
result += testContainer.get(index);
}
@Override
void putResults() {
readResult += result;
readTime += duration;
}
}
class Writer extends TestTask {
@Override
void test() {
for (long i = 0; i < testCycles; i++)
for (int index = 0; index < containerSize; index++)
testContainer.set(index, writeData[index]);
}
@Override
void putResults() {
writeTime += duration;
}
}
void startReadersAndWriters() {
for (int i = 0; i < nReaders; i++)
exec.execute(new Reader());
for (int i = 0; i < nWriters; i++)
exec.execute(new Writer());
}
}
class SynchronizedArrayListTest extends ListTest{
SynchronizedArrayListTest(int nReaders, int nWriters) {
super("Synched ArrayList", nReaders, nWriters);
}
@Override
List<Integer> containerInitializer() {
return Collections.synchronizedList(new ArrayList<Integer>(new CountingIntegerList(containerSize)));
}
}
class CopyOnWriteArrayListTest extends ListTest {
@Override
List<Integer> containerInitializer() {
return new CopyOnWriteArrayList<>(new CountingIntegerList(containerSize));
}
CopyOnWriteArrayListTest(int nReaders, int nWriters) {
super("CopyOnWriteArrayList", nReaders, nWriters);
}
}
public class ListComparisions {
public static void main(String[] args) {
Tester.initMain(args);
new SynchronizedArrayListTest(10,0);
new SynchronizedArrayListTest(9,1);
new SynchronizedArrayListTest(5,5);
new CopyOnWriteArrayListTest(10,0);
new CopyOnWriteArrayListTest(9,1);
new CopyOnWriteArrayListTest(5,5);
}
}
/*
原始输出：
Type Read time Write time
Synched ArrayList 10r 0 w 5610305101 0
Synched ArrayList 10r 0 w 3025621488 0
Synched ArrayList 10r 0 w 3176419308 0
Synched ArrayList 10r 0 w 4872977562 0
Synched ArrayList 10r 0 w 6391438021 0
Synched ArrayList 10r 0 w 6938175002 0
Synched ArrayList 10r 0 w 5620420430 0
Synched ArrayList 10r 0 w 5455676762 0
Synched ArrayList 10r 0 w 5276662598 0
Synched ArrayList 10r 0 w 5334414698 0
Synched ArrayList 9r 1 w 4786549413 568587941
readTime + writeTime = 5355137354
Synched ArrayList 9r 1 w 5738364857 659456412
readTime + writeTime = 6397821269
Synched ArrayList 9r 1 w 5021856384 607123974
readTime + writeTime = 5628980358
Synched ArrayList 9r 1 w 5040457594 588756865
readTime + writeTime = 5629214459
Synched ArrayList 9r 1 w 5071138160 597874813
readTime + writeTime = 5669012973
Synched ArrayList 9r 1 w 4789427513 582201979
readTime + writeTime = 5371629492
Synched ArrayList 9r 1 w 4866769446 556473758
readTime + writeTime = 5423243204
Synched ArrayList 9r 1 w 4483384085 533055109
readTime + writeTime = 5016439194
Synched ArrayList 9r 1 w 4701111980 565864774
readTime + writeTime = 5266976754
Synched ArrayList 9r 1 w 5057865366 576460699
readTime + writeTime = 5634326065
Synched ArrayList 5r 5 w 3776807242 4607324025
readTime + writeTime = 8384131267
Synched ArrayList 5r 5 w 2365641681 4032676777
readTime + writeTime = 6398318458
Synched ArrayList 5r 5 w 3857738636 4816249444
readTime + writeTime = 8673988080
Synched ArrayList 5r 5 w 3115984572 4292139465
readTime + writeTime = 7408124037
Synched ArrayList 5r 5 w 2647305522 4853323428
readTime + writeTime = 7500628950
Synched ArrayList 5r 5 w 2540811416 4204565679
readTime + writeTime = 6745377095
Synched ArrayList 5r 5 w 2577392155 4403273331
readTime + writeTime = 6980665486
Synched ArrayList 5r 5 w 2441728099 4058141783
readTime + writeTime = 6499869882
Synched ArrayList 5r 5 w 2368237967 4071609214
readTime + writeTime = 6439847181
Synched ArrayList 5r 5 w 3478167053 4623390954
readTime + writeTime = 8101558007
CopyOnWriteArrayList 10r 0 w 251525964 0
CopyOnWriteArrayList 10r 0 w 81496896 0
CopyOnWriteArrayList 10r 0 w 75469815 0
CopyOnWriteArrayList 10r 0 w 108743833 0
CopyOnWriteArrayList 10r 0 w 50087465 0
CopyOnWriteArrayList 10r 0 w 64974385 0
CopyOnWriteArrayList 10r 0 w 78545732 0
CopyOnWriteArrayList 10r 0 w 65652965 0
CopyOnWriteArrayList 10r 0 w 50031217 0
CopyOnWriteArrayList 10r 0 w 65861050 0
CopyOnWriteArrayList 9r 1 w 64623445 48965668
readTime + writeTime = 113589113
CopyOnWriteArrayList 9r 1 w 71029595 42148126
readTime + writeTime = 113177721
CopyOnWriteArrayList 9r 1 w 100936991 51194763
readTime + writeTime = 152131754
CopyOnWriteArrayList 9r 1 w 70990246 40111340
readTime + writeTime = 111101586
CopyOnWriteArrayList 9r 1 w 112966916 41557337
readTime + writeTime = 154524253
CopyOnWriteArrayList 9r 1 w 96792174 43101184
readTime + writeTime = 139893358
CopyOnWriteArrayList 9r 1 w 60299261 39679658
readTime + writeTime = 99978919
CopyOnWriteArrayList 9r 1 w 69810196 40320174
readTime + writeTime = 110130370
CopyOnWriteArrayList 9r 1 w 74679802 43674987
readTime + writeTime = 118354789
CopyOnWriteArrayList 9r 1 w 68364435 39034319
readTime + writeTime = 107398754
CopyOnWriteArrayList 5r 5 w 32333150 1449274826
readTime + writeTime = 1481607976
CopyOnWriteArrayList 5r 5 w 32341743 1393562637
readTime + writeTime = 1425904380
CopyOnWriteArrayList 5r 5 w 29014123 1421587468
readTime + writeTime = 1450601591
CopyOnWriteArrayList 5r 5 w 33004215 1438439818
readTime + writeTime = 1471444033
CopyOnWriteArrayList 5r 5 w 40328790 1330090086
readTime + writeTime = 1370418876
CopyOnWriteArrayList 5r 5 w 41409292 1366642997
readTime + writeTime = 1408052289
CopyOnWriteArrayList 5r 5 w 35904266 1326295148
readTime + writeTime = 1362199414
CopyOnWriteArrayList 5r 5 w 39832754 1372299363
readTime + writeTime = 1412132117
CopyOnWriteArrayList 5r 5 w 38153621 1389125404
readTime + writeTime = 1427279025
CopyOnWriteArrayList 5r 5 w 40431146 1460588763
readTime + writeTime = 1501019909
整理后
Type Read time Write time
Synched ArrayList 10r 0 w 5334414698 0
CopyOnWriteArrayList 10r 0 w 65861050 0
CopyOnWriteArrayList在没有写入时，速度会快很多。
Synched ArrayList 9r 1 w 5057865366 576460699
readTime + writeTime = 5634326065
CopyOnWriteArrayList 9r 1 w 68364435 39034319
readTime + writeTime = 107398754
Synched ArrayList 5r 5 w 3478167053 4623390954
readTime + writeTime = 8101558007
CopyOnWriteArrayList 5r 5 w 40431146 1460588763
readTime + writeTime = 1501019909
CopyOnWriteArrayList在5个写入时，速度也是明显快。
*/
package concurrency;
import arrays.CountingGenerator;
import net.mindview.util.MapData;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
abstract class MapTest
extends Tester<Map<Integer, Integer>> {
MapTest(String testId, int nReaders, int nWriters) {
super(testId, nReaders, nWriters);
}
class Reader extends TestTask {
long result = 0;
@Override
void test() {
for (long i = 0; i < testCycles; i++)
for (int index = 0; index < containerSize; index++)
result += testContainer.get(index);
}
@Override
void putResults() {
readResult += result;
readTime += duration;
}
}
class Writer extends TestTask {
@Override
void test() {
for (long i = 0; i < testCycles; i++)
for (int index = 0; index < containerSize; index++)
testContainer.put(index, writeData[index]);
}
@Override
void putResults() {
writeTime += duration;
}
}
void startReadersAndWriters() {
for (int i = 0; i < nReaders; i++)
exec.execute(new Reader());
for (int i = 0; i < nWriters; i++)
exec.execute(new Writer());
}
}
class SynchronizedMapTest extends MapTest {
SynchronizedMapTest(int nReaders, int nWriters) {
super("Synched HashMap", nReaders, nWriters);
}
@Override
Map<Integer, Integer> containerInitializer() {
return Collections.synchronizedMap(new HashMap<>(MapData.map(new CountingGenerator.Integer(),
new CountingGenerator.Integer(),
containerSize)));
}
}
class ConcurrentHashMapTest extends MapTest{
ConcurrentHashMapTest(int nReaders, int nWriters) {
super("ConcurrentHashMap", nReaders, nWriters);
}
@Override
Map<Integer, Integer> containerInitializer() {
return new ConcurrentHashMap<>(MapData.map(new CountingGenerator.Integer(),
new CountingGenerator.Integer(),
containerSize));
}
}
public class MapComparisons {
public static void main(String[] args) {
Tester.initMain(args);
new SynchronizedMapTest(10,0);
new SynchronizedMapTest(9,1);
new SynchronizedMapTest(5,5);
new ConcurrentHashMapTest(10,0);
new ConcurrentHashMapTest(9,1);
new ConcurrentHashMapTest(5,5);
Tester.exec.shutdown();
}
}
/*
整理前：
Type Read time Write time
Synched HashMap 10r 0 w 6089992243 0
Synched HashMap 10r 0 w 5258052865 0
Synched HashMap 10r 0 w 5667909435 0
Synched HashMap 10r 0 w 7375416757 0
Synched HashMap 10r 0 w 6308420080 0
Synched HashMap 10r 0 w 6029639784 0
Synched HashMap 10r 0 w 6183771120 0
Synched HashMap 10r 0 w 5977273571 0
Synched HashMap 10r 0 w 11496262870 0
Synched HashMap 10r 0 w 5729938732 0
Synched HashMap 9r 1 w 5388966270 637589844
readTime + writeTime = 6026556114
Synched HashMap 9r 1 w 5469537992 627355786
readTime + writeTime = 6096893778
Synched HashMap 9r 1 w 6030070463 676397091
readTime + writeTime = 6706467554
Synched HashMap 9r 1 w 7412860107 763753806
readTime + writeTime = 8176613913
Synched HashMap 9r 1 w 5998990304 684768346
readTime + writeTime = 6683758650
Synched HashMap 9r 1 w 5329940736 488324658
readTime + writeTime = 5818265394
Synched HashMap 9r 1 w 6189617505 629958439
readTime + writeTime = 6819575944
Synched HashMap 9r 1 w 6822824068 780700371
readTime + writeTime = 7603524439
Synched HashMap 9r 1 w 5631757707 640090950
readTime + writeTime = 6271848657
Synched HashMap 9r 1 w 5961147320 516452947
readTime + writeTime = 6477600267
Synched HashMap 5r 5 w 3776665900 3390468187
readTime + writeTime = 7167134087
Synched HashMap 5r 5 w 3524566830 3646794168
readTime + writeTime = 7171360998
Synched HashMap 5r 5 w 3256917115 3407497549
readTime + writeTime = 6664414664
Synched HashMap 5r 5 w 3430014452 3301065680
readTime + writeTime = 6731080132
Synched HashMap 5r 5 w 3736640630 3892165601
readTime + writeTime = 7628806231
Synched HashMap 5r 5 w 3470630888 2893063909
readTime + writeTime = 6363694797
Synched HashMap 5r 5 w 4087087795 3922603600
readTime + writeTime = 8009691395
Synched HashMap 5r 5 w 4001178159 3930809421
readTime + writeTime = 7931987580
Synched HashMap 5r 5 w 4247478944 4401801386
readTime + writeTime = 8649280330
Synched HashMap 5r 5 w 4189483378 4189342379
readTime + writeTime = 8378825757
ConcurrentHashMap 10r 0 w 799532101 0
ConcurrentHashMap 10r 0 w 498883520 0
ConcurrentHashMap 10r 0 w 301427945 0
ConcurrentHashMap 10r 0 w 260554416 0
ConcurrentHashMap 10r 0 w 306817963 0
ConcurrentHashMap 10r 0 w 221274524 0
ConcurrentHashMap 10r 0 w 378544222 0
ConcurrentHashMap 10r 0 w 360332515 0
ConcurrentHashMap 10r 0 w 230206682 0
ConcurrentHashMap 10r 0 w 225503246 0
ConcurrentHashMap 9r 1 w 158711100 31132911
readTime + writeTime = 189844011
ConcurrentHashMap 9r 1 w 185371549 36341818
readTime + writeTime = 221713367
ConcurrentHashMap 9r 1 w 314637769 46187551
readTime + writeTime = 360825320
ConcurrentHashMap 9r 1 w 156439601 30364556
readTime + writeTime = 186804157
ConcurrentHashMap 9r 1 w 130465436 25087416
readTime + writeTime = 155552852
ConcurrentHashMap 9r 1 w 399573804 54843573
readTime + writeTime = 454417377
ConcurrentHashMap 9r 1 w 158873072 29602969
readTime + writeTime = 188476041
ConcurrentHashMap 9r 1 w 340250024 49177060
readTime + writeTime = 389427084
ConcurrentHashMap 9r 1 w 263855988 54537027
readTime + writeTime = 318393015
ConcurrentHashMap 9r 1 w 128703639 20501072
readTime + writeTime = 149204711
ConcurrentHashMap 5r 5 w 71652639 546217150
readTime + writeTime = 617869789
ConcurrentHashMap 5r 5 w 72004782 551243649
readTime + writeTime = 623248431
ConcurrentHashMap 5r 5 w 87565410 633265168
readTime + writeTime = 720830578
ConcurrentHashMap 5r 5 w 170281167 612403364
readTime + writeTime = 782684531
ConcurrentHashMap 5r 5 w 74977286 603616564
readTime + writeTime = 678593850
ConcurrentHashMap 5r 5 w 76098239 533697850
readTime + writeTime = 609796089
ConcurrentHashMap 5r 5 w 71040505 626198413
readTime + writeTime = 697238918
ConcurrentHashMap 5r 5 w 67103365 555462162
readTime + writeTime = 622565527
ConcurrentHashMap 5r 5 w 68443147 646141040
readTime + writeTime = 714584187
ConcurrentHashMap 5r 5 w 75745042 561922272
readTime + writeTime = 637667314
整理后
Type Read time Write time
Synched HashMap 10r 0 w 5729938732 0
ConcurrentHashMap 10r 0 w 225503246 0
Synched HashMap 9r 1 w 5961147320 516452947
readTime + writeTime = 6477600267
ConcurrentHashMap 9r 1 w 128703639 20501072
readTime + writeTime = 149204711
Synched HashMap 5r 5 w 4189483378 4189342379
readTime + writeTime = 8378825757
ConcurrentHashMap 5r 5 w 75745042 561922272
readTime + writeTime = 637667314
*/
ConcurrentHashMap synchronizedMap 25倍以上
ConcurrentHashMap 使用了一种不同的技术，明显最小化写入所造成的影响。
9.3 乐观加锁
package concurrency;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class FastSimulation {
static final int N_ELEMENTS = 100000;
static final int N_GENES = 30;
static final int N_EVOLVERS = 50;
static final AtomicInteger[][] GRID = new AtomicInteger[N_ELEMENTS][N_GENES];
static Random rand = new Random(47);
static class Evolver implements Runnable{ //进化
@Override
public void run() {
while (!Thread.interrupted()){
// Randomly select an element to work on:
int element = rand.nextInt(N_ELEMENTS);
for (int i = 0; i < N_GENES; i++){
int previous = element -1;
if (previous < 0) previous = N_ELEMENTS - 1;
int next = element + 1;
if (next >= N_ELEMENTS) next = 0;
int oldvalue = GRID[element][i].get();
// Perform some kind of modeling calculation:
int newvalue = oldvalue + GRID[previous][i].get() + GRID[next][i].get();
newvalue /= 3; // Average the three values
if (!GRID[element][i].compareAndSet(oldvalue,newvalue)){
// Policy here to deal with failure. Here, we
// just report it and ignore it; our model
// will eventually deal with it.
System.out.println("Old value change from " + oldvalue);
}
}
}
}
}
public static void main(String[] args) throws InterruptedException {
ExecutorService exec = Executors.newCachedThreadPool();
for (int i = 0; i < N_ELEMENTS;i++)
for (int j = 0; j < N_GENES; j++)
GRID[i][j] = new AtomicInteger(rand.nextInt(1000));
for (int i = 0; i < N_EVOLVERS; i++)
exec.execute(new Evolver());
TimeUnit.SECONDS.sleep(5);
exec.shutdownNow();
}
}
9.4 ReadWriteLock
读频繁，写不频繁的优化，使用ReadWriteLock
当写锁已被其他任务持有，则读取者不能访问，直至写锁释放
package concurrency;
import org.omg.CORBA.PRIVATE_MEMBER;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
public class ReaderWriterList <T>{
private ArrayList<T> lockedList;
// Make the ordering fair:
private ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);
public ReaderWriterList(int size,T initialValue){
lockedList = new ArrayList<>(Collections.nCopies(size,initialValue));
}
public T set(int index,T element){
Lock wlock = lock.writeLock();
wlock.lock();
try{
return lockedList.set(index,element);
}finally {
wlock.unlock();
}
}
public T get(int index){
Lock rlock = lock.readLock();
rlock.lock();
try{
// Show that multiple readers
// may acquire the read lock:
if (lock.getReadHoldCount() > 1)
System.out.println(lock.getReadLockCount()); // 不执行该句
return lockedList.get(index);
}finally {
rlock.unlock();
}
}
public static void main(String[] args) {
new ReaderWriterListTest(30,1);
}
}
class ReaderWriterListTest{
ExecutorService exec = Executors.newCachedThreadPool();
private final static int SIZE = 100;
private static Random rand = new Random(47);
private ReaderWriterList<Integer> list = new ReaderWriterList<>(SIZE,0);
private class Writer implements Runnable{
@Override
public void run() {
try{
for (int i = 0; i < 20; i++){ // 2秒 测试
list.set(i,rand.nextInt());
TimeUnit.MILLISECONDS.sleep(100);
}
}catch (InterruptedException e){
// Acceptable way to exit
}
System.out.println("Writer finished,shutting dowm");
exec.shutdownNow();
}
}
private class Reader implements Runnable{
@Override
public void run() {
try{
while (!Thread.interrupted()){
for (int i = 0; i < SIZE; i ++){
list.get(i);
TimeUnit.MILLISECONDS.sleep(1);
}
}
}catch (InterruptedException e){
// Acceptable way to exit
}
}
}
public ReaderWriterListTest(int readers,int writers){
for (int i = 0; i < readers; i++)
exec.execute(new Reader());
for (int i = 0; i < writers; i++)
exec.execute(new Writer());
}
}
//Writer finished,shutting dowm
10 活动对象
package concurrency;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;
public class ActiveObjectDemo {
private ExecutorService ex = Executors.newSingleThreadExecutor();
private Random rand = new Random(47);
// Insert a random delay to produce the effect
// of a calculation time:
private void pause(int factor) {
try {
TimeUnit.MILLISECONDS.sleep(100 + rand.nextInt(factor));
} catch (InterruptedException e) {
System.out.println("Sleep() interrupted");
}
}
public Future<Integer> calculateInt(final int x, final int y) {
return ex.submit(new Callable<Integer>() {
@Override
public Integer call() throws Exception {
System.out.println("starting " + x + " + " + y);
pause(500);
return x + y;
}
});
}
public Future<Float> calculateFloat(final float x, final float y) {
return ex.submit(new Callable<Float>() {
@Override
public Float call() throws Exception {
System.out.println("starting " + x + " + " + y);
pause(2000);
return x + y;
}
});
}
public void shutdown() {
ex.shutdown();
}
public static void main(String[] args) {
ActiveObjectDemo d1 = new ActiveObjectDemo();
// Prevents ConcurrentModificatinException:
List<Future<?>> results = new CopyOnWriteArrayList<Future<?>>();
for (float f = 0.0f; f < 1.0f; f += 0.2f)
results.add(d1.calculateFloat(f, f));
for (int i = 0; i < 5; i++)
results.add(d1.calculateInt(i, i));
System.out.println("All asynch calls made");
while (results.size() > 0) {
for (Future<?> f : results)
if (f.isDone()) {
try {
System.out.println(f.get());
} catch (Exception e) {
throw new RuntimeException(e);
}
results.remove(f);
}
}
d1.shutdown();
}
}
/*
All asynch calls made
starting 0.0 + 0.0
0.0
starting 0.2 + 0.2
starting 0.4 + 0.4
0.4
0.8
starting 0.6 + 0.6
1.2
starting 0.8 + 0.8
1.6
starting 0 + 0
0
starting 1 + 1
2
starting 2 + 2
4
starting 3 + 3
6
starting 4 + 4
8
*/