手写JAVA线程池

news2024/10/5 19:13:11

前言

手写一个简单的java线程池：重点关注，如何确保一直有运行的线程？如何确保线程消费提交的任务信息？。一直保存有运行的线程底层使用的是死循环。使用消息队列确保信息的提交和消费。消息队列使用先进先出原则。

步骤

线程池核心点：复用机制

1、提前创建好固定的线程一直在运行状态----死循环实现
2、提交的线程任务缓存到一个并发队列集合中，交给我们正在运行的线程执行
3、正在运行的线程就从队列中获取该任务执行

1、创建固定线程

根据传入的参数，确定要生成几个一直运行的线程。比如传入的参数是2，就会创建出2个一直运行的线程。一直运行的线程使用死循环来实现。

public class MyTestExecutors {
    private List<WorkThread> workThread;

    /**
     * 最大线程数
     * @param maxThreadCount
     */
    public MyTestExecutors(int maxThreadCount){
        //提前创建好固定的线程一直在运行状态---死循环实现
        workThread = new ArrayList<WorkThread>(maxThreadCount);
        for (int i = 0; i < maxThreadCount; i++) {
            new WorkThread().start();
        }
    }

    class WorkThread extends Thread{
        @Override
        public void run() {
            while (true){

            }
        }
    }
}

2、提交任务到消息队列

提交的线程任务缓存到一个并发队列集合中，交给我们正在运行的线程执行。

提示：offer()方法是Java队列中经常使用的方法，它的作用是向队列尾部添加一个元素。具体来说，offer()方法会将指定的元素插入到此队列的末尾，如果队列已满，它将返回false。否则，它会返回true，表示已成功添加了元素。

    /**
     * 线程任务缓存到一个并发队列集合
     * @param command
     * @return
     */
    public boolean execute(Runnable command){
        return runnableDeque.offer(command);
    }


public class MyTestExecutors {
    private List<WorkThread> workThread; //工作线程
    private BlockingDeque<Runnable> runnableDeque; //队列


    /**
     * @param maxThreadCount 最大线程数
     * @param dequeSize      缓存消息队列
     */
    public MyTestExecutors(int maxThreadCount,int dequeSize){
        //1、限制队列容量缓存
        runnableDeque = new LinkedBlockingDeque<Runnable>(dequeSize);
        //2、提前创建好固定的线程一直在运行状态---死循环实现
        workThread = new ArrayList<WorkThread>(maxThreadCount);
        for (int i = 0; i < maxThreadCount; i++) {
            new WorkThread().start();
        }

    }

    class WorkThread extends Thread{
        @Override
        public void run() {
            while (true){

            }
        }
    }

    /**
     * 线程任务缓存到一个并发队列集合
     * @param command
     * @return
     */
    public boolean execute(Runnable command){
        return runnableDeque.offer(command);
    }

    public static void main(String[] args) {
        MyTestExecutors testExecutors = new MyTestExecutors(2, 2);
        for (int i = 0; i < 10; i++) {
            final int finalI = i;
            testExecutors.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName()+","+finalI);
                }
            });
        }
    }
}

3、线程消费消息队列

正在运行的线程就从队列中获取该任务执行

    class WorkThread extends Thread{
        @Override
        public void run() {
            while (true){
                Runnable runnable = runnableDeque.poll();
                if(runnable != null){
                    runnable.run();
                }
            }
        }
    }

4、测试

2个的情况，两个固定线程还未执行任务，且队列中只缓存了两个线程任务。

在这里插入图片描述

4个的情况，固定两个线程正在执行任务，队列中缓存了两个线程任务。
在这里插入图片描述

5、线程停止【完整代码】

如何在线程任务执行完成后，停止两个一直存在的线程。需要将死循环的条件改变，通过给一个标志位。在任务结束后，将该标志位设置为false，同时需要判断，队列中的任务是否执行结束。队列中缓存的任务全部结束，才停止线程。

while (isRun || runnableDeque.size()>0)

package com.grg.demo.testDemo02;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * @author zyz
 * @version 1.0
 * @data 2023/7/11 23:01
 * @Description: 手写线程池
 *  线程池核心点 ：复用机制 -----
 *  1、提前创建好固定的线程一直在运行状态----死循环实现
 *  2、提交的线程任务缓存到一个并发队列集合中，交给我们正在运行的线程执行
 *  3、正在运行的线程就从队列中获取该任务执行
 */
public class MyTestExecutors {
    private List<WorkThread> workThread; //工作线程
    private BlockingDeque<Runnable> runnableDeque; //队列
    private Boolean isRun = true;

    /**
     *
     * @param maxThreadCount 最大线程数
     * @param dequeSize      缓存消息队列
     */
    public MyTestExecutors(int maxThreadCount,int dequeSize){
        //2、限制队列容量缓存
        runnableDeque = new LinkedBlockingDeque<Runnable>(dequeSize);
        //1、提前创建好固定的线程一直在运行状态---死循环实现
        workThread = new ArrayList<WorkThread>(maxThreadCount);
        for (int i = 0; i < maxThreadCount; i++) {
            new WorkThread().start();
        }

    }

    class WorkThread extends Thread{
        @Override
        public void run() {
            while (isRun || runnableDeque.size()>0){
                Runnable runnable = runnableDeque.poll();
                if(runnable != null){
                    runnable.run();
                }
            }
        }
    }

    /**
     * 线程任务缓存到一个并发队列集合
     * @param command
     * @return
     */
    public boolean execute(Runnable command){
        return runnableDeque.offer(command);
    }

    public static void main(String[] args) {
        MyTestExecutors testExecutors = new MyTestExecutors(2, 2);
        for (int i = 0; i < 10; i++) {
            final int finalI = i+1;
            testExecutors.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName()+","+finalI);
                }
            });
        }
        testExecutors.isRun = false;
    }
}