Java多线程 两阶段终止模式Two-Phase Termination Patter

目录

• 1、两阶段终止模式介绍

• 2、Terminator代码演示

• 3、TerminationRequester

• 4、模拟客户端或者服务端都可能终止服务的例子

• 5、mac telnet模拟客户端输入

1、两阶段终止模式介绍

有时候，我们希望提前结束线程，但安全可靠地停止线程，并不是一件容易的事情，如果立即停止线程，会使共享的数据结构处于不一致的状态，如目前已经废弃使用的Thread类的stop方法(它会使线程在抛出java.lang.ThreadDeath之后终止线程，即使是在执行synchronized方法的时候)。更好的做法是执行完终止处理，再终止线程，即Two-phase Termination，两阶段终止模式。

该模式有两个角色：

• Terminator，终止者，负责接收终止请求，执行终止处理，处理完成后再终止自己。

• TerminationRequester：终止请求发出者，用来向Terminator发出终止请求。
  

2、Terminator代码演示

该模式示例代码如下：

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 {
           while (!terminated) {
               System.out.println(Thread.currentThread().getName()+" "+counter++);
               Thread.sleep(random.nextInt(1000));
           }
       } catch (InterruptedException e) {
           e.printStackTrace();
       } finally {
           this.clean();
       }
   }

private void clean() {
       System.out.println("do some clean work for the second phase,current counter "+counter);

}

public void close() {
       this.terminated = true;
       this.interrupt();
   }
}

3、TerminationRequester

public class CounterTest {
   public static void main(String[] args) throws InterruptedException {
       CounterIncrement counterIncrement = new CounterIncrement();
       counterIncrement.start();

Thread.sleep(15_000L);
       //主动清理
       counterIncrement.close();
   }
}

这段代码可以看出实现两阶段终止模式必须注意的是：

使用线程停止标志和interrupt方法，两者缺一不可

 public void close() {
       this.terminated = true;
       this.interrupt();
   }

这里使用了terminated作为线程停止标志，变量采用volatile修饰，避免了使用显式锁的开销，又保证了内存可见性。线程run方法会检查terminated属性，如果属性为true，就停止线程，但线程可能调用了阻塞方法，处于wait状态，任务也就可能永远不会检查terminated标志；线程也有可能处于sleep()状态，等sleep时间过后再执行终止状态，程序的响应性就下降了。你可以把方法改成如下运行，线程停止明显变慢了许多：

 public void close() {
       terminated = true;
 }

4、模拟客户端或者服务端都可能终止服务的例子

public class AppServer extends Thread {

private static final int DEFAULT_PORT = 12722;
   private final static ExecutorService executor = Executors.newFixedThreadPool(10);
   private int port;
   private volatile boolean start = true;
   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);
               executor.submit(clientHandler);
               this.clientHandlers.add(clientHandler);
           }

} catch (IOException e) {
           //throw new RuntimeException();
       } finally {
           this.dispose();
       }
   }

public void dispose() {
       System.out.println("dispose");
       this.clientHandlers.stream().forEach(ClientHandler::stop);
       this.executor.shutdown();
   }

public void shutdown() throws IOException {
       this.start = false;
       this.interrupt();
       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 printWriter = new PrintWriter(outputStream)) {
           while (running) {
               String message = br.readLine();
               if (message == null) {
                   break;
               }
               System.out.println("Come from client >" + message);
               printWriter.write("echo " + message+"\n");
               printWriter.flush();
           }
       } catch (IOException e) {
           //自动关闭的时候 将running
           this.running = false;
       }finally {
           this.stop();
       }

}

public void stop() {
       if (!running) {
           return;
       }
       this.running = false;
       try {
           this.socket.close();

} catch (IOException e) {

}
   }
}

public class AppServerClient {
   public static void main(String[] args) throws InterruptedException, IOException {
       AppServer server = new AppServer(12135);
       server.start();

Thread.sleep(20_000L);
       server.shutdown();
   }
}

5、mac telnet模拟客户端输入

bogon:~ kpioneer$ telnet localhost 12135
Trying ::1...
Connected to localhost.
Escape character is '^]'.
hello
echo hello
I love you
echo I love you
Connection closed by foreign host.

服务端输出：

Come from client >hello
Come from client >I love you
dispose

总结：

可以看到，在子类使用两阶段终止模式时，其只需要实现各自所需要执行的任务，并且更新当前任务的数量即可。在某些情况下，当前任务的数量也可以不进行更新，比如在进行终止时，不关心当前剩余多少任务需要执行。

来源：https://juejin.cn/post/7023614457683116069