疑問
- 多個線程執(zhí)行結束后怎么執(zhí)行某一特定操作？
- 怎么限制執(zhí)行某塊業(yè)務的線程的數(shù)量炭菌？
CountDownLatch
計數(shù)器鎖罪佳，初始化一個 count（數(shù)）鎖，每個業(yè)務線程依次 countDown （遞減）黑低，主線程阻塞 await （等待）直至 count 等于 0赘艳，或者指定 await 時間：
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
/**
* @author caojiantao
*/
public class Test {
public static void main(String[] args) {
System.out.println("========================= 開始 =========================");
// 工作線程數(shù)量
int workCount = 5;
// 模擬耗時范圍
int cost = 3000;
CountDownLatch latch = new CountDownLatch(workCount);
for (int i = 0; i < workCount; i++) {
new Thread(() -> {
System.out.println(getTimeFmtString() + " " + Thread.currentThread().getName() + " 開始執(zhí)行...");
try {
TimeUnit.MILLISECONDS.sleep(new Random().nextInt(cost));
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(getTimeFmtString() + " " + Thread.currentThread().getName() + " 執(zhí)行完...");
latch.countDown();
}).start();
}
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("========================= 結束 =========================");
}
private static String getTimeFmtString(){
return LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_TIME);
}
}
CyclicBarrier
籬柵，功能與 CountDownLatch 大體相同克握，但是其特色支持計數(shù)器重置蕾管，循環(huán)使用：
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
/**
* @author caojiantao
*/
public class Test {
public static void main(String[] args) {
System.out.println("========================= 開始 =========================");
// 工作線程數(shù)量
int workCount = 5;
// 模擬耗時范圍
int cost = 3000;
CyclicBarrier barrier = new CyclicBarrier(workCount, () -> {
System.out.println("========================= 結束 =========================");
});
for (int i = 0; i < workCount; i++) {
new Thread(() -> {
try {
System.out.println(getTimeFmtString() + " " + Thread.currentThread().getName() + " 開始執(zhí)行...");
TimeUnit.MILLISECONDS.sleep(new Random().nextInt(cost));
System.out.println(getTimeFmtString() + " " + Thread.currentThread().getName() + " 執(zhí)行完...");
barrier.await();
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
}).start();
}
}
private static String getTimeFmtString() {
return LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_TIME);
}
}
注：相比較 CountDownLatch，CyclicBarrier 能夠 reset 重置計數(shù)器菩暗，同時注意只有 await 方法會阻塞當前線程掰曾，countDown 并不會。
Semaphore
信號量停团，保持當前信號量（執(zhí)行線程數(shù)量）最多為 permits：
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Random;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
/**
* @author caojiantao
*/
public class Test {
public static void main(String[] args) {
// 工作線程數(shù)量
int workCount = 10;
// 模擬耗時范圍
int cost = 3000;
Semaphore semaphore = new Semaphore(5);
for (int i = 0; i < workCount; i++) {
new Thread(() -> {
try {
// 申請執(zhí)行權限
semaphore.acquire();
System.out.println(getTimeFmtString() + " " + Thread.currentThread().getName() + " 開始執(zhí)行...");
TimeUnit.MILLISECONDS.sleep(new Random().nextInt(cost));
System.out.println(getTimeFmtString() + " " + Thread.currentThread().getName() + " 執(zhí)行完...");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
// 釋放權限
semaphore.release();
}
}).start();
}
}
private static String getTimeFmtString() {
return LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_TIME);
}
}
案例
三方平臺比價接口
假如存在 A旷坦、B、C 三個平臺正在出售某商品 goods佑稠，現(xiàn)在需要多線程獲取三個平臺該商品 goods 價格秒梅，最終輸出最低價格價格信息。
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
/**
* @author caojiantao
*/
public class Test {
private static Map<String, Double> infoMap = new ConcurrentHashMap<>(3);
public static void main(String[] args) {
CountDownLatch latch = new CountDownLatch(3);
Thread a = new Thread(new Task("A", latch));
Thread b = new Thread(new Task("B", latch));
Thread c = new Thread(new Task("C", latch));
a.start();
b.start();
c.start();
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
Map.Entry<String, Double> lowest = null;
for (Map.Entry<String, Double> entry : infoMap.entrySet()) {
if (lowest == null || entry.getValue() < lowest.getValue()) {
lowest = entry;
}
}
assert lowest != null;
System.out.println(getTimeFmtString() + " 最低價格信息為：" + lowest.getKey() + " " + lowest.getValue());
}
private static String getTimeFmtString() {
return LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_TIME);
}
static class Task implements Runnable {
private String name;
private CountDownLatch latch;
public Task(String name, CountDownLatch latch) {
this.name = name;
this.latch = latch;
}
@Override
public void run() {
int cost = 5000;
try {
TimeUnit.MILLISECONDS.sleep(new Random().nextInt(cost));
double price = new BigDecimal(new Random().nextDouble() * 1000).setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();
infoMap.put(name, price);
System.out.println(getTimeFmtString() + " " + name + " 報價：" + price);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
latch.countDown();
}
}
}
}
限制接口訪問次數(shù)
存在某接口 queryData舌胶，需要控制在 10 個訪問數(shù)以內捆蜀，超過的請求阻塞直至有正在執(zhí)行的請求已經完成。
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Random;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
/**
* @author caojiantao
*/
public class Service {
private Semaphore semaphore = new Semaphore(10);
public void queryData() {
// 模擬耗時范圍
int cost = 3000;
try {
// 申請執(zhí)行權限
semaphore.acquire();
System.out.println(getTimeFmtString() + " " + Thread.currentThread().getName() + " 請求...");
TimeUnit.MILLISECONDS.sleep(new Random().nextInt(cost));
System.out.println(getTimeFmtString() + " " + Thread.currentThread().getName() + " 結束");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
// 釋放權限
semaphore.release();
}
}
private String getTimeFmtString() {
return LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_TIME);
}
}
