自定义线程池 04 - 使用演示

完整代码已上传gitee ,地址 :朱元杰的开源仓库 -- ThreadPool核心源码仿写

完整文章栏目地址在:Water的学习分享 - ThreadPool仿写

现在我们已经完善了自定义线程池,接下来进行效果展示

线程池 、 任务队列 、 拒绝策略 代码如下

import lombok.extern.slf4j.Slf4j;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j(topic = "MyThreadPool")
public class MyThreadPool {
    // 任务队列
    private MyBlockingQueue<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) {
        // 当任务数没有超过 coreSize 时,直接交给 worker 对象执行
        // 如果任务数超过 coreSize 时,加入任务队列暂存
        synchronized (workers) {
            if (workers.size() < coreSize) {
                Worker worker = new Worker(task);
                log.info("新增 worker{}, {}", worker, task);
                workers.add(worker);
                worker.start();
            } else {
                // taskQueue.put(task);
                taskQueue.tryPut(rejectPolicy, task);
            }
        }
    }

    public MyThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapcity,
                        RejectPolicy<Runnable> rejectPolicy) {
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.taskQueue = new MyBlockingQueue<>(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.take()) != null) {
            while (task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) {
                try {
                    log.debug("正在执行...{}", task);
                    task.run();
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    task = null;
                }
            }
            synchronized (workers) {
                log.debug("worker 被移除{}", this);
                workers.remove(this);
            }
        }
    }
}

// 拒绝策略
@FunctionalInterface
interface RejectPolicy<T> {
    void reject(MyBlockingQueue<T> queue, T task);
}

//阻塞队列
@Slf4j(topic = "MyBlockingQueue")
class MyBlockingQueue<T> {

    // 任务队列
    private Deque<T> queue = new ArrayDeque<>();

    // 锁
    private ReentrantLock lock = new ReentrantLock();

    // 生产者条件变量
    private Condition fullWaitSet = lock.newCondition();

    // 消费者条件变量
    private Condition emptyWaitSet = lock.newCondition();

    // 容量
    private int capcity;

    public MyBlockingQueue(int capcity) {
        this.capcity = capcity;
    }

    // 阻塞获取
    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 {
                    log.debug("等待加入任务队列 {} ...", task);
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("加入任务队列 {}", task);
            queue.addLast(task);
            emptyWaitSet.signal();
        } finally {
            lock.unlock();
        }
    }

    // 带超时阻塞获取
    public T poll(long timeout, TimeUnit unit) {
        lock.lock();
        try {
            // 将 timeout 统一转换为 纳秒
            long nanos = unit.toNanos(timeout);
            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 boolean offer(T task, long timeout, TimeUnit timeUnit) {
        lock.lock();
        try {
            long nanos = timeUnit.toNanos(timeout);
            while (queue.size() == capcity) {
                try {
                    if (nanos <= 0) {
                        return false;
                    }
                    log.debug("等待加入任务队列 {} ...", task);
                    nanos = fullWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("加入任务队列 {}", task);
            queue.addLast(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 { // 有空闲
                log.debug("加入任务队列 {}", task);
                queue.addLast(task);
                emptyWaitSet.signal();
            }
        } finally {
            lock.unlock();
        }
    }
}

编写测试代码如下:

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.TimeUnit;

@Slf4j(topic = "MyBlockingQueue")
class MyThreadPoolTest {

    public static void main(String[] args) {
        MyThreadPool threadPool = new MyThreadPool(1, 1000, TimeUnit.MILLISECONDS, 1, (queue, task) -> {

            // 1) 死等
//            queue.put(task);

            // 2) 带超时等待
//            queue.offer(task, 500, TimeUnit.MILLISECONDS);

            // 3) 让调用者放弃任务执行
//            log.debug("放弃{}", task);

            // 4) 让调用者抛出异常
//            throw new RuntimeException("任务执行失败 " + task);

            // 5) 让调用者自己执行任务
//            task.run();

        });
        for (int i = 0; i < 4; i++) {
            int j = i;
            threadPool.execute(() -> {
                try {
                    Thread.sleep(1000L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.debug("{}", j);
            });
        }
    }

}

测试代码中完成了两件事,创建线程池 和 提交任务

提交任务

依次提交 4 个任务,每个任务的内容都相同,即 睡眠 1000毫秒 后 ,log.debug 自己的任务序号

创建线程池

我们指定了

  • 核心线程数 = 1
  • 线程超时获取任务的时间 = 1000 毫秒
  • 任务队列容量 = 1
  • 实现了 拒绝策略对象

为了能更清除的看清除执行过程,我们将 核心线程数 和 任务队列容量 都置为了 1

运行效果

死等的拒绝策略

运行日志如下:

2023-08-10 11:32:29 402 [main] [MyThreadPool] [INFO] - 新增 workerThread[Thread-0,5,main], com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@23ab930d
2023-08-10 11:32:29 403 [main] [MyBlockingQueue] [DEBUG] - 加入任务队列 com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@1ed6993a
2023-08-10 11:32:29 403 [Thread-0] [MyThreadPool] [DEBUG] - 正在执行...com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@23ab930d
2023-08-10 11:32:29 403 [main] [MyBlockingQueue] [DEBUG] - 等待加入任务队列 com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@7e32c033 ...
2023-08-10 11:32:30 409 [Thread-0] [MyBlockingQueue] [DEBUG] - 0
2023-08-10 11:32:30 409 [Thread-0] [MyThreadPool] [DEBUG] - 正在执行...com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@1ed6993a
2023-08-10 11:32:30 409 [main] [MyBlockingQueue] [DEBUG] - 加入任务队列 com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@7e32c033
2023-08-10 11:32:30 410 [main] [MyBlockingQueue] [DEBUG] - 等待加入任务队列 com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@7ab2bfe1 ...
2023-08-10 11:32:31 421 [Thread-0] [MyBlockingQueue] [DEBUG] - 1
2023-08-10 11:32:31 421 [Thread-0] [MyThreadPool] [DEBUG] - 正在执行...com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@7e32c033
2023-08-10 11:32:31 421 [main] [MyBlockingQueue] [DEBUG] - 加入任务队列 com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@7ab2bfe1
2023-08-10 11:32:32 436 [Thread-0] [MyBlockingQueue] [DEBUG] - 2
2023-08-10 11:32:32 436 [Thread-0] [MyThreadPool] [DEBUG] - 正在执行...com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@7ab2bfe1
2023-08-10 11:32:33 449 [Thread-0] [MyBlockingQueue] [DEBUG] - 3
2023-08-10 11:32:34 463 [Thread-0] [MyThreadPool] [DEBUG] - worker 被移除Thread[Thread-0,5,main]

第一行:第一个任务(尾号930d,在第三行体现)到来时,线程数为零小于核心线程数,所以创建线程,直接将任务交给线程 第二行:第二个任务(尾号993a)到来时,线程数等于核心线程数,而任务队列还没满,放入任务队列 第四行:第三个任务(尾号c033)到来时,线程数等于核心线程数,任务队列也满了,进行阻塞等待,main线程阻塞 第三、五行:第一个任务(尾号930d)执行完 第六行:第二个任务(尾号993a)被线程执行 第七行:因为第二个任务(尾号993a)被消费者线程从任务队列中取出,任务队列有空位,main线程被唤醒,将第三个任务(尾号c033)放入任务队列 ... 接下来同理,不在分析

带超时等待的拒绝策略

当我们将超时时间设置为500毫秒时,因为一个任务执行完的时间大雨1000毫秒(因为有Thread.sleep(1000L)),那么必定有任务在放入队列的时候会等待超时

运行结果如下:

2023-08-10 11:39:03 909 [main] [MyThreadPool] [INFO] - 新增 workerThread[Thread-0,5,main], com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@23ab930d
2023-08-10 11:39:03 911 [main] [MyBlockingQueue] [DEBUG] - 加入任务队列 com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@1ed6993a
2023-08-10 11:39:03 911 [main] [MyBlockingQueue] [DEBUG] - 等待加入任务队列 com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@7e32c033 ...
2023-08-10 11:39:03 911 [Thread-0] [MyThreadPool] [DEBUG] - 正在执行...com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@23ab930d
2023-08-10 11:39:04 418 [main] [MyBlockingQueue] [DEBUG] - 等待加入任务队列 com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@7ab2bfe1 ...
2023-08-10 11:39:04 920 [Thread-0] [MyBlockingQueue] [DEBUG] - 0
2023-08-10 11:39:04 920 [Thread-0] [MyThreadPool] [DEBUG] - 正在执行...com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@1ed6993a
2023-08-10 11:39:05 932 [Thread-0] [MyBlockingQueue] [DEBUG] - 1
2023-08-10 11:39:06 945 [Thread-0] [MyThreadPool] [DEBUG] - worker 被移除Thread[Thread-0,5,main]

我们看到 第三行 的任务(尾号c033) 和 第五行的任务(尾号bfe1),都有等待加入任务队列,但都因为等待超时而没有被放入任务队列从而也没有被执行

但如果我们将等待时间设置足够长 为 1500 毫秒,运行结果如下:

2023-08-10 11:42:43 371 [main] [MyThreadPool] [INFO] - 新增 workerThread[Thread-0,5,main], com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@23ab930d
2023-08-10 11:42:43 372 [main] [MyBlockingQueue] [DEBUG] - 加入任务队列 com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@1ed6993a
2023-08-10 11:42:43 372 [main] [MyBlockingQueue] [DEBUG] - 等待加入任务队列 com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@7e32c033 ...
2023-08-10 11:42:43 372 [Thread-0] [MyThreadPool] [DEBUG] - 正在执行...com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@23ab930d
2023-08-10 11:42:44 382 [Thread-0] [MyBlockingQueue] [DEBUG] - 0
2023-08-10 11:42:44 382 [Thread-0] [MyThreadPool] [DEBUG] - 正在执行...com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@1ed6993a
2023-08-10 11:42:44 382 [main] [MyBlockingQueue] [DEBUG] - 加入任务队列 com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@7e32c033
2023-08-10 11:42:44 383 [main] [MyBlockingQueue] [DEBUG] - 等待加入任务队列 com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@7ab2bfe1 ...
2023-08-10 11:42:45 384 [Thread-0] [MyBlockingQueue] [DEBUG] - 1
2023-08-10 11:42:45 384 [Thread-0] [MyThreadPool] [DEBUG] - 正在执行...com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@7e32c033
2023-08-10 11:42:45 384 [main] [MyBlockingQueue] [DEBUG] - 加入任务队列 com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@7ab2bfe1
2023-08-10 11:42:46 386 [Thread-0] [MyBlockingQueue] [DEBUG] - 2
2023-08-10 11:42:46 386 [Thread-0] [MyThreadPool] [DEBUG] - 正在执行...com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@7ab2bfe1
2023-08-10 11:42:47 396 [Thread-0] [MyBlockingQueue] [DEBUG] - 3
2023-08-10 11:42:48 404 [Thread-0] [MyThreadPool] [DEBUG] - worker 被移除Thread[Thread-0,5,main]

可以看到所有的任务都成功被执行

让调用者放弃任务执行的拒绝策略

运行结果:

2023-08-10 11:43:50 130 [main] [MyThreadPool] [INFO] - 新增 workerThread[Thread-0,5,main], com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@23ab930d
2023-08-10 11:43:50 131 [main] [MyBlockingQueue] [DEBUG] - 加入任务队列 com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@1ed6993a
2023-08-10 11:43:50 131 [main] [MyBlockingQueue] [DEBUG] - 放弃com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@7e32c033
2023-08-10 11:43:50 131 [main] [MyBlockingQueue] [DEBUG] - 放弃com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@7ab2bfe1
2023-08-10 11:43:50 131 [Thread-0] [MyThreadPool] [DEBUG] - 正在执行...com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@23ab930d
2023-08-10 11:43:51 140 [Thread-0] [MyBlockingQueue] [DEBUG] - 0
2023-08-10 11:43:51 140 [Thread-0] [MyThreadPool] [DEBUG] - 正在执行...com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@1ed6993a
2023-08-10 11:43:52 149 [Thread-0] [MyBlockingQueue] [DEBUG] - 1
2023-08-10 11:43:53 160 [Thread-0] [MyThreadPool] [DEBUG] - worker 被移除Thread[Thread-0,5,main]

因为执行第一个任务的时间最少需要 1000 毫秒,这个时间 main 线程是完全可以遍历完for循环,尝试将第三、四个任务放入任务队列的,但此时任务队列已经被第二个任务占满,因此main线程会直接放弃第三、四个任务

让调用者抛出异常的拒绝策略

运行结果:

2023-08-10 11:46:43 380 [main] [MyThreadPool] [INFO] - 新增 workerThread[Thread-0,5,main], com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@23ab930d
2023-08-10 11:46:43 381 [main] [MyBlockingQueue] [DEBUG] - 加入任务队列 com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@1ed6993a
2023-08-10 11:46:43 381 [Thread-0] [MyThreadPool] [DEBUG] - 正在执行...com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@23ab930d
Exception in thread "main" java.lang.RuntimeException: 任务执行失败 com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@7e32c033
	at com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest.lambda$main$0(MyThreadPoolTest.java:22)
	at com.zhuyuanjie.myBlockingQueue.MyBlockingQueue.tryPut(MyThreadPool.java:214)
	at com.zhuyuanjie.myBlockingQueue.MyThreadPool.execute(MyThreadPool.java:37)
	at com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest.main(MyThreadPoolTest.java:30)
2023-08-10 11:46:44 396 [Thread-0] [MyBlockingQueue] [DEBUG] - 0
2023-08-10 11:46:44 396 [Thread-0] [MyThreadPool] [DEBUG] - 正在执行...com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@1ed6993a
2023-08-10 11:46:45 409 [Thread-0] [MyBlockingQueue] [DEBUG] - 1
2023-08-10 11:46:46 414 [Thread-0] [MyThreadPool] [DEBUG] - worker 被移除Thread[Thread-0,5,main]

让调用者放弃任务执行的拒绝策略 类似,只不过是在抛弃任务的时候抛出异常

让调用者自己执行任务的拒绝策略

运行结果:

2023-08-10 11:48:43 370 [main] [MyThreadPool] [INFO] - 新增 workerThread[Thread-0,5,main], com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@23ab930d
2023-08-10 11:48:43 371 [main] [MyBlockingQueue] [DEBUG] - 加入任务队列 com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@1ed6993a
2023-08-10 11:48:43 371 [Thread-0] [MyThreadPool] [DEBUG] - 正在执行...com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@23ab930d
2023-08-10 11:48:44 373 [main] [MyBlockingQueue] [DEBUG] - 2
2023-08-10 11:48:44 373 [Thread-0] [MyBlockingQueue] [DEBUG] - 0
2023-08-10 11:48:45 377 [main] [MyBlockingQueue] [DEBUG] - 3
2023-08-10 11:48:45 377 [Thread-0] [MyThreadPool] [DEBUG] - 正在执行...com.zhuyuanjie.myBlockingQueue.MyThreadPoolTest$$Lambda$2/758705033@1ed6993a
2023-08-10 11:48:46 379 [Thread-0] [MyBlockingQueue] [DEBUG] - 1
2023-08-10 11:48:47 393 [Thread-0] [MyThreadPool] [DEBUG] - worker 被移除Thread[Thread-0,5,main]

我们看到 序号为 2 和 3 的任务(对应第三和第四个任务)都是由main线程自己执行的

end

评论

新增邮件回复功能,回复将会通过邮件形式提醒,请填写有效的邮件!