2. 程式人生 > >Java多執行緒簡單實現取消和進度

Java多執行緒簡單實現取消和進度

阿新 • • 發佈：2019-02-14

Java中簡單實現多執行緒排程時的可取消和顯示進度

一個簡單的多執行緒排程實現，統一開始，為了使得所有執行緒統一開始，類似運動員在聽到發令槍時一起進行，使用了CountDownLatch進行控制。

CountDownLatch beginLatch = new CountDownLatch(1);
CountDownLatch endLatch = new CountDownLatch(personCount);

主執行緒建立執行緒池，並進行排程，由於要在最後進行彙總結果，使用了FutureTask

List<FutureTask<String>> futureTaskList = new ArrayList<FutureTask<String>>();
for (int i = 0; i < personCount; i++) {
	futureTaskList.add(new FutureTask<String>(new ExecuteCallable(beginLatch, endLatch，i)));
}

ExecutorService execService = Executors.newFixedThreadPool(threadCount);
	for (FutureTask<String> futureTask : futureTaskList) {
		execService.execute(futureTask);
	}
beginLatch.countDown();

 這樣所有執行緒就會統一開始執行，執行完成後，彙總結果，並關閉執行緒池。

endLatch.await();
System.out.println("--------------");
for (FutureTask<String> futureTask : futureTaskList) {
	System.out.println(futureTask.get());
}
execService.shutdown();

 對於每個執行緒的執行，都需要共享變數beginLatch和endLatch，各執行緒程式碼：

public class ExecuteCallable implements Callable<String> {

	private int id;
	private CountDownLatch beginLatch;
	private CountDownLatch endLatch;

	public ExecuteCallable(CountDownLatch beginLatch, CountDownLatch endLatch,
			Exchanger<Integer> exchanger, int id,
			ConcurrentTaskExecutor concurrentTaskExecutor) {
		this.beginLatch = beginLatch;
		this.endLatch = endLatch;
		this.id = id;
	}

	@Override
	public String call() throws Exception {
		beginLatch.await();
		long millis = (long) (Math.random() * 10 * 1000);
		String result = String.format("Player :%s arrived, use %s millis", id, millis);
		Thread.sleep(millis);
		System.out.println(result);
		endLatch.countDown();
		return result;
	}

}

 每個執行緒在開始等待發令槍(beginLatch)，隨機等待一段時間（模擬執行時間），最後通知endLatch減一（執行完畢通知），並返回結果。

到這裡只是一個簡單的實現，我們並不能在主執行緒中實時瞭解各執行緒的執行情況，除非到了所有執行緒執行完畢(endLatch解除阻塞狀態)。這時候我們使用Exchanger機制來進行執行緒之間資料的交換，在每個執行緒執行完成後，將其完成的資料量傳給主執行緒進行重新整理（模擬進度條工作）。

主執行緒ConcurrentTaskExecutor類中：

Exchanger<Integer> exchanger = new Exchanger<Integer>();
beginLatch.countDown();

Integer totalResult = Integer.valueOf(0);
for (int i = 0; i < personCount; i++) {
	Integer partialResult = exchanger.exchange(Integer.valueOf(0));
        if(partialResult != 0){
	        totalResult = totalResult + partialResult;
		System.out.println(String.format("Progress: %s/%s", totalResult, personCount));
	}
}

endLatch.await();

 執行緒類ExecuteCallable建構函式加入exchanger

@Override
public String call() throws Exception {
        beginLatch.await();
	long millis = (long) (Math.random() * 10 * 1000);
	String result = String.format("Player :%s arrived, use %s millis", id, millis);
	Thread.sleep(millis);
	System.out.println(result);
	exchanger.exchange(1);
	endLatch.countDown();
	return result;
}

 在執行完成進行資料交換，返回本次執行進度給主執行緒（當前預設設定成1，可修改），主執行緒在所有執行緒執行完成前，endLatch.await()必定是阻塞狀態的，這樣主執行緒就能實時拿到子執行緒執行完成的進度資料。

下面我們再加入一個可以取消的功能，加入系統隨機在某個時間點進行取消操作，那麼開始執行的執行緒是無法進行實時響應了，只能等待當前操作執行完畢；如果執行緒還沒有開始執行，那麼就取消其行為。

更改的ExecuteCallable執行方法如下：

@Override
public String call() throws Exception {
	beginLatch.await();
	if(concurrentTaskExecutor.isCanceled()){
		endLatch.countDown();
		exchanger.exchange(0);
		return String.format("Player :%s is given up", id);
	}
	long millis = (long) (Math.random() * 10 * 1000);
	String result = String.format("Player :%s arrived, use %s millis", id, millis);
	Thread.sleep(millis);
	System.out.println(result);
	exchanger.exchange(1);
	endLatch.countDown();
	return result;
}

 其中concurrentTaskExecutor類中加入一個型別為boolean的canceled變數，注意這個變數必須是volatile的，以便能夠線上程間共享資料，並且該變數的setter和getter方法也是原子性的。

我們的取消操作不能放在主執行緒中操作，需要額外建立一個執行緒，並且這個執行緒也不能通過執行緒池進行排程，新建的InterruptRunnable類：

public class InterruptRunnable implements Runnable {

	private CountDownLatch beginLatch;
	private ConcurrentTaskExecutor concurrentTaskExecutor;

	public InterruptRunnable(ConcurrentTaskExecutor currConcurrentTaskExecutor, CountDownLatch beginLatch) {
		this.beginLatch = beginLatch;
		this.concurrentTaskExecutor = currConcurrentTaskExecutor;
	}

	@Override
	public void run() {
		try {
			beginLatch.await();
			long millis = (long) (Math.random() * 10 * 1000);
			System.out.println(String.format("System need sleep %s millis", millis));
			Thread.sleep(millis);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		concurrentTaskExecutor.setCanceled(true);
	}

}

 更改後的ConcurrentTaskExecutor，在執行發令前，先讓該中斷執行緒啟動，以便一起等待開始命令：

new Thread(new InterruptRunnable(this, beginLatch)).start();

beginLatch.countDown();

 最後執行結果（取決於中斷執行緒的隨機時間長短）：

System need sleep 2920 millis
Player :4 arrived, use 917 millis
Progress: 1/10
Player :5 arrived, use 1076 millis
Progress: 2/10
Player :3 arrived, use 2718 millis
Progress: 3/10
Player :1 arrived, use 4013 millis
Progress: 4/10
Player :0 arrived, use 8541 millis
Progress: 5/10
Player :2 arrived, use 8570 millis
Progress: 6/10
Player :6 arrived, use 7261 millis
Progress: 7/10
Player :7 arrived, use 7015 millis
Progress: 8/10
--------------
Player :0 arrived, use 8541 millis
Player :1 arrived, use 4013 millis
Player :2 arrived, use 8570 millis
Player :3 arrived, use 2718 millis
Player :4 arrived, use 917 millis
Player :5 arrived, use 1076 millis
Player :6 arrived, use 7261 millis
Player :7 arrived, use 7015 millis
Player :8 is given up
Player :9 is given up

最後，附上最終的程式程式碼

ConcurrentTaskExecutor：

public class ConcurrentTaskExecutor {

	private volatile boolean canceled = false;

	public void executeTask() throws Exception {
		int personCount = 10;
		int threadCount = 5;

		CountDownLatch beginLatch = new CountDownLatch(1);
		CountDownLatch endLatch = new CountDownLatch(personCount);
		Exchanger<Integer> exchanger = new Exchanger<Integer>();

		List<FutureTask<String>> futureTaskList = new ArrayList<FutureTask<String>>();
		for (int i = 0; i < personCount; i++) {
			futureTaskList.add(new FutureTask<String>(new ExecuteCallable(beginLatch, endLatch, exchanger, i, this)));
		}

		ExecutorService execService = Executors.newFixedThreadPool(threadCount);
		for (FutureTask<String> futureTask : futureTaskList) {
			execService.execute(futureTask);
		}
		
		new Thread(new InterruptRunnable(this, beginLatch)).start();

		beginLatch.countDown();

		Integer totalResult = Integer.valueOf(0);
		for (int i = 0; i < personCount; i++) {
			Integer partialResult = exchanger.exchange(Integer.valueOf(0));
			if(partialResult != 0){
				totalResult = totalResult + partialResult;
				System.out.println(String.format("Progress: %s/%s", totalResult, personCount));
			}
		}

		endLatch.await();
		System.out.println("--------------");
		for (FutureTask<String> futureTask : futureTaskList) {
			System.out.println(futureTask.get());
		}
		execService.shutdown();
	}

	public boolean isCanceled() {
		return canceled;
	}
	
	public void setCanceled(boolean canceled){
		this.canceled = canceled;
	}

	public static void main(String[] args) throws Exception {
		ConcurrentTaskExecutor executor = new ConcurrentTaskExecutor();
		executor.executeTask();
	}

}

 ExecuteCallable

public class ExecuteCallable implements Callable<String> {

	private int id;
	private CountDownLatch beginLatch;
	private CountDownLatch endLatch;
	private Exchanger<Integer> exchanger;
	private ConcurrentTaskExecutor concurrentTaskExecutor;

	public ExecuteCallable(CountDownLatch beginLatch, CountDownLatch endLatch,
			Exchanger<Integer> exchanger, int id,
			ConcurrentTaskExecutor concurrentTaskExecutor) {
		this.beginLatch = beginLatch;
		this.endLatch = endLatch;
		this.exchanger = exchanger;
		this.id = id;
		this.concurrentTaskExecutor = concurrentTaskExecutor;
	}

	@Override
	public String call() throws Exception {
		beginLatch.await();
		if(concurrentTaskExecutor.isCanceled()){
			endLatch.countDown();
			exchanger.exchange(0);
			return String.format("Player :%s is given up", id);
		}
		long millis = (long) (Math.random() * 10 * 1000);
		String result = String.format("Player :%s arrived, use %s millis", id, millis);
		Thread.sleep(millis);
		System.out.println(result);
		exchanger.exchange(1);
		endLatch.countDown();
		return result;
	}

}

 InterruptRunnable

public class InterruptRunnable implements Runnable {

	private CountDownLatch beginLatch;
	private ConcurrentTaskExecutor concurrentTaskExecutor;

	public InterruptRunnable(ConcurrentTaskExecutor currConcurrentTaskExecutor, CountDownLatch beginLatch) {
		this.beginLatch = beginLatch;
		this.concurrentTaskExecutor = currConcurrentTaskExecutor;
	}

	@Override
	public void run() {
		try {
			beginLatch.await();
			long millis = (long) (Math.random() * 10 * 1000);
			System.out.println(String.format("System need sleep %s millis", millis));
			Thread.sleep(millis);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		concurrentTaskExecutor.setCanceled(true);
	}

}

相關推薦

Java多執行緒簡單實現取消和進度

Java中簡單實現多執行緒排程時的可取消和顯示進度 一個簡單的多執行緒排程實現，統一開始，為了使得所有執行緒統一開始，類似運動員在聽到發令槍時一起進行，使用了CountDownLatch進行控制。 CountDownLatch beginLatch = new CountDownLatch(1); Cou

再次理解java多執行緒的實現 Thread 和Runnable的區別

Thread和Runnable的區別如果一個類繼承Thread，則不適合資源共享。但是如果實現了Runable介面的話，則很容易的實現資源共享。main函式，例項化執行緒物件也有所不同，extends Thread ：t.start();implements Runnable

Java多執行緒之—Synchronized方式和CAS方式實現執行緒安全效能對比

效能比較猜想 1.大膽假設 在設計試驗方法之前，針對Synchronized和CAS兩種方式的特點，我們先來思考一下兩種方式效率如何？ 首先，我們在回顧一下兩種方式是如何保證執行緒安全的。Synchronized方式通過大家應該很熟悉，他的行為非常悲觀，只要有一個執行緒進

Java多執行緒之 使用wait和notify實現生產者消費者模型

分享一下我老師大神的人工智慧教程！零基礎，通俗易懂！http://blog.csdn.net/jiangjunshow 也歡迎大家轉載本篇文章。分享知識，造福人民，實現我們中華民族偉大復興！        

java多執行緒簡單小例子2——實現Runnable介面

/** * 實現Runnable介面的類 * * @author */ public class DoSomething implements Runnable { private St

Java多執行緒中Synchronized簡介和Static Synchronized的區別

在進行Java開發時，多執行緒的開發是經常會使用的。首先會問一個小問題啊，在Java中有幾種方法可以建立一個執行緒？ 我給的答案是3種。（如果還有其他的請留言告訴我哈。） 1、建立直接繼承自Thread類建立執行緒子類。   步驟如下：a 定義一個子類，同時

JAVA多執行緒 重入鎖和讀寫鎖

在java多執行緒中，我們真的可以使用synchronized關鍵字來實現執行緒間的同步互斥工作，那麼其實還有一個更優秀的機制去完成這個“同步互斥”工作，他就是Lock物件，重入鎖和讀寫鎖。他們具有比synchronized更為強大的功能，並且有嗅探鎖定、多路分支等功能。 一、重入鎖

Java多執行緒-併發之sleep() 和 wait(n) 、 wait() 的區別

sleep() 和 wait(n) 、 wait() 的區別 答： sleep 方法：是 Thread 類的靜態方法，當前執行緒將睡眠 n 毫秒，執行緒進入阻塞狀態。當睡眠時間到了，會接觸阻塞，進入可執行狀態，等待 CPU 的到來。睡眠不釋放鎖（如果有的話） wai

java多執行緒：1程序和執行緒

多執行緒樣例 我們用工人卸貨舉例：有一集裝箱的貨物等待卸車，共100個箱子，一個工人一次只能搬一個箱子。 如果只有一個工人，那麼該工人需要搬運100次，而且是不停歇的搬運。 如果有5個或者10個工人，那麼平均每個工人只需要搬運20或者10次就可以了。 甚至有1

【Dr.Chen的系列問題】Java多執行緒的實現操作

一、什麼是多執行緒？ 多執行緒（英語：multithreading），是指從軟體或者硬體上實現多個執行緒併發執行的技術。具有多執行緒能力的計算機因有硬體支援而能夠在同一時間執行多於一個執行緒，進而提升整體處理效能。具有這種能力的系統包括對稱多處理機、多核心處理器以及晶片級多

Java多執行緒的實現（程序與執行緒的概念、Java繼承Thread類實現多執行緒、Java實現Runnable介面實現多執行緒、Thread與Runnable的區別、實現Callable介面實現多執行緒）

1 程序與執行緒 1.1 程序與執行緒的概念 什麼是程序？   程序： 作業系統中一個程式的執行週期。（比如我們想要在電腦上登入QQ，從雙擊qq按鈕---->關閉qq這個過程就是一個程序）   多程序： 同一時刻跑多個程式。   在DOS（磁碟作業系統時

Java多執行緒之 Lock介面和ReentrantLock的使用

                在多執行緒開發中，除了synchronized這個關鍵字外，我們還能通過Lock介面來實現這種效果。通過Lock介面來實現這種多執行緒加鎖效果的好處是非常的靈活，我們不在需要對整個函式加鎖，而且可以很方便的把他放在我們函式的任何一個地方，非常的稱心，而且從效率上來說，使用Loc

java多執行緒的實現方式總結

一個專案所用的技術往往是由業務所驅動的，平常我們寫的最多的是業務的CRUD，這是一個普遍的現象，要想往上提升，往往繞不開多執行緒知識，這篇文章是LZ對多執行緒實現的一些看法 多執行緒的三種實現方式 目前多執行緒的實現方式有三種，讓我帶您一一看來 繼承Thre

【Java程式設計】Java多執行緒的實現

多執行緒 程式：是一個指令的集合。 程序：正在執行中的程式，是一個靜態的概念。 執行緒：是程序中的一個單一的連續控制流程，執行緒又本稱為輕量級程序。 一個程序可擁有多個並行的執行緒，一個程序中的執行緒共享相同的記憶體單元，記憶體地址空間，可以訪問相同的變數和物件，而且他們從

Java多執行緒之建立任務和執行緒

Runnable建立執行緒 任務就是物件。為了建立任務，必須首先為任務定義一個類。任務類必須實現Runnable介面。Runnable介面非常簡單，它只包含一個run方法。需要實現這個方法來告訴系統執行緒將如何執行。開發一個任務類的模板如圖29-2a所示。

java多執行緒交替列印數字和字母

題目： 1.開啟兩個執行緒，一個執行緒列印A～Z，另一個執行緒列印1～52的資料。 2.實現交替列印，輸出結果為12A34B………..5152Z. 3.請用多執行緒方式實現。 使用阻塞佇列實現

java 多執行緒之future用法和意義

在併發程式設計時，一般使用runnable，然後扔給執行緒池完事，這種情況下不需要執行緒的結果。所以run的返回值是void型別。 如果是一個多執行緒協作程式，比如菲波拉切數列，1，1，2，3，5，8...使用多執行緒來計算。但後者需要前者的結果，就需要用callable介

QT5 多執行緒簡單實現

檔案中載入介面顯示的控制元件，變數都放在了threaddlg.h檔案中. threaddld.h原始碼如下： #ifndef THREADDLG_H #define THREADDLG_H #include <QDialog> #include <QPushButton> #i

Java多執行緒簡單樣例（一）：銀行存取錢問題

Bank類 public class Bank { private static int money; public int getMoney(){ return money; } public void saveMon

Java多執行緒之Executor框架和手寫簡易的執行緒池

目錄 Java多執行緒之一執行緒及其基本使用 Java多執行緒之二(Synchronized) Java多執行緒之三volatile與等待通知機制示例 執行緒池 什麼是執行緒池 執行緒池一種執行緒使用模式，執行緒池會維護多個執行緒，等待著分配可併發執行的任務，當有任務需要執行緒執行時