自定義執行緒池
package com.appletree24;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
class Main {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ThreadPool threadPool = new ThreadPool(2, 1000, TimeUnit.MICROSECONDS, 5, (queue, task) -> {
//帶超時等待
// queue.offer(task,500,TimeUnit.MILLISECONDS);
});
for (int i = 0; i < 10; i++) {
int j = i;
threadPool.execute(() -> {
try {
Thread.sleep(1000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(j);
});
}
}
}
//策略模式介面 此處使用策略模式是因為在實作拒絕策略時，有許多種拒絕的方式，這些方式如果不使用恰當的模式，就需要大量的if..else來撰寫
//且方式數量大于4個，會造成類膨脹的問題，推薦使用混合模式
//https://www.runoob.com/design-pattern/strategy-pattern.html
@FunctionalInterface
interface RejectPolicy<T> {
void reject(BlockingQueue<T> queue, T task);
}
class ThreadPool {
//任務佇列
private BlockingQueue<Runnable> taskQueue;
//執行緒集合
private HashSet<Worker> workers = new HashSet<>();
//執行緒數
private int coreSize;
//超時時間
private long timeout;
private TimeUnit timeUnit;
private RejectPolicy<Runnable> rejectPolicy;
//執行任務
public void execute(Runnable task) {
//當任務數未超過核心執行緒數時，直接交給Worker物件執行
//如果超過，則加入阻塞任務佇列，暫存起來
synchronized (workers) {
if (workers.size() < coreSize) {
Worker worker = new Worker(task);
workers.add(worker);
worker.start();
} else {
//第一種選擇死等
// taskQueue.put(task);
//第二種為超時等待
//第三種為消費者放棄任務執行
//第四種為主執行緒拋出例外
//第五種讓呼叫者自行執行任務
taskQueue.tryPut(rejectPolicy, task);
}
}
}
public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapcity, RejectPolicy<Runnable> rejectPolicy) {
this.coreSize = coreSize;
this.timeout = timeout;
this.timeUnit = timeUnit;
this.taskQueue = new BlockingQueue<>(queueCapcity);
this.rejectPolicy = rejectPolicy;
}
class Worker extends Thread {
private Runnable task;
public Worker(Runnable task) {
this.task = task;
}
@Override
public void run() {
//執行任務
//1.當傳遞過來的task不為空，執行任務
//2.當task執行完畢，再接著取下一個任務并執行
while (task != null || (task = taskQueue.poll(1000, TimeUnit.MILLISECONDS)) != null) {
try {
task.run();
} catch (Exception e) {
e.printStackTrace();
} finally {
task = null;
}
}
synchronized (workers) {
workers.remove(this);
}
}
}
}
class BlockingQueue<T> {
//1. 任務佇列
private final Deque<T> queue = new ArrayDeque<>();
//2. 鎖
private final ReentrantLock lock = new ReentrantLock();
//3. 生產者條件變數
private final Condition fullWaitSet = lock.newCondition();
//4. 消費者條件變數
private final Condition emptyWaitSet = lock.newCondition();
//5. 容量上限
private int capcity;
public BlockingQueue(int capcity) {
this.capcity = capcity;
}
//帶超時的等待獲取
public T poll(long timeout, TimeUnit unit) {
lock.lock();
long nanos = unit.toNanos(timeout);
try {
while (queue.isEmpty()) {
try {
if (nanos <= 0) {
return null;
}
nanos = emptyWaitSet.awaitNanos(nanos);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
T t = queue.removeFirst();
fullWaitSet.signal();
return t;
} finally {
lock.unlock();
}
}
//消費者拿取任務的方法
public T take() {
lock.lock();
try {
while (queue.isEmpty()) {
try {
emptyWaitSet.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
T t = queue.removeFirst();
fullWaitSet.signal();
return t;
} finally {
lock.unlock();
}
}
//阻塞添加
public void put(T task) {
lock.lock();
try {
while (queue.size() == capcity) {
try {
fullWaitSet.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
queue.offerLast(task);
//添加完后喚醒消費者等待
emptyWaitSet.signal();
} finally {
lock.unlock();
}
}
//帶超時時間的阻塞添加
public boolean offer(T task, long timeout, TimeUnit unit) {
lock.lock();
try {
long nanos = unit.toNanos(timeout);
while (queue.size() == capcity) {
try {
if (nanos <= 0) return false;
nanos = fullWaitSet.awaitNanos(nanos);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
queue.offerLast(task);
//添加完后喚醒消費者等待
emptyWaitSet.signal();
return true;
} finally {
lock.unlock();
}
}
//獲取當前阻塞佇列大小
public int size() {
lock.lock();
try {
return queue.size();
} finally {
lock.unlock();
}
}
public void tryPut(RejectPolicy<T> rejectPolicy, T task) {
lock.lock();
try {
//判斷佇列是否已滿
if (queue.size() == capcity) {
rejectPolicy.reject(this, task);
} else {
queue.addLast(task);
emptyWaitSet.signal();
}
} finally {
lock.unlock();
}
}
}
上述的自定義執行緒池雖然能夠執行完畢主執行緒給予的任務，但任務全部執行結束后，開辟的執行緒池內核心執行緒仍然在運行，并沒有結束，這是因為目前執行緒池中的take方法仍然為不會有超時等待的take方法，造成了死等，需要為其加入超時停止的功能，也就是替代take()的poll()
JDK自帶執行緒池
介紹
ThreadPoolExecutor使用int的高三位表示執行緒池狀態，低29位表示執行緒數量
在ThreadPoolExecutor中，同樣也存在拒絕策略，其圖結構如下：
其中介面就對應著在自定義執行緒池中實作的策略模式介面，下面的四個實作類就對應著四種不同的拒絕方式：
利用工具類創建固定大小執行緒池
利用工具類創建帶緩沖的執行緒池
從原始碼可以看出，帶緩沖的執行緒池中緩沖佇列的使用的是一個名為SynchronousQueue的佇列，這個佇列的特點如下：佇列不具有容量，當沒有執行緒來取時，是無法對其內部放入資料的，例如佇列內部已有一個數字1，但此時沒有執行緒取走，則線此佇列目前并不能繼續存入資料，直到1被取走
利用工具類創建單執行緒執行緒池
從原始碼可以看出，單執行緒執行緒池中核心執行緒數與最大執行緒數相等，即不存在應急執行緒，只能解決一個任務
那么這個執行緒池和我自己創建一個執行緒的執行緒池有什么區別呢？區別如下：
ThreadPoolExecutor-submit method
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService pool = Executors.newFixedThreadPool(2);
Future<String> result = pool.submit(() -> {
System.out.println("running");
Thread.sleep(1000);
return "ok";
});
System.out.println(result.get());
}
submit方法可以傳入Runnable和Callable型別的引數，并且將執行緒內部所執行任務的結果回傳，用Future包裝
ThreadPoolExecutor-invokeAll
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService pool = Executors.newFixedThreadPool(2);
List<Future<String>> results = pool.invokeAll(Arrays.asList(() -> {
System.out.println("begin");
Thread.sleep(1000);
return "1";
},
() -> {
System.out.println("begin");
Thread.sleep(500);
return "2";
}));
results.forEach(f -> {
try {
System.out.printf(f.get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
});
}
invokeAll方法可以傳入任務的集合，同樣的任務的回傳值也會以串列形式回傳
ThreadPoolExecutor-invokeAny
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService pool = Executors.newFixedThreadPool(2);
String result = pool.invokeAny(Arrays.asList(() -> {
System.out.println("begin");
Thread.sleep(1000);
return "1";
},
() -> {
System.out.println("begin");
Thread.sleep(500);
return "2";
}));
pool.awaitTermination(1000, TimeUnit.MILLISECONDS);
System.out.println(result);
}
invokeAny方法同樣可以傳入任務的集合，只不過最后回傳的結果并不是任務的結果集合，而是最早完成的那個任務的結果，
ThreadPoolExecutor-shutdown
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService pool = Executors.newFixedThreadPool(2);
List<Future<String>> results = pool.invokeAll(Arrays.asList(() -> {
System.out.println("begin");
Thread.sleep(1000);
return "1";
},
() -> {
System.out.println("begin");
Thread.sleep(500);
return "2";
}));
pool.shutdown();
results.forEach(f -> {
try {
System.out.println(f.get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
});
}
shutdown方法會將執行緒池的狀態變為SHUTDOWN
- 不會接受新任務
- 但已提交的任務會執行完
- 此方法不會阻塞呼叫執行緒的執行
ThreadPoolExecutor-shutdownNow
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService pool = Executors.newFixedThreadPool(2);
List<Future<String>> results = pool.invokeAll(Arrays.asList(() -> {
System.out.println("begin");
Thread.sleep(1000);
return "1";
},
() -> {
System.out.println("begin");
Thread.sleep(500);
return "2";
}));
List<Runnable> runnables = pool.shutdownNow();
results.forEach(f -> {
try {
System.out.println(f.get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
});
}
shutdownNow方法會將執行緒池狀態變為STOP
- 不會接受新任務
- 會將佇列中現有的任務回傳
- 并且用interrupt方法中斷正在執行的任務
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/544559.html
標籤：Java
上一篇：Java＋Jquer實作趨勢圖
下一篇：RabbitMQ中間件