1Chủ đềPool

package concurrent.threadpool; /** * @className: ThreadPool * @description: Giao diện nhóm luồng* @date: 2022/4/2 * @author: cakin */ giao diện công khai ThreadPool { // Gửi tác vụ tới nhóm luồng void thực thi ( Runnable runnable); // Tắt nhóm luồng voidshutout(); // Lấy kích thước ban đầu của nhóm luồng int getInitSize(); // Lấy số lượng luồng tối đa trong nhóm luồng int getMaxSize(); // Lấy số lượng luồng lõi trong nhóm luồng int getCoreSize(); // Lấy kích thước của hàng đợi tác vụ trong nhóm luồng int getQueueSize(); // Lấy số lượng luồng đang hoạt động trong luồng pool int getActiveCount(); // Kiểm tra xem thread pool đã bị tắt hay chưa boolean isShutdown();

2 Hàng đợi có thể chạy được

gói concurrent.threadpool; /** * @className: RunnableQueue * @description: Hàng đợi tác vụ, chủ yếu được sử dụng để lưu vào bộ nhớ đệm các tác vụ được gửi tới nhóm luồng * @date: 2022/4/2 * @author: cakin */ giao diện công khai RunnableQueue { // Khi một tác vụ mới xuất hiện, trước tiên nó sẽ được cung cấp cho hàng đợi void Offer(Runnable runnable); // Luồng công việc lấy Runnable thông qua phương thức take Runnable take() ném InterruptedException; //ném InterruptedException; // Lấy số lượng tác vụ trong hàng đợi tác vụ int size();

Ba Chủ ĐềNhà Máy

package concurrent.threadpool; /** * @className: ThreadFactory * @description: Thread Factory, cung cấp giao diện để tạo các chủ đề để tạo điều kiện cho việc tùy chỉnh Chủ đề được cá nhân hóa, chẳng hạn như Nhóm Chủ đề nên được đặt vào, mức độ ưu tiên, tên chủ đề, và Liệu đó có phải là một luồng daemon hay không, v.v. * @date: 2022/4/2 * @author: cakin */ @FunctionalInterface giao diện công khai ThreadFactory { // Được sử dụng để tạo luồng Thread createThread(Có thể chạy được);

4. Chính sách từ chối

package concurrent.threadpool; /** * @className: DenyPolicy * @description: Chiến lược từ chối: Chủ yếu được sử dụng để quyết định sử dụng chiến lược nào để thông báo cho người gửi khi Runnable trong Hàng đợi đạt đến giới hạn trên. Giao diện này xác định ba cách triển khai mặc định * @date: 2022/4/2 * @author: cakin */ @FunctionalInterface public giao diện DenyPolicy { void chối(Runnable runnable, ThreadPool threadPool); // Chính sách từ chối này sẽ trực tiếp loại bỏ lớp nhiệm vụ DiscardDenyPolicy; triển khai DenyPolicy { @Override public void từ chối(Runnable runnable, ThreadPool threadPool) { // không làm gì } } // Chính sách từ chối này sẽ đưa ra một ngoại lệ cho lớp người gửi tác vụ AbortDenyPolicy triển khai DenyPolicy { @Override public void từ chối(Runnable runnable, ThreadPool threadPool) { Throw new RuntimeDenyException("Runnable " + runnable + " sẽ bị hủy."); } } // Chính sách từ chối này sẽ khiến tác vụ được thực thi trong luồng nơi người gửi được đặt class RunnerDenyPolicy triển khai DenyPolicy { @Override public void từ chối(Runnable runnable, ThreadPool threadPool) { if (!threadPool.isShutdown()) { runnable.run();

5RuntimeDenyNgoại lệ

package concurrent.threadpool; /** * @className: RuntimeDenyException * @description: Lớp ngoại lệ: Thông báo cho người gửi tác vụ rằng hàng đợi tác vụ không còn có thể chấp nhận tác vụ mới * @date: 2022/4/2 * @author: cakin */ lớp công khai RuntimeDenyException mở rộng RuntimeException { public RuntimeDenyException(String message) { super(message } }

SáuNhiệm vụ nội bộ

gói concurrent.threadpool; lớp công khai InternalTask ​​​​thực hiện Runnable { end RunnableQueue runnableQueue; boolean biến động riêng tư chạy = true; public InternalTask(RunnableQueue runnableQueue) { this.runnableQueue = runnableQueue } @Override public void run() { // If tác vụ hiện tại Nếu nó đang chạy và không bị gián đoạn, nó sẽ tiếp tục lấy Runnable từ hàng đợi rồi thực thi run Phương thức while (running && !Thread.currentThread().isInterrupted()) { try { Runnable task = runnableQueue.take(); task.run(); } Catch (Ngoại lệ e) { Running = false; // Để dừng tác vụ hiện tại, public void stop() { this.running = false; } } chủ yếu được sử dụng trong phương thức tắt của nhóm luồng.

7Liên kếtRunnableQueue

package concurrent.threadpool; import java.util.LinkedList; public class LinkedRunnableQueue triển khai RunnableQueue { // Dung lượng tối đa của hàng đợi tác vụ, được chuyển vào trong quá trình xây dựng giới hạn int cuối cùng riêng tư; // Nếu các tác vụ trong hàng đợi tác vụ đã đầy, thì nó cần được thực thi Chính sách từ chối riêng tư cuối cùng DenyPolicy từ chốiPolicy // Hàng đợi để lưu trữ các tác vụ riêng tư cuối cùng LinkedList runnableList = new LinkedList<>(); Nhóm luồng riêng cuối cùng ThreadPool threadPool; //Constructor public LinkedRunnableQueue(int limit, DenyPolicy denPolicy, ThreadPool threadPool) { this.limit = limit; this.denyPolicy = denPolicy; threadPool = threadPool } @Override public void Offer(Runnable runnable; ) { được đồng bộ hóa (runnableList) { if (runnableList.size() >= limit) { // Thực thi chính sách từ chối khi không thể cung cấp các tác vụ mới denPolicy.reject(runnable, threadPool); } else { // Thêm tác vụ vào cuối hàng đợi và đánh thức luồng bị chặn runnableList.addLast( runnable); runnableList.notifyAll(); } } } @Override public Runnable take() ném InterruptedException { được đồng bộ hóa (runnableList) { while (runnableList.isEmpty()) { try { // Nếu không có tác vụ nào trong hàng đợi tác vụ để thực thi, luồng hiện tại sẽ treo và nhập bộ chờ màn hình được liên kết với runnableList để chờ đánh thức (tác vụ mới được thêm) runnableList .wait(); } Catch (InterruptedException e) { Throw e; } } // Xóa một tác vụ khỏi phần đầu của hàng đợi tác vụ return runnableList.removeFirst() } } @Override public int size() { được đồng bộ hóa (runnableList) { // Trả về số lượng tác vụ trong hàng đợi tác vụ hiện tại return runnableList.size();

8Chủ đề cơ bảnPool

gói concurrent.threadpool; nhập java.util.ArrayDeque; nhập java.util.Queue; nhập java.util.concurrent.TimeUnit; nhập java.util.concurrent.atomic.AtomicInteger; Số luồng riêng tư cuối cùng int initSize; // Số lượng luồng tối đa trong nhóm luồng riêng tư cuối cùng int maxSize // Số lượng luồng lõi trong nhóm luồng riêng tư; cuối cùng int coreSize; // Số lượng chủ đề đang hoạt động hiện tại riêng tư int activeCount; // Nhà máy được yêu cầu tạo chủ đề riêng tư cuối cùng ThreadFactory threadFactory; // Hàng đợi tác vụ riêng tư cuối cùng RunnableQueue runnableQueue; boolean isShutdown = false; // Hàng đợi luồng công nhân, lưu trữ các luồng đang hoạt động riêng tư Final Queue threadQueue = new ArrayDeque<>(); Từ chối chính sách riêng tư cuối cùng tĩnh DenyPolicy DEFAULT_DENY_POLICY = new DenyPolicy.DiscardDenyPolicy(); // Thread Factory riêng tư cuối cùng tĩnh ThreadFactory DEFAULT_THREAD_FACTORY = new DefaultThreadFactory(); Các tham số cần được chuyển khi xây dựng hàng đợi: số luồng ban đầu, số luồng tối đa, số luồng lõi, số lượng hàng đợi tác vụ tối đa public BasicThreadPool(int initSize, int maxSize, int coreSize, int queueSize) { this(initSize , maxSize, coreSize , DEFAULT_THREAD_FACTORY, queueSize, DEFAULT_DENY_POLICY, 10, TimeUnit.SECONDS } public BasicThreadPool(int initSize, int maxSize, int coreSize, ThreadFactory threadFactory, int queueSize, DenyPolicy denPolicy, int keepAliveTime, TimeUnit timeUnit) { this.initSize = this.maxSize = maxSize; threadFactory; this.runnableQueue = new LinkedRunnableQueue(queueSize, denPolicy, this); this.keepAliveTime = keepAliveTime; this.timeUnit = timeUnit() } // Trong quá trình khởi tạo, trước tiên hãy tạo các luồng initSize riêng tư void init() { / / Bắt đầu chuỗi riêng của BasicThreadPool start(); for (int i = 0; i < initSize; i++) { newThread(); } } riêng void newThread() { // Tạo một chuỗi tác vụ và bắt đầu InternalTask ​​​internalTask ​​​= new InternalTask(runnableQueue); = ThreadTask mới( thread, InternalTask); threadQueue.offer(threadTask); thread.start(); } private void deleteThread() { // Xóa một luồng khỏi nhóm luồng ThreadTask threadTask = threadQueue.remove(); threadTask.internalTask.stop(); runnable vào runnableQueue @Override public void exec(Runnable runnable) { if (isShutdown) { ném IllegalStateException mới ("nhóm luồng là destroy"); } // Gửi một tác vụ chỉ cần chèn hàng đợi tác vụ vào this.runnableQueue.offer(runnable); } // Để dừng luồng BasicThreadPool, hãy dừng các luồng đang hoạt động trong nhóm luồng và thay đổi biến chuyển đổi isShutdown thành true @Override public void tắt máy() { được đồng bộ hóa (điều này) { if (isShutdown) { return; } isShutdown = true; threadQueue.forEach(threadTask -> { threadTask.internalTask.stop(); threadTask.thread.interrupt(); }); this.interrupt(); } } @Override public int getInitSize() { if (isShutdown) { ném mới IllegalStateException("nhóm luồng bị hủy"); } trả về cái này.initSize; } @Override public int getMaxSize() { if (isShutdown) { Throw new IllegalStateException("the thread pool is destroy"); } return this.maxSize; } @Override public int getCoreSize() { if (isShutdown) { Throw IllegalStateException mới ("nhóm luồng bị hủy"); } return this.coreSize } @Override public int; getQueueSize() { if (isShutdown) { ném IllegalStateException mới ("nhóm luồng bị hủy"); } return runnableQueue.size(); } @Override public int getActiveCount() { được đồng bộ hóa (this){ return this.activeCount } } @Ghi đè boolean công khai isShutdown() { return this.isShutdown; } // Chủ yếu được sử dụng để duy trì số lượng luồng, chẳng hạn như mở rộng, tái chế, v.v. @Override public void run() { while (!isShutdown && !isInterrupted()) { try { timeUnit.sleep(keepAliveTime } Catch (InterruptedException e); ) { isShutdown = true; break; } được đồng bộ hóa (this) { if (isShutdown) { break; } // Có các tác vụ trong hàng đợi hiện tại chưa được xử lý và activeCount < coreSize Sau đó tiếp tục mở rộng if (runnableQueue.size() > 0 && activeCount < coreSize) { for (int i = initSize; i < coreSize; i++) { newThread() } // Lý do cho countinue là vì chúng tôi không muốn việc mở rộng luồng bắt đầu trực tiếp tới maxSize tiếp tục; } // Có các tác vụ trong hàng đợi hiện tại chưa được xử lý và activeCount < maxSize, tiếp tục mở rộng nếu (runnableQueue.size() > 0 && activeCount < maxSize) { for (int i = coreSize; i < maxSize; i++) { newThread(); } } // Nếu không có tác vụ nào trong hàng đợi tác vụ, nó cần được tái chế thành coreSize if (runnableQueue. size() == 0 && activeCount > coreSize) { for (int i = coreSize; i < activeCount; i++) { RemoveThread(); } } } } } // ThreadTask chỉ là sự kết hợp của InterTask và Thread riêng lớp tĩnh ThreadTask { public ThreadTask(Thread thread, InternalTask ​​​internalTask) { this.thread = thread; this.internalTask ​​​​= InternalTask ​​} Thread InternalTask ​​​internalTask; ; } lớp tĩnh riêng DefaultThreadFactory triển khai ThreadFactory { cuối cùng tĩnh riêng AtomicInteger GROUP_COUNTER = mới AtomicInteger(1); riêng tư tĩnh cuối cùng Nhóm ThreadGroup = new ThreadGroup("MyThreadPool=" + GROUP_COUNTER.getAndDecrement()); riêng tư tĩnh cuối cùng AtomicInteger COUNTER = new AtomicInteger(0); Thread(group, có thể chạy được, "thread-pool-" + COUNTER.getAndDecrement()); } } }sleep(keepAliveTime); } Catch (InterruptedException e) { isShutdown = true; break; } được đồng bộ hóa (this) { if (isShutdown) { break; } // Có các tác vụ trong hàng đợi hiện tại chưa được xử lý và activeCount < coreSize, tiếp tục mở rộng if (runnableQueue.size() > 0 && activeCount < coreSize) { for (int i = initSize; i < coreSize; i++) { newThread(); } // Lý do cho countinue là vì chúng ta không muốn thread mở rộng trực tiếp sang maxSize continue } // Nếu có các task trong hàng đợi hiện tại chưa được xử lý và activeCount < maxSize , tiếp tục mở rộng if (runnableQueue.size() > 0 && activeCount < maxSize) { for (int i = coreSize; i < maxSize; i++) { newThread() } } //; Nếu không có tác vụ nào trong hàng tác vụ, nó cần được tái chế thành coreSize if (runnableQueue.size() == 0 && activeCount > coreSize) { for (int i = coreSize; i < activeCount; i++) { RemoveThread() ; } } } } } // ThreadTask chỉ là sự kết hợp giữa InterTask và Thread Private static class ThreadTask { public ThreadTask(Thread thread, InternalTask nội bộTask) { this.thread = thread; this.internalTask ​​​​= InternalTask ​​} Chủ đề luồng; InternalTask ​​​internalTask; } lớp tĩnh riêng DefaultThreadFactory triển khai ThreadFactory { cuối cùng tĩnh riêng AtomicInteger GROUP_COUNTER = new AtomicInteger(1); nhóm = ThreadGroup mới ("MyThreadPool=" + GROUP_COUNTER.getAndDecrement()); cuối cùng tĩnh riêng AtomicInteger COUNTER = new AtomicInteger(0); @Override public Thread createThread(Runnable runnable) { return new Thread(group, runnable, "thread-pool-" + COUNTER.getAndDecrement()) ; } } }sleep(keepAliveTime); } Catch (InterruptedException e) { isShutdown = true; break; } được đồng bộ hóa (this) { if (isShutdown) { break; } // Có các tác vụ trong hàng đợi hiện tại chưa được xử lý và activeCount < coreSize, tiếp tục mở rộng if (runnableQueue.size() > 0 && activeCount < coreSize) { for (int i = initSize; i < coreSize; i++) { newThread(); } // Lý do cho countinue là vì chúng ta không muốn thread mở rộng trực tiếp sang maxSize continue } // Nếu có các task trong hàng đợi hiện tại chưa được xử lý và activeCount < maxSize , tiếp tục mở rộng if (runnableQueue.size() > 0 && activeCount < maxSize) { for (int i = coreSize; i < maxSize; i++) { newThread() } } //; Nếu không có tác vụ nào trong hàng đợi tác vụ, nó cần được tái chế thành coreSize if (runnableQueue.size() == 0 && activeCount > coreSize) { for (int i = coreSize; i < activeCount; i++) { RemoveThread() ; } } } } } // ThreadTask chỉ là sự kết hợp giữa InterTask và Thread Private static class ThreadTask { public ThreadTask(Thread thread, InternalTask nội bộTask) { this.thread = thread; this.internalTask ​​​​= InternalTask; } Chủ đề luồng; InternalTask ​​​internalTask; } lớp tĩnh riêng DefaultThreadFactory triển khai ThreadFactory { cuối cùng tĩnh riêng AtomicInteger GROUP_COUNTER = new AtomicInteger(1); nhóm = ThreadGroup mới ("MyThreadPool=" + GROUP_COUNTER.getAndDecrement()); cuối cùng tĩnh riêng AtomicInteger COUNTER = new AtomicInteger(0); @Override public Thread createThread(Runnable runnable) { return new Thread(group, runnable, "thread-pool-" + COUNTER.getAndDecrement()) ; } } }

NineThreadPoolTest

package concurrent.threadpool; import java.util.concurrent.TimeUnit; public class ThreadPoolTest { public static void main(String[] args) { //Xác định nhóm luồng, luồng ban đầu là 2, luồng lõi là 4 và luồng chính là 4 luồng tối đa là 6, Hàng đợi tác vụ cho phép 1000 tác vụ cuối cùng ThreadPool threadPool = new BasicThreadPool(2, 6, 4, 1000); // Xác định 20 tác vụ và gửi nhóm luồng cho (int i = 0; i < 20; i++) { threadPool.execute(() -> { try { TimeUnit.SECONDS.sleep(10); System.out.println(Thread.currentThread().getName() + " đang chạy và đã xong."); } Catch (InterruptedException e) { e.printStackTrace(); } }); } for (; ; ) { // Xuất liên tục thông tin nhóm luồng System.out.println("getActiveCount:" + threadPool.getActiveCount()); System.out.println("getQueueSize:" + threadPool.getQueueSize()); " + threadPool.getCoreSize()); System.out.println("getMaxSize: " + threadPool.getMaxSize()); thử { TimeUnit.SECONDS.sleep(5); } bắt (InterruptedException e) { e.printStackTrace();

Mười kết quả kiểm tra

getActiveCount: 2 getQueueSize: 18 getCoreSize: 4 getMaxSize: 6 getActiveCount: 2 getQueueSize: 18 getCoreSize: 4 getMaxSize: 6 thread-pool--1 đang chạy và hoàn tất. getActiveCount: 4 getQueueSize: 14 getCoreSize: 4 getMaxSize: 6 getActiveCount: 4 getQueueSize: 14 getCoreSize: 4 getMaxSize: 6 thread-pool--3 đang chạy và hoàn tất. thread-pool--2 đang chạy và hoàn tất. xong. getActiveCount: 6 getQueueSize: 8 getCoreSize: 4 getMaxSize: 6 getActiveCount: 6 getQueueSize: 8 getCoreSize: 4 getMaxSize:6 thread-pool--3 đang chạy và hoàn tất. thread-pool--4 đang chạy và hoàn tất. -0 đang chạy và hoàn tất. thread-pool--1 đang chạy và xong. getQueueSize: 2 getCoreSize: 4 getMaxSize: 6 getActiveCount: 6 getQueueSize: 2 getCoreSize: 4 getMaxSize: 6 thread-pool--4 đang chạy và hoàn tất. thread-pool--3 đang chạy và hoàn tất. thread-pool--1 đang chạy và hoàn tất. thread-pool-0 đang chạy và hoàn tất. getQueueSize: 0 getCoreSize: 4 getMaxSize: 6 thread-pool--4 đang chạy và hoàn tất. getActiveCount: 5 getQueueSize: 0 getCoreSize: 4 getMaxSize: 6