Java中手动床架一个线程池

原文：Java多线程实战-从零手搓一个简易线程池(一)定义任务等待队列_java等待队列-CSDN博客

package com.thread;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import java.util.ArrayDeque;
import java.util.Deque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;/*** @program: FlinkTutorial-1.17.0* @description:* @author: gaoyf* @create: 2025-08-03 19:36**/
public class BlockQueue<T> {private static final Logger log = LoggerFactory.getLogger(BlockQueue.class);// 双端队列private Deque<T> deque = new ArrayDeque<>();// 队列的容量private int size;public BlockQueue(int size) {this.size = size;}// ReentrantLock可重入锁来解决这个问题private ReentrantLock lock = new ReentrantLock();// 分别定义两个Condition代表队列满情况的等待室与队列空情况的等待室// 队列空情况的休息室Condition emptyCondition = lock.newCondition();// 队列满情况的休息室Condition fullCondition = lock.newCondition();// 定义阻塞添加任务方法put// 添加任务 阻塞添加public void put(T task) {lock.lock();try {while (size == deque.size()) {try {fullCondition.await();} catch (InterruptedException e) {throw new RuntimeException(e);}}log.debug("task 添加成功 ，{}", task);deque.addLast(task);emptyCondition.signal();} finally {lock.unlock();}}// 这里添加任务会存在两张情况：//队列满了：调用fullCondition.await()方法挂起当前生产者线程，也就是让当前生产者线程等待。//队列没满：将任务加入队列中，并调用emptyCondition.signal()通知挂起的消费者。// 编写堵塞拿取任务方法take// 获取任务public T take() {lock.lock();try {while (deque.isEmpty()) {try {emptyCondition.await();} catch (InterruptedException e) {throw new RuntimeException(e);}}T t = deque.removeFirst();fullCondition.signal();log.debug("获取了任务 {}", t);return t;} finally {lock.unlock();}}// 同理，这里拿取任务也会存在两张情况：//队列空的：调用emptyCondition.await()方法挂起当前消费者线程，也就是让当前消费者线程等待。//队列存在元素：任务出栈，并返回出栈的任务元素，然后调用fullCondition.signal()通知挂起的生产者。// 以上我们其实我们就完成了一个简单的任务堵塞队列，但是我们会发现，如果这两个方法都是会一直堵塞，显然是不合理的，所以我们这里新增添加和获取任务的超时方法。// 带超时时间的阻塞添加方法offer// 带超时时间阻塞添加public boolean offer(T task, long timeout, TimeUnit timeUnit) {lock.lock();try {long nanos = timeUnit.toNanos(timeout);while (deque.size() == size) {try {if (nanos <= 0) {return false;}log.debug("等待加入任务队列 {} ...", task);nanos = fullCondition.awaitNanos(nanos);} catch (InterruptedException e) {e.printStackTrace();}}log.debug("加入任务队列 {}", task);deque.addLast(task);emptyCondition.signal();return true;} finally {lock.unlock();}}// 我们可以利用fullCondition.awaitNanos来实现超时等待，当超过给定参数时间时，就会被自动唤醒，并且将传入时间-等待时间作为返回值，下一次循环nanos <= 0时，就会判定为本次操作超时失败了// 带超时时间的阻塞获取方法poll// 带超时时间阻塞获取public T poll(long timeout, TimeUnit unit) {// 1.上锁lock.lock();try {long nanos = unit.toNanos(timeout); // 转为毫秒// 2.首先检查队列是否存在元素while (deque.isEmpty()) {try {// 2.1超时判断，返回值是剩余时间if (nanos <= 0) {return null;}// 2.2超时等待log.debug("等待获取任务");nanos = emptyCondition.awaitNanos(nanos);} catch (InterruptedException e) {e.printStackTrace();}}// 3.拿取元素T task = deque.removeFirst();log.info("任务拿取成功:{}", task);// 4.唤醒挂起的生产者fullCondition.signal();return task;} finally {// 释放锁lock.unlock();}}
}