【详解】Java多线程之两个阶段终止设计模式
分析
Two-Phase Termination Pattern,指的就是当希望结束一个线程的时候,送出一个终止请求,但是不会马上停止,做一些刷新工作。进入“终止处理中”,在该状态下,不会进行该线程日常工作任务的操作,而是进行一些终止操作。
这个方式所考虑的因素如下:
-
1,必须要考虑到使得该线程能够安全的结束,Thread中的stop会有问题的,因为它会不管线程执行到哪里,都会马上停止,不能保证安全的结束。
-
2,一定能够进行正常的终止处理,在java中,这点可以使用finally来实现
-
3,能够高响应的终止,收到终止后,当线程在wait或者sleep或者join的时候,不用等到时间到才终止,而是马上中断线程的这些状态,进而进行终止操作。
当一个线程正在执行周期性的工作时候,在“作业中”发了停止执行绪的请求,此时该线程不能马上离开停止,而应该先做完本次周期内部的工作,然后进入“善后阶段”完成一些善后的工作,所谓的两阶段终止,即中止“运作阶段”,并完成“善后阶段”
,完整的完成执行绪的工作。
基本架构
- 主要通过的是try…finally…,进行必要的资源关闭
public class CounterIncrement extends Thread {
private volatile boolean terminated = false;
private int counter = 0;
private Random random = new Random(System.currentTimeMillis());
@Override
public void run() {
try {
try {
while (!terminated) {
System.out.println(Thread.currentThread().getName() + " " + counter++);
Thread.sleep(random.nextInt(1000));
}
} catch (InterruptedException e) {
// e.printStackTrace();
}
} finally {
clean();
}
}
private void clean() {
System.out.println("Do some clean work for the second phase. " + counter);
}
public void close() {
this.terminated = true;
this.interrupt();
}
}
网络编程中一个常见例子
public class AppServer extends Thread{
private int port;
private static final int DEFAULT_PORT = 12755;
private volatile boolean start = true;
private final ExecutorService executor = Executors.newFixedThreadPool(10);
private List<ClientHandler> clientHandlers = new ArrayList<>();
private ServerSocket server;
public AppServer(){
this(DEFAULT_PORT);
}
public AppServer(int port) {
this.port = port;
}
@Override
public void run() {
try {
server = new ServerSocket(port);
//监听是否有客户端连接
while (start){
Socket client = server.accept();
ClientHandler clientHandler = new ClientHandler(client);
clientHandlers.add(clientHandler);
executor.submit(clientHandler);
}
} catch (IOException e) {
throw new RuntimeException(e);
}finally {
this.dispose();
}
}
/** * 执行关闭操作 */
private void dispose() {
clientHandlers.forEach(ClientHandler::stop);
this.executor.shutdown();
try {
this.server.close();
} catch (IOException e) {
e.printStackTrace();
}
}
public void shutdown() throws IOException {
this.start = false;
this.server.close();
}
}
public class ClientHandler implements Runnable {
private final Socket socket;
private volatile boolean running = true;
public ClientHandler(Socket socket) {
this.socket = socket;
}
@Override
public void run() {
try (InputStream inputStream = socket.getInputStream();
OutputStream outputStream = socket.getOutputStream();
BufferedReader br = new BufferedReader(new InputStreamReader(inputStream));
PrintWriter pw = new PrintWriter(outputStream)
) {
while (running) {
String message = br.readLine();
if (message == null) {
return;
}
System.out.println("come from client > " + message);
pw.write("echo " + message + "\n");
pw.flush();
}
} catch (IOException e) {
e.printStackTrace();
this.running = false;
} finally {
stop();
}
}
public void stop() {
if (!running) {
return;
}
this.running = false;
try {
this.socket.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
public class AppServerClient {
public static void main(String[] args) throws InterruptedException, IOException {
AppServer appServer = new AppServer(13345);
appServer.start();
Thread.sleep(20000);
appServer.shutdown();
}
}