多线程设计模式总结(二)

2014/07/03

本文原创作者:Cloud Chou. 欢迎转载,请注明出处和本文链接

接上一篇《多线程设计模式总结(一)》,这篇博客再介绍5个多线程设计模式

7)Thread-Per-Message

实现某个方法时创建新线程去完成任务,而不是在本方法里完成任务,这样可提高响应性,因为有些任务比较耗时。

示例程序:

1
2
3
4
5
6
7
8
9
10
public class Host {
private final Handler _handler=new Handler();
public void request(final int count, final char c){
new Thread(){
public void run(){
_handler.handle(count, c);
}
}.start();
}
}

实现Host类的方法时,新建了一个线程调用Handler对象处理request请求。

每次调用Host对象的request方法时都会创建并启动新线程,这些新线程的启动顺序不是确定的。

适用场景:

适合在操作顺序无所谓时使用,因为请求的方法里新建的线程的启动顺序不是确定的。

在不需要返回值的时候才能使用,因为request方法不会等待线程结束才返回,而是会立即返回,这样得不到请求处理后的结果。

注意事项:

每次调用都会创建并启动一个新线程,对新建线程没有控制权,实际应用中只有很简单的请求才会用Thread-Per-Message这个模式,因为通常我们会关注返回结果,也会控制创建的线程数量,否则系统会吃不消。

8)Worker Thread

在Thread-Per-Message模式里,每次函数调用都会启动一个新线程,但是启动新线程的操作其实是比较繁重的,需要比较多时间,系统对创建的线程数量也会有限制。我们可以预先启动一定数量的线程,组成线程池,每次函数调用时新建一个任务放到任务池,预先启动的线程从任务池里取出任务并执行。这样便可以控制线程的数量,也避免了每次启动新线程的高昂代价,实现了资源重复利用。

示例程序:

Channel类:

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
public class Channel {
private static final int MAX_REQUEST = 100;
private final Request[] _request_queue;
private int tail;
private int head;
private int count;

private WorkerThread[] _thread_pool;

public Channel(int threads) {
_request_queue = new Request[MAX_REQUEST];
tail = 0;
head = 0;
count = 0;
_thread_pool = new WorkerThread[threads];
for (int i = 0; i < threads; i++) {
_thread_pool[i] = 
   new WorkerThread("Worker-" + i, this);
}
}

public void startWorkers() {
for (int i = 0; i < _thread_pool.length; i++)
_thread_pool[i].start();
}

public synchronized Request takeRequest() 
             throws InterruptedException {
while (count <= 0) {
wait();
}
Request request = _request_queue[head];
head = (head + 1) % _request_queue.length;
count--;
notifyAll();
return request;
}

public synchronized void putRequest(Request request)
throws InterruptedException {
while (count >= _request_queue.length) {
wait();
}
_request_queue[tail] = request;
tail = (tail + 1) % _request_queue.length;
count++;
notifyAll();
}

}

WorkerThread类:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
public class WorkerThread extends Thread {
private final Channel _channel;

public WorkerThread(String name, Channel channel) {
super(name);
_channel = channel;
}

@Override
public void run() {
while (true) {
Request request;
try {
request = _channel.takeRequest();
request.execute();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} 
}
}

}

channel类集成了线程池和任务池,对外提供了startWorkers方法,外界可调用该方法启动所有工作线程,然后通过putRequest方法向任务池添加任务,工作者线程会自动从任务池里取出任务并执行。

适用场景:

和Thread-Per-Message模式一样,Worker Thread模式实现了invocation和exectution的分离,即调用和执行分离,调用者调用方法运行在一个线程,任务的执行在另一个线程。调用者调用方法后可立即返回,提高了程序的响应性。另外也正是因为调用和执行分离了,我们可以控制任务的执行顺序,还可以取消任务,还能分散处理,将任务交给不同的机器执行,如果没有将调用和执行分离,这些特性是无法实现的。

适合有大量任务并且还需要将任务执行分离的程序,比如象应用分发类App,需要经常和服务器通信获取数据,并且通信消息可能还有优先级。

注意事项:

注意控制工作者线程的数量,如果过多,那么会有不少工作者线程并没有工作,会浪费系统资源,如果过少会使得任务池里塞满,导致其它线程长期阻塞。可根据实际工作调整线程数量,和任务池里的最大任务池数。

如果worker thread只有一条,工人线程处理的范围就变成单线程了,可以省去共享互斥的必要。通常GUI框架都是这么实现的,操作界面的线程只有一个,界面元素的方法不需要进行共享互斥。如果操作界面的线程有多个,那么必须进行共享互斥,我们还会经常设计界面元素的子类,子类实现覆盖方法时也必须使用synchronized进行共享互斥,引入共享互斥后会引入锁同步的开销,使程序性能降低,并且如果有不恰当的获取锁的顺序,很容易造成死锁,这使得GUI程序设计非常复杂,故此GUI框架一般都采用单线程。

Java 5的并发包里已经有线程池相关的类,无需自己实现线程池。可使用Executors的方法启动线程池,这些方法包括newFixedThreadPool,newSingleThreadExecutor,newCachedThreadPool,newScheduledThreadPool等等。

9)Future

在Thread-Per-Message模式和Worker Thread模式里,我们实现了调用和执行分离。但是通常我们调用一个函数是可以获得返回值的,在上述两种模式里,虽然实现了调用和执行相分离,但是并不能获取调用执行的返回结果。Future模式则可以获得执行结果,在调用时返回一个Future,可以通过Future获得真正的执行结果。

示例程序:

Host类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public class Host {

public Data request(final int count, final char c) {
System.out.println(" request (" + count 
                 + ", " + c + " ) BEGIN");
final FutureData future = new FutureData();
new Thread() {
@Override
public void run() {
RealData realData = new RealData(count, c);
future.setRealData(realData);
}

}.start();
return future;
}

}

Data接口

1
2
3
public interface Data {
public String getContent();
}

FutureData类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public class FutureData implements Data {
private boolean _ready = false;
private RealData _real_data = null;

public synchronized void setRealData(RealData realData) {
if (_ready)
return;
_real_data = realData;
_ready = true;
notifyAll();
}

@Override
public synchronized String getContent() {
while (!_ready) {
try {
wait();
} catch (InterruptedException e) {
}
}
return _real_data.getContent();
}

}

RealData类

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
public class RealData implements Data {

private final String _content;

public RealData(int count, char c) {
System.out.println("Making  Realdata(" 
          + count + "," + c + ") BEGIN");
char[] buffer = new char[count];
for (int i = 0; i < count; i++) {
buffer[i] = c;
try {
Thread.sleep(100);
} catch (Exception e) {
}
}
System.out.println(" making Real Data(" 
          + count + "," + c + ") END");
_content = new String(buffer);
}

@Override
public String getContent() {
return _content;
}

}

适用场景:

如果既想实现调用和执行分离,又想获取执行结果,适合使用Future模式。

Future模式可以获得异步方法调用的”返回值”,分离了”准备返回值”和”使用返回值”这两个过程。

注意事项:

Java 5并发包里已经有Future接口,不仅能获得返回结果,还能取消任务执行。当调用ExecutorService对象的submit方法向任务池提交一个Callable任务后,可获得一个Future对象,用于获取任务执行结果,并可取消任务执行。

10)Two-Phase Termination

这一节介绍如何停止线程,我们刚开始学习线程时,可能很容易犯的错就是调用Thread的stop方法停止线程,该方法确实能迅速停止线程,并会让线程抛出异常。但是调用stop方法是不安全的,如果该线程正获取了某个对象的锁,那么这个锁是不会被释放的,其他线程将继续被阻塞在该锁的条件队列里,并且也许线程正在做的工作是不能被打断的,这样可能会造成系统破坏。从线程角度看,只有执行任务的线程本身知道该何时恰当的停止执行任务,故此我们需要用Two-Phase Termination模式来停止线程,在该模式里如果想停止某个线程,先设置请求线程停止的标志为true,然后调用Thread的interrupt方法,在该线程里每完成一定工作会检查请求线程停止的标志,如果为true,则安全地结束线程。

示例程序:

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
public class CountupThread extends Thread {
private long counter = 0;

private volatile boolean _shutdown_requested = false;

public void shutdownRequest() {
_shutdown_requested = true;
interrupt();
}

public boolean isShutdownRequested() {
return _shutdown_requested;
}

@Override
public void run() {
try {
while (!_shutdown_requested) {
doWork();
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
doShutdown();
}
}

private void doWork() throws InterruptedException {
counter++;
System.out.println("doWork: counter = " + counter);
Thread.sleep(500);
}

private void doShutdown() {
System.out.println("doShutDown: counter = " + counter);
}

}

外界可调用shutdownRequest来停止线程。

适用场景:

需要停止线程时,可考虑使用Two-Phase Termination模式

注意事项:

我们在请求线程停止时,若只设置请求停止标志,是不够的,因为如果线程正在执行sleep操作,那么会等sleep操作执行完后,再执行到检查停止标志的语句才会退出,这样程序响应性不好。

响应停止请求的线程如果只检查中断状态(不是说我们设置的停止标志)也是不够的,如果线程正在sleep或者wait,则会抛出InterruptedException异常,就算没有抛出异常,线程也会变成中断状态,似乎我们没必要设置停止标志,只需检查InterruptedException或者用isInterrupted方法检查当前线程的中断状态就可以了,但是这样做会引入潜在的危险,如果该线程调用的方法忽略了InterruptedException,或者该线程使用的对象的某个方法忽略了InterruptedException,而这样的情况是很常见的,尤其是如果我们使用某些类库代码时,又不知其实现,即使忽略了InterruptedException,我们也不知道,在这种情况下,我们无法检查到是否有其它线程正在请求本线程退出,故此说设置终端标志是有必要的,除非能保证线程所引用的所有对象(包括间接引用的)不会忽略InterruptedException,或者能保存中断状态。

中断状态和InterruptedException可以互转:

1)中断状态 -> InterruptedException

  • 1)中断状态 -> InterruptedException
  • ```java if(Thread.interrupted){ throw new InterruptedException() } ```
  • 2)InterruptedException -> 中断状态
  • ```java try{ Thread.sleep(1000); }catch (InterruptedException e) { Thread.currentThread().interrupt(); } ```
  • 3)InterruptedException -> InterruptedException
  • ```java InterruptedException savedInterruptException = null; ... try{ Thread.sleep(1000); }catch (InterruptedException e) { savedInterruptException=e; } ... if(savedInterruptException != null ) throw savedInterruptException; ```

11)Thread-Specific Storage

我们知道,如果一个对象不会被多个线程访问,那么就不存在线程安全问题。Thread-Specific Storage模式就是这样一种设计模式,为每个线程生成单独的对象,解决线程安全问题。不过为线程生成单独的对象这些细节对于使用者来说是隐藏的,使用者只需简单使用即可。需要用到ThreadLocal类,它是线程保管箱,为每个线程保存单独的对象。

示例程序:

Log类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public class Log {
private static ThreadLocal<TSLog> _ts_log_collection = 
                      new ThreadLocal<TSLog>();

public static void println(String s) {
getTSLog().println(s);
}

public static void close() {
getTSLog().close();
}

private static TSLog getTSLog() {
TSLog tsLog = _ts_log_collection.get();
if (tsLog == null) {
tsLog = new TSLog(
 Thread.currentThread().getName() + "-log.txt"
 );
_ts_log_collection.set(tsLog);
}
return tsLog;
}

}

TSLog类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
public class TSLog {
private PrintWriter _writer = null;

public TSLog(String fileName) {
try {
_writer = new PrintWriter(fileName);
} catch (FileNotFoundException e) {
e.printStackTrace();
}
}

public void println(String s) {
_writer.write(s);
}

public void close() {
_writer.close();
}
}

适用场景:

使用Thread-Specific Storgae模式可很好的解决多线程安全问题,每个线程都有单独的对象,如果从ThreadLocal类里获取线程独有对象的时间远小于调用对象方法的执行时间,可提高程序性能。因此在日志系统里如果可以为每个线程建立日志文件,那么特别适合使用Thread-Specific Storage模式。

注意事项:

采用Thread-Specific Storage模式意味着将线程特有信息放在线程外部,在示例程序里,我们将线程特有的TSLog放在了ThreadLocal的实例里。通常我们一般将线程特有的信息放在线程内部,比如建立一个Thread类的子类MyThread,我们声明的MyThread的字段,就是线程特有的信息。因为把线程特有信息放在线程外部,每个线程访问线程独有信息时,会取出自己独有信息,但是调试时会困难一些,因为有隐藏的context(当前线程环境), 程序以前的行为,也可能会使context出现异常,而是造成现在的bug的真正原因,我们比较难找到线程先前的什么行为导致context出现异常。

设计多线程程序,主体是指主动操作的对象,一般指线程,客体指线程调用的对象,一般指的是任务对象,会因为重点放在“主体”与“客体”的不同,有两种开发方式:

  • 1)Actor-based 注重主体
  • 2)Task-based 注重客体

Actor-based 注重主体,偏重于线程,由线程维护状态,将工作相关的信息都放到线程类的字段,类似这样

1
2
3
4
5
6
class Actor extends Thread{
操作者内部的状态
public void run(){
 从外部取得任务,改变自己内部状态的循环
}
}

Task-based注重客体,将状态封装到任务对象里,在线程之间传递这些任务对象,这些任务对象被称为消息,请求或者命令。使用这种开发方式的最典型的例子是Worker Thread 模式,生产者消费者模式。任务类似这样:

1
2
3
4
5
6
class Task implements Runnable{
 执行任务所需的信息
 public void run(){ 
   执行任务所需的处理内容
 }
}

实际上这两个开发方式是混用的,本人刚设计多线程程序时,总是基于Actor-based的思维方式,甚至在解决生产者消费者问题时也使用Actor-based思维方式,造成程序结构混乱,因此最好按实际场景来,适合使用Actor-based开发方式的就使用Actor-based,适合Task-based开发方式的就使用Task-based。

¥打赏5毛

取消

感谢您的支持,我会继续努力的!

扫码支持
赏个5毛,支持我把

打开支付宝扫一扫,即可进行扫码打赏哦

本篇目录