我有一個有 8 個執行緒的執行緒池
private static final ExecutorService SERVICE = Executors.newFixedThreadPool(8);
我的機制模擬 100 個用戶（100 個任務）的作業：
List<Callable<Boolean>> callableTasks = new ArrayList<>();
for (int i = 0; i < 100; i ) { // Number of users == 100
callableTasks.add(new Task(client));
}
SERVICE.invokeAll(callableTasks);
SERVICE.shutdown();
用戶執行生成檔案的任務。
- 獲取任務的 UUID；
- 每 10 秒獲取一次任務狀態；
- 如果任務準備好獲取檔案。
public class Task implements Callable<Boolean> {
private final ReportClient client;
public Task(ReportClient client) {
this.client = client;
}
@Override
public Boolean call() {
final var uuid = client.createDocument(documentId);
GetStatusResponse status = null;
do {
try {
Thread.sleep(10000); // This stop current thread, but not a Task!!!!
} catch (InterruptedException e) {
return Boolean.FALSE;
}
status = client.getStatus(uuid);
} while (Status.PENDING.equals(status.status()));
final var document = client.getReport(uuid);
return Boolean.TRUE;
}
}
我想將空閑時間（10 秒）交給另一個任務。但是當命令Thread.sleep(10000); 被呼叫時，當前執行緒暫停其執行。前 8 個任務暫停，92 個任務等待 10 秒。如何同時完成 100 個任務？
uj5u.com熱心網友回復：
關于今天的 Java ，Yevgeniy的答案看起來是正確的。你想吃蛋糕也想吃，因為你想讓執行緒在重復任務之前休眠，但你也希望那個執行緒做其他作業。這在今天是不可能的，但將來可能會。
專案織機
在當前的 Java 中，Java 執行緒直接映射到主機OS執行緒。在所有常見的作業系統中，如 macOS、BSD、Linux、Windows 等，當在主機執行緒中執行的代碼阻塞（停止等待睡眠、存盤 I/O 或網路 I/O 等）時，執行緒也會阻塞塊。被阻塞的執行緒掛起，主機作業系統通常在該未使用的內核上運行另一個執行緒。但關鍵點是掛起的執行緒在您對 sleep 的阻塞呼叫回傳之前不會執行進一步的作業。
在不遠的將來，這種情況可能會發生變化。Project Loom試圖將虛擬執行緒添加到 Java 的并發設施中。
在這項新技術中，許多 Java 虛擬執行緒被映射到每個主機作業系統執行緒。處理許多 Java 虛擬執行緒由 JVM 而非作業系統管理。當 JVM 檢測到虛擬執行緒的執行代碼被阻塞時，該虛擬執行緒被“停放”，被 JVM 擱置一旁，另一個虛擬執行緒被換出以在該“真實”主機作業系統執行緒上執行。當另一個執行緒從它的阻塞呼叫中回傳時，它可以被重新分配給一個“真正的”主機作業系統執行緒以供進一步執行。在Project Loom下，主機作業系統執行緒保持忙碌狀態，在任何掛起的虛擬執行緒有作業要做時從不空閑。
虛擬執行緒之間的這種交換非常有效，因此可以在傳統計算機硬體上一次運行數千甚至數百萬個執行緒。
使用虛擬執行緒，您的代碼確實會像您希望的那樣作業：Java 中的阻塞呼叫不會阻塞主機作業系統執行緒。但是虛擬執行緒是實驗性的，仍在開發中，計劃作為Java 19中的預覽功能。包含 Loom 技術的 Java 19 的早期訪問版本現在可供您試用。但是對于今天的生產部署，您需要遵循 Yevgeniy 的答案中的建議。
因為我不是并發方面的專家，所以在這里我的報道要持保留態度。您可以從真正的專家那里聽到，在Project Loom團隊成員（包括 Ron Pressler 和 Alan Bateman）的文章、采訪和演講中。
uj5u.com熱心網友回復：
編輯：我剛剛發布了這個答案，并意識到您似乎正在使用該代碼來模擬與某些系統的真實用戶互動。我強烈建議為此使用負載測驗實用程式，而不是嘗試自己開發。但是，在這種情況下，僅使用 CachedThreadPool 就可以解決問題，盡管可能不是一個非常健壯或可擴展的解決方案。
此處的 Thread.sleep() 行為按預期作業：它暫停執行緒以讓 CPU 執行其他執行緒。請注意，在這種狀態下，執行緒可能會因與您的代碼無關的多種原因而被中斷，在這種情況下，您的 Task 回傳false：我假設您實際上有一些重試邏輯。
所以你想要兩個相互排斥的東西：一方面，如果檔案沒有準備好，執行緒應該可以自由地做其他事情，但應該以某種方式回傳并在 10 秒內再次檢查該檔案的狀態。
這意味著您必須選擇：
您肯定需要每 10 秒檢查一次每個檔案 - 在這種情況下，可能使用 cachedThreadPool 并讓它根據需要生成盡可能多的執行緒，請記住，您將承擔大量執行緒的開銷虛擬執行沒有什么。
或者，您可以首先啟動該異步檔案創建程序，然后僅檢查可呼叫物件中的狀態，并根據需要重試。
就像是：
public class Task implements Callable<Boolean> {
private final ReportClient client;
private final UUID uuid;
// all args constructor omitted for brevity
@Override
public Boolean call() {
GetStatusResponse status = client.getStatus(uuid);
if (Status.PENDING.equals(status.status())) {
final var document = client.getReport(uuid);
return Boolean.TRUE;
} else {
return Boolean.FALSE; //retry next time
}
}
}
List<Callable<Boolean>> callableTasks = new ArrayList<>();
for (int i = 0; i < 100; i ) {
var uuid = client.createDocument(documentId); //not sure where documentId comes from here in your code
callableTasks.add(new Task(client, uuid));
}
List<Future<Boolean>> results = SERVICE.invokeAll(callableTasks);
// retry logic until all results come back as `true` here
這假設這createDocument是相對有效的，但該階段也可以并行化，您只需要使用單獨的Runnable任務串列并使用執行器服務呼叫它們。請注意，我們還假設檔案的狀態最終確實會更改為 PENDING 以外的狀態，而且很可能并非如此。您可能希望重試超時。
uj5u.com熱心網友回復：
在您的情況下，您似乎需要每 x 秒檢查一次是否滿足某個條件。事實上，從您的代碼來看，檔案生成似乎是異步的，此后 Task 一直在做的只是等待檔案生成發生。
您可以從 Thread-Main 啟動每個檔案生成，并使用 aScheduledThreadPoolExecutor每 x 秒驗證檔案生成是否已完成。此時，您檢索結果并取消相應任務的調度。
基本上，一個ConcurrentHashMap在 thread-main 和您安排的任務 (mapRes) 之間共享，而另一個 mapTask 僅在 thread-main 中本地使用，以跟蹤 eachScheduledFuture回傳的Task。
public class Main {
public static void main(String[] args) {
ScheduledThreadPoolExecutor pool = (ScheduledThreadPoolExecutor) Executors.newScheduledThreadPool(8);
//ConcurrentHashMap shared among the submitted tasks where each Task updates its corresponding outcome to true as soon as the document has been produced
ConcurrentHashMap<Integer, Boolean> mapRes = new ConcurrentHashMap<>();
for (int i = 0; i < 100; i ) {
mapRes.put(i, false);
}
String uuid;
ScheduledFuture<?> schedFut;
//HashMap containing the ScheduledFuture returned by scheduling each Task to cancel their repetition as soon as the document has been produced
Map<String, ScheduledFuture<?>> mapTask = new HashMap<>();
for (int i = 0; i < 100; i ) {
//Starting the document generation from the thread-main
uuid = client.createDocument(documentId);
//Scheduling each Task 10 seconds apart from one another and with an initial delay of i*10 to not start all of them at the same time
schedFut = pool.scheduleWithFixedDelay(new Task(client, uuid, mapRes), i * 10, 10000, TimeUnit.MILLISECONDS);
//Adding the ScheduledFuture to the map
mapTask.put(uuid, schedFut);
}
//Keep checking the outcome of each task until all of them have been canceled due to completion
while (!mapTask.values().stream().allMatch(v -> v.isCancelled())) {
for (String key : mapTask.keySet()) {
//If the Task has set the outcome to true then its scheduling is canceled
if (mapRes.get(key)) {
schedFut = mapTask.get(key);
schedFut.cancel(true);
}
}
}
pool.shutdown();
}
}
class Task implements Runnable {
private final ReportClient client;
private final String uuid;
private final ConcurrentHashMap mapRes;
public Task(ReportClient client, String uuid, ConcurrentHashMap mapRes) {
this.client = client;
this.uuid = uuid;
this.mapRes = mapRes;
}
@Override
public void run() {
//This is taken form your code and I'm assuming that if it's not pending then it's completed
if (!Status.PENDING.equals(client.getStatus(uuid).status())) {
mapRes.replace(uuid, true);
}
}
}
我已經在本地測驗了您的案例，通過模擬一個場景，其中 nTasks等待創建具有相同 id 的檔案夾（或在您的案例中為 uuid）。如果您想先嘗試一些更簡單的東西，我會將它作為示例發布在這里。
public class Main {
public static void main(String[] args) {
ScheduledThreadPoolExecutor pool = (ScheduledThreadPoolExecutor) Executors.newScheduledThreadPool(2);
ConcurrentHashMap<Integer, Boolean> mapRes = new ConcurrentHashMap<>();
for (int i = 0; i < 16; i ) {
mapRes.put(i, false);
}
ScheduledFuture<?> schedFut;
Map<Integer, ScheduledFuture<?>> mapTasks = new HashMap<>();
for (int i = 0; i < 16; i ) {
schedFut = pool.scheduleWithFixedDelay(new MyTask(i, mapRes), i * 20, 3000, TimeUnit.MILLISECONDS);
mapTasks.put(i, schedFut);
}
while (!mapTasks.values().stream().allMatch(v -> v.isCancelled())) {
for (Integer key : mapTasks.keySet()) {
if (mapRes.get(key)) {
schedFut = mapTasks.get(key);
schedFut.cancel(true);
}
}
}
pool.shutdown();
}
}
class MyTask implements Runnable {
private int num;
private ConcurrentHashMap mapRes;
public MyTask(int num, ConcurrentHashMap mapRes) {
this.num = num;
this.mapRes = mapRes;
}
@Override
public void run() {
System.out.println("Task " num " is checking whether the folder exists: " Files.exists(Path.of("./" num)));
if (Files.exists(Path.of("./" num))) {
mapRes.replace(num, true);
}
}
}
轉載請註明出處，本文鏈接：https://www.uj5u.com/caozuo/482815.html
上一篇：在無限回圈中獨立運行3個異步任務