赞
踩
到目前为止,你学到的都是有关顺序编程的知识.即程序中的所有事物在任意时刻都只能执行一个步骤.
我们可以很公正地说,并发"具有可论证的确定性,但是实际上具有不可确定性".这就说,你可以得出结论,通过仔细设计和代码审查,编写能够正确工作的并发程序是可能的.
实际上,你可能无法编写出能够针对你的并发程序生成故障条件的测试代码.所产生的故障经常是偶尔发生的,并且是以客户端的形式出现的.这是研究并发问题的最强理由:如果视而不见,你就会遭其反噬.
Web系统是最常见的Java应用系统之一,而基本的Web库类和Servlet具有天生的多线程性–这很重要,因为Web服务器经常包含多个处理器,而并发是充分利用这些处理器的理想方式.
注意,你可能很容易就会变得过分自信,在你编写任何复杂程序之前,应该学习一下专门讨论这个主题的书籍.
并发编程令人困惑的一个主要原因是:使用并发时需要解决的问题有多个,而实现并发的方式也有多种,并且在这两者之间没有明显的映射关系(而且通常只具有模糊的界限).因此你必须理解所有这些问题和特例,以便有效地使用并发.
并发通常是提高运行在单处理器上的程序的性能.这有点违背直觉.如果仔细考虑一下就会发现在但处理器上运行的并发程序开销确实应该比该程序的所有部分都顺序的开销大,因为其中增加了所谓上下文切换的代价(从一个任务切换到另一个任务).表面上看,将程序的所有部分当作单个的任务运行好像是开销更小一点,并且可以节省上下文切换的代价.
是这个问题变得有些不同的是阻塞.如果程序中的某个任务因为该程序控制范围之外的某些条件(通常是I/O)而导致不能继续执行,那么我们就说这个任务或线程阻塞了.如果没有并发,则整个程序都将停止下来,直至外部条件发生变化.事实上从性能的角度看,如果没有任务会阻塞,那么在单处理器机器上使用并发就没有任何意义.
常见示例是事件驱动的编程.实际上,使用并发最吸引人的一个原因就是要产生具有可响应的用户界面.如果不适用并发,则产生可响应用户界面的唯一方式就是所有的任务都周期地检查用户输入.通过创建单独的执行线程来响应用户的输入,即使这个线程在大多数时间都是阻塞的,但是程序可以保证具有一定成都的可响应性.
实现并发最直接的方式实在操作系统级别使用进程(隔离性).但是由于进程通常有数量和开销的限制所以不行,于此相反的是,像java所使用的这种并发系统会共享诸如内存和I/O这样的资源,因此编写多线程程序的基本的困难在于在协调不同线程驱动的任务之间对这些资源的使用,以使得这些资源不会同时被多个任务访问.
某些编程语言被设计为可以将并发任务彼此隔离,这些语言通常被称为函数型语言.Erlang就是这样的语言,如果你发现程序中某个部分必须大量使用并发,并且在试图构建这个部分碰到了过多的问题,那么你可以考虑使用这样的并发语言来创建这个部分.
协作多线程,java线程机制是抢占式的,这表示调度机制会周期性的终端线程,将上下文切换到另一个线程.在协作式系统中,每个任务都会自动地放弃控制,这要求程序员要有意识地在每个任务中插入某种类型的让步语句.协作式系统的优势是双重的:上下文切换的开销通常比抢占式系统要低廉的许多,并且对可以同时执行的线程数量在理论上没有任何限制.当你处理大量的仿真元素时,这是一中理想的解决方案.
(手动码字的时候突然发现自己还是对线程基本了解不足所以把基本代码还是码上吧!!!)
线程的一大好处是可以是你从这个层次抽身出来,即代码不必知道它是运行在具有一个还是多个CPU的机器上.所以,使用线程机制是一种建立透明的 可扩展的程序的方法.多任务和多线程往往是使用多处理器系统的最合理方式.
//: concurrency/LiftOff.java // Demonstration of the Runnable interface. public class LiftOff implements Runnable { protected int countDown = 10; // Default private static int taskCount = 0; private final int id = taskCount++; public LiftOff() {} public LiftOff(int countDown) { this.countDown = countDown; } public String status() { return "#" + id + "(" + (countDown > 0 ? countDown : "Liftoff!") + "), "; } public void run() { while(countDown-- > 0) { System.out.print(status()); Thread.yield(); } } } ///:~ public class MainThread { public static void main(String[] args) { LiftOff launch = new LiftOff(); launch.run(); } } /* Output: #0(9), #0(8), #0(7), #0(6), #0(5), #0(4), #0(3), #0(2), #0(1), #0(Liftoff!), *///:~
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
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();
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
下面来探究一下Exector及其实现源码
可以看到它是ExecutorSevice的父接口有三个实现类
可以看到AbstractExecutorService提供了ExecutorSevice方法的默认实现,而另两个实现继承此类
可以看到API的规律,实现类中Abstract开头的类总是默认实现了接口大部分的方法
带有接口+s的类作为一个辅助工具类出现
public class SelManaged implements Runnable { private int countDown = 5; private Thread t = new Thread(this); public SelManaged(){ t.start(); } public String toString(){ return Thread.currentThread().getName() + "(" + countDown + "), "; } @Override public void run() { while(true){ System.out.print(this); if (--countDown == 0) return; } } public static void main(String[] args) { for (int i = 0; i < 5; i++) { new SelManaged(); } } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
class InnerThread1{ private int countDown = 5; private Inner inner; private class Inner extends Thread{ Inner(String name){ super(name); start(); } public void run(){ try { while(true){ print(this); if (--countDown == 0){ return ; } sleep(10); } }catch (InterruptedException e){ print("interrupted"); } } public String toString(){ return getName() + ": " + countDown; } } public InnerThread1(String name){ inner = new Inner(name); } } //Using an anonymous inner class: class InnerThread2{ private int countDown = 5; private Thread t; public InnerThread2(String name){ t = new Thread(name){ public void run(){ try { while(true){ print(this); if (--countDown == 0)return; sleep(10); } }catch (InterruptedException e){ print("sleep() interrupted"); } } public String toString(){ return getName() + ": " + countDown; } }; t.start(); } } //Using a named Runnable implementation 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(); } public void run(){ try { while(true){ print(this); if (--countDown == 0)return ; TimeUnit.MILLISECONDS.sleep(10); } }catch (InterruptedException e){ print("sleep() interruption"); } } public String toString(){ return t.getName() + ": " + countDown; } } public InnerRunnable1(String name){ inner = new Inner(name); } } // Using an anonymous Runnable implementation: class InnerRunnable2 { private int countDown = 5; private Thread t; public InnerRunnable2(String name) { t = new Thread(new Runnable() { public void run() { try { while(true) { print(this); if(--countDown == 0) return; TimeUnit.MILLISECONDS.sleep(10); } } catch(InterruptedException e) { print("sleep() interrupted"); } } public String toString() { return Thread.currentThread().getName() + ": " + countDown; } }, name); t.start(); } } // A separate method to run some code as a task: 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) { print(this); if(--countDown == 0) return; sleep(10); } } catch(InterruptedException e) { print("sleep() interrupted"); } } public String toString() { return getName() + ": " + countDown; } }; t.start(); } } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
- 115
- 116
- 117
- 118
- 119
- 120
- 121
- 122
- 123
- 124
- 125
- 126
- 127
- 128
- 129
- 130
- 131
- 132
- 133
- 134
- 135
- 136
- 137
- 138
- 139
import java.util.concurrent.*; public class ExceptionThread implements Runnable { public void run() { throw new RuntimeException(); } public static void main(String[] args) { ExecutorService exec = Executors.newCachedThreadPool(); exec.execute(new ExceptionThread()); } } ///:~ //将main的主体放到try-catch语句块中是没有作用的:这将产生与前面实例相同的效果:为捕获的异常 public class NaiveExceptionHandling { public static void main(String[] args) { try { ExecutorService exec = Executors.newCachedThreadPool(); exec.execute(new ExceptionThread()); } catch(RuntimeException ue) { // This statement will NOT execute! System.out.println("Exception has been handled!"); } } } ///:~
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
class ExceptionThread2 implements Runnable { @Override public void run() { Thread t = Thread.currentThread(); System.out.println("run() by " + t); System.out.println("eh = " + t.getUncaughtExceptionHandler()); throw new RuntimeException(); } } class MyUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler{ @Override public void uncaughtException(Thread t, Throwable e) { System.out.println("caught " + e); } } class HandlerThreadFactory implements ThreadFactory{ public Thread newThread(Runnable r){ System.out.println(this + " creating new Thread"); Thread t = new Thread(r); System.out.println("created " + t); t.setUncaughtExceptionHandler( new MyUncaughtExceptionHandler() ); System.out.println("eh = " + t.getUncaughtExceptionHandler()); return t; } } public class CaptureUncaughtException { public static void main(String[] args) { ExecutorService exec = Executors.newCachedThreadPool(new HandlerThreadFactory()); exec.execute(new ExceptionThread2()); } }/*out zhuofai.demo12_1.HandlerThreadFactory@57829d67 creating new Thread created Thread[Thread-0,5,main] eh = zhuofai.demo12_1.MyUncaughtExceptionHandler@2acf57e3 run() by Thread[Thread-0,5,main] eh = zhuofai.demo12_1.MyUncaughtExceptionHandler@2acf57e3 zhuofai.demo12_1.HandlerThreadFactory@57829d67 creating new Thread created Thread[Thread-1,5,main] eh = zhuofai.demo12_1.MyUncaughtExceptionHandler@7e4e6121 caught java.lang.RuntimeException *///~
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
这个程序请Debugger一下试试,不是几句话能讲清楚的.
但是这里还是要探究几个函数和几个内部类(嵌套类))几个工厂类,有点迷不过感觉用的一般,所以到看的时候再次理解一下就行了.
处理器的使用:系统会检查线程专有版本,如果没有发现,则检查线程组是否有其专有的uncaughtException()方法,如果也没有,再调用defaultUncaughtExceptionHandler.
如果你正在写一个变量,它可能接下来将被另一个线程读取,或者正在读取一个上一次已经被另一个线程写过的变量,那么你必须使用同步,并且读写线程都必须用相同的监视器锁同步.(每个访问临界共享资源的方法都必须被同步,否则它们就不会正确地工作)
Lock 实现提供了比使用 synchronized 方法和语句可获得的更广泛的锁定操作。此实现允许更灵活的结构,可以具有差别很大的属性,可以支持多个相关的 Condition 对象。
有关java线程的讨论中,一个常不正确的知识是"原子操作不需要进行同步控制". 原子操作是不能被线程调度机制中断的操作,一旦操作开始,那么它一定可以在可能发生的"上下文切换"之前(切换到其他线程执行)执行完毕.
Goetz测试:如果你可以编写用于现代微处理器的高性能JVM,那么就有资格去考虑是否可以避免同步.
原子性可以应用于除long和double之外的所有基本类型之上的"简单操作".对于读取和写入除long和double之外的基本类型变量这样的操作,可以保证它们会被当作不可分(原子)的操作来操作内存.但是JVM可以将64位(long和double变量)的读取和写入当作两个分离的32位操作来执行,这就产生了一个读取和写入操作中间发生上下文切换,从而导致不同的任务可以看到不正确结果的可能性(这有时被称为撕裂,因为你可能会看到部分被修改过的数值).
但是,当你定义long或double变量时,如果使用volatile关键字,就会获得(简单的赋值与返回操作的)原子性(注意,在JavaSE5之前volatile一直未能正确工作).不同的JVM可以任一地提供更强的保证,但是你不应该依赖于平台相关性的特性.
synchronized(syncObject){
//This code can be accessed
//by only one task at a time
}
- 1
- 2
- 3
- 4
class DualSynch { private Object syncObject = new Object(); public synchronized void f() { for(int i = 0; i < 5; i++) { print("f()"); Thread.yield(); } } public void g() { synchronized(syncObject) { for(int i = 0; i < 5; i++) { print("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(); } } /* Output: (Sample) g() f() g() f() g() f() g() f() g() f() *///:~
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
class Accessor implements Runnable { private final int id; public Accessor(int idn) { id = idn; } 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 Exception { ExecutorService exec = Executors.newCachedThreadPool(); for(int i = 0; i < 5; i++) exec.execute(new Accessor(i)); TimeUnit.SECONDS.sleep(3); // Run for a while exec.shutdownNow(); // All Accessors will quit } } /* Output: (Sample) #0: 9259 #1: 556 #2: 6694 #3: 1862 #4: 962 #0: 9260 #1: 557 #2: 6695 #3: 1863 #4: 963 ... *///:~
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
//有点复杂,可以做练习然后写一个简写版的 class Count { private int count = 0; private Random rand = new Random(47); // Remove the synchronized keyword to see counting fail: public synchronized int increment() { int temp = count; if(rand.nextBoolean()) // Yield half the time 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<Entrance>(); private int number = 0; // Doesn't need synchronization to read: private final int id; //**为了结束线程而设的静态volatile,多个线程共享同一份canceled** private static volatile boolean canceled = false; // Atomic operation on a volatile field: public static void cancel() { canceled = true; } public Entrance(int id) { this.id = id; // Keep this task in a list. Also prevents // garbage collection of dead tasks: entrances.add(this); } public void run() { while(!canceled) { synchronized(this) { ++number; } print(this + " Total: " + count.increment()); try { TimeUnit.MILLISECONDS.sleep(100); } catch(InterruptedException e) { print("sleep interrupted"); } } print("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 Exception { ExecutorService exec = Executors.newCachedThreadPool(); for(int i = 0; i < 5; i++) exec.execute(new Entrance(i)); // Run for a while, then stop and collect the data: TimeUnit.SECONDS.sleep(3); Entrance.cancel(); exec.shutdown(); if(!exec.awaitTermination(250, TimeUnit.MILLISECONDS)) print("Some tasks were not terminated!"); print("Total: " + Entrance.getTotalCount()); print("Sum of Entrances: " + Entrance.sumEntrances()); } } /* Output: (Sample) Entrance 0: 1 Total: 1 Entrance 2: 1 Total: 3 Entrance 1: 1 Total: 2 Entrance 4: 1 Total: 5 Entrance 3: 1 Total: 4 ... Entrance 1: 30 Total: 146 Entrance 0: 30 Total: 149 Entrance 3: 30 Total: 148 Entrance 4: 30 Total: 150 Stopping Entrance 2: 30 Stopping Entrance 1: 30 Stopping Entrance 0: 30 Stopping Entrance 3: 30 Stopping Entrance 4: 30 Total: 150 Sum of Entrances: 150 *///:~
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
class SynchronizedBlocked implements Runnable{ public synchronized void f(){ while(true)//Never releases lock Thread.yield(); } public SynchronizedBlocked(){ new Thread(){ public void run(){ f();//Lock acquired by this thread } }.start();//初始化过程完毕了已经并没有一直在初始化块里面没出来 } @Override public void run() { print("Trying to call f()"); f(); print("Exiting SynchronizedBlocked.run()"); } } public class Interrupting { private static ExecutorService exec = Executors.newCachedThreadPool(); static void test(Runnable r)throws InterruptedException{ Future<?> f = exec.submit(r); TimeUnit.MILLISECONDS.sleep(100); print("Interrupting " + r.getClass().getName()); f.cancel(true);//Interrupts if running print("Interrupt sent to " + r.getClass().getName()); } public static void main(String[] args) throws InterruptedException { test(new SynchronizedBlocked()); TimeUnit.SECONDS.sleep(3); print("Aborting with System.exit(0)"); System.exit(0);//...since last 2 interrupts failed } }/* Trying to call f() Interrupting SynchronizedBlocked Interrupt sent to SynchronizedBlocked Aborting with System.exit(0) *///:~
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
为了演示SynchronizedBlock我们首先必须获取锁.在匿名Thread类的对象通过调用f()获取了锁(这个线程必须有别于为SynchronizedBlock驱动run()线程,因为一个线程可以多次获得某个对象锁).由于f()永远都不会返回.因此这个锁永远不会释放,而SynchronizedBlock.run()在试图调用f(),并阻塞以等待这个锁被释放.
你可以中断对sleep()的调用(或者任何要求抛出InterruptedException的调用).但是你不能中断正在试图获取synchronized锁或者试图执行I/O操作的线程,这意味着I/O具有锁住你的多线程程序的潜在可能.特别是基于Web的程序.这更是关乎利害.(shutdownNow()被调用之后以及在输入流上调用close()之前的延迟强调的是一旦底层资源关闭,任务将解除阻塞)关于这个后面的JavaSE5并发类库中添加了一个新特性.(后面有写)
public class MultiLock { public synchronized void f1(int count) { if(count-- > 0) { print("f1() calling f2() with count " + count); f2(count); } } public synchronized void f2(int count) { if(count-- > 0) { print("f2() calling f1() with count " + count); f1(count); } } public static void main(String[] args) throws Exception { final MultiLock multiLock = new MultiLock(); new Thread() { public void run() { multiLock.f1(10); } }.start(); } } /* Output: 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 f2() calling f1() with count 0 *///:~
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
//经典程序 展示了典型的惯用法,应该在run()方法中使用它来处理在终端状态被设置时,被阻塞和不被阻塞的各种可能: class NeedsCleanup { private final int id; public NeedsCleanup(int ident) { id = ident; print("NeedsCleanup " + id); } public void cleanup() { print("Cleaning up " + id); } } class Blocked3 implements Runnable { private volatile double d = 0.0; public void run() { try { while(!Thread.interrupted()) { // point1 NeedsCleanup n1 = new NeedsCleanup(1); // Start try-finally immediately after definition // of n1, to guarantee proper cleanup of n1: try { print("Sleeping"); TimeUnit.SECONDS.sleep(1); // point2 NeedsCleanup n2 = new NeedsCleanup(2); // Guarantee proper cleanup of n2: try { print("Calculating"); // A time-consuming, non-blocking operation: for(int i = 1; i < 2500000; i++) d = d + (Math.PI + Math.E) / d; print("Finished time-consuming operation"); } finally { n2.cleanup(); } } finally { n1.cleanup(); } } print("Exiting via while() test"); } catch(InterruptedException e) { print("Exiting via InterruptedException"); } } } public class InterruptingIdiom { public static void main(String[] args) throws Exception { if(args.length != 1) { print("usage: java InterruptingIdiom delay-in-mS"); System.exit(1); } Thread t = new Thread(new Blocked3()); t.start(); TimeUnit.MILLISECONDS.sleep(new Integer(args[0])); t.interrupt(); } } /* Output: (Sample) NeedsCleanup 1 Sleeping NeedsCleanup 2 Calculating Finished time-consuming operation Cleaning up 2 Cleaning up 1 NeedsCleanup 1 Sleeping Cleaning up 1 Exiting via InterruptedException *///:~
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
BlockingQueue 实现主要用于生产者-使用者队列,但它另外还支持 Collection 接口。因此,举例来说,使用 remove(x) 从队列中移除任意一个元素是有可能的。然而,这种操作通常不 会有效执行,只能有计划地偶尔使用,比如在取消排队信息时。
//以下是基于典型的生产者-使用者场景的一个用例。注意,BlockingQueue 可以安全地与多个生产者和多个使用者一起使用。 class Producer implements Runnable { private final BlockingQueue queue; Producer(BlockingQueue q) { queue = q; } public void run() { try { while(true) { queue.put(produce()); } } catch (InterruptedException ex) { ... handle ...} } Object produce() { ... } } class Consumer implements Runnable { private final BlockingQueue queue; Consumer(BlockingQueue q) { queue = q; } public void run() { try { while(true) { consume(queue.take()); } } catch (InterruptedException ex) { ... handle ...} } void consume(Object x) { ... } } class Setup { void main() { BlockingQueue q = new SomeQueueImplementation(); Producer p = new Producer(q); Consumer c1 = new Consumer(q); Consumer c2 = new Consumer(q); new Thread(p).start(); new Thread(c1).start(); new Thread(c2).start(); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
//示例用法: 下面给出了两个类,其中一组 worker 线程使用了两个倒计数锁存器: //第一个类是一个启动信号,在 driver 为继续执行 worker 做好准备之前,它会阻止所有的 worker 继续执行。 //第二个类是一个完成信号,它允许 driver 在完成所有 worker 之前一直等待。 class Driver { // ... void main() throws InterruptedException { CountDownLatch startSignal = new CountDownLatch(1); CountDownLatch doneSignal = new CountDownLatch(N); for (int i = 0; i < N; ++i) // create and start threads new Thread(new Worker(startSignal, doneSignal)).start(); doSomethingElse(); // don't let run yet startSignal.countDown(); // let all threads proceed doSomethingElse(); doneSignal.await(); // wait for all to finish } } class Worker implements Runnable { private final CountDownLatch startSignal; private final CountDownLatch doneSignal; Worker(CountDownLatch startSignal, CountDownLatch doneSignal) { this.startSignal = startSignal; this.doneSignal = doneSignal; } public void run() { try { startSignal.await(); doWork(); doneSignal.countDown(); } catch (InterruptedException ex) {} // return; } void doWork() { ... } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
/*public CyclicBarrier(int parties, Runnable barrierAction)创建一个新的 CyclicBarrier,它将在给定数量的参与者(线程)处于等待状态时启动,并在启动 barrier 时执行给定的屏障操作,该操作由最后一个进入 barrier 的线程执行。 参数: parties - 在启动 barrier 前必须调用 await() 的线程数 barrierAction - 在启动 barrier 时执行的命令;如果不执行任何操作,则该参数为 null 抛出: IllegalArgumentException - 如果 parties 小于 1 */ class Solver { final int N; final float[][] data; final CyclicBarrier barrier; class Worker implements Runnable { int myRow; Worker(int row) { myRow = row; } public void run() { while (!done()) { processRow(myRow); try { barrier.await(); } catch (InterruptedException ex) { return; } catch (BrokenBarrierException ex) { return; } } } } public Solver(float[][] matrix) { data = matrix; N = matrix.length; barrier = new CyclicBarrier(N, new Runnable() { public void run() { mergeRows(...); } }); for (int i = 0; i < N; ++i) new Thread(new Worker(i)).start(); waitUntilDone(); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
/* 一个计数信号量。从概念上讲,信号量维护了一个许可集。如有必要,在许可可用前会阻塞每一个 acquire(),然后再获取该许可。每个 release() 添加一个许可,从而可能释放一个正在阻塞的获取者。但是,不使用实际的许可对象,Semaphore 只对可用许可的号码进行计数,并采取相应的行动。 Semaphore 通常用于限制可以访问某些资源(物理或逻辑的)的线程数目。例如,下面的类使用信号量控制对内容池的访问: */ class Pool { private static final int MAX_AVAILABLE = 100; private final Semaphore available = new Semaphore(MAX_AVAILABLE, true); public Object getItem() throws InterruptedException { available.acquire(); return getNextAvailableItem(); } public void putItem(Object x) { if (markAsUnused(x)) available.release(); } // Not a particularly efficient data structure; just for demo protected Object[] items = ... whatever kinds of items being managed protected boolean[] used = new boolean[MAX_AVAILABLE]; protected synchronized Object getNextAvailableItem() { for (int i = 0; i < MAX_AVAILABLE; ++i) { if (!used[i]) { used[i] = true; return items[i]; } } return null; // not reached } protected synchronized boolean markAsUnused(Object item) { for (int i = 0; i < MAX_AVAILABLE; ++i) { if (item == items[i]) { if (used[i]) { used[i] = false; return true; } else return false; } } return false; } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
/* 可以在对中对元素进行配对和交换的线程的同步点。每个线程将条目上的某个方法呈现给 exchange 方法,与伙伴线程进行匹配,并且在返回时接收其伙伴的对象。Exchanger 可能被视为 SynchronousQueue 的双向形式。Exchanger 可能在应用程序(比如遗传算法和管道设计)中很有用。 用法示例:以下是重点介绍的一个类,该类使用 Exchanger 在线程间交换缓冲区,因此,在需要时,填充缓冲区的线程获取一个新腾空的缓冲区,并将填满的缓冲区传递给腾空缓冲区的线程。 */ class FillAndEmpty { Exchanger<DataBuffer> exchanger = new Exchanger<DataBuffer>(); DataBuffer initialEmptyBuffer = ... a made-up type DataBuffer initialFullBuffer = ... class FillingLoop implements Runnable { public void run() { DataBuffer currentBuffer = initialEmptyBuffer; try { while (currentBuffer != null) { addToBuffer(currentBuffer); if (currentBuffer.isFull()) currentBuffer = exchanger.exchange(currentBuffer); } } catch (InterruptedException ex) { ... handle ... } } } class EmptyingLoop implements Runnable { public void run() { DataBuffer currentBuffer = initialFullBuffer; try { while (currentBuffer != null) { takeFromBuffer(currentBuffer); if (currentBuffer.isEmpty()) currentBuffer = exchanger.exchange(currentBuffer); } } catch (InterruptedException ex) { ... handle ...} } } void start() { new Thread(new FillingLoop()).start(); new Thread(new EmptyingLoop()).start(); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。