栏目分类:
子分类:
返回
名师互学网用户登录
快速导航关闭
当前搜索
当前分类
子分类
实用工具
热门搜索
名师互学网 > IT > 软件开发 > 后端开发 > Java

自定义线程池总结

Java 更新时间: 发布时间: IT归档 最新发布 模块sitemap 名妆网 法律咨询 聚返吧 英语巴士网 伯小乐 网商动力

自定义线程池总结

线程池:

分成两个部分:线程池和阻塞任务队列
先说阻塞任务队列:
属性:
1. 阻塞任务队列数据结构
2. 从阻塞任务队列中添加任务和取走任务需要的锁
3. 任务取完条件变量
4. 任务为空条件变量
5. 最大容量
方法:
1. 获取任务
2. 带超时的获取任务
3. 添加任务到阻塞任务队列
4. 带超时的添加任务到阻塞任务队列
5. 基于各种拒绝策略的放入任务到阻塞任务队列

class BlockQueue{
    //1.阻塞任务队列
    private Deque queue = new ArrayDeque<>();

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

    //3.任务取完条件变量
    private Condition fullWaitSet = lock.newCondition();

    //4.任务为空条件变量
    private Condition emptyWaitSet = lock.newCondition();

    //5.容量
    private int capcity;

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

    //获取任务
    public T take(){
        lock.lock();
        try {
            while(queue.isEmpty()){
                try {
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T task = queue.removeFirst();
            fullWaitSet.signal();
            return task;
        }finally {
            lock.unlock();
        }
    }
    //带超时的获取
    public T take(long timeout, TimeUnit unit){
        lock.lock();
        try {
            //转换时间
            long nanos = unit.toNanos(timeout);
            while(queue.isEmpty()){
                try {
                    if (nanos<=0){
                        return null;
                    }
                    //更新nanos
                    nanos = emptyWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T task = queue.removeFirst();
            fullWaitSet.signal();
            return task;
        }finally {
            lock.unlock();
        }
    }

    //添加任务到阻塞任务队列
    public void put(T task){
        lock.lock();
        try {
            while (queue.size()== capcity){
                System.out.println("等待加入任务队列"+task);
                try {
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println("加入任务队列"+task);
            queue.addLast(task);
            emptyWaitSet.signal();
        }finally {
            lock.unlock();
        }
    }

    //带超时的添加任务到阻塞任务队列
    public boolean put(T task,long timeout,TimeUnit unit){

        lock.lock();
        try {
            //转换时间
            long nanos = unit.toNanos(timeout);
            while (queue.size()== capcity){
                System.out.println("等待加入任务队列"+task);
                if (nanos<=0){

                    return false;
                }
                try {
                    nanos = fullWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println("加入任务队列"+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 rejectPolicy, T task) {
        lock.lock();
        try{
            if (queue.size()== capcity){
                rejectPolicy.reject(this,task);
            }else{
                System.out.println("加入任务队列"+task);
                queue.addLast(task);
                emptyWaitSet.signal();
            }
        }finally {
            lock.unlock();
        }
    }
}

再说线程池:
属性:
1. 保存正在执行线程的数据结构
2. 阻塞任务队列
3. 核心线程数
4. 获取任务的超时时间
5. 拒绝策略
6. 内部类worker包装线程
方法:
1. 线程池执行任务

class ThreadPool{
    //阻塞任务队列
    private BlockQueue taskQueue;

    //线程集合,使用woker类包装Thread,携带更多信息
    private HashSet workers = new HashSet<>();

    //核心线程数
    private int coreSize;

    //获取任务的超时时间
    private long timeout;
    private TimeUnit timeUnit;

    //拒绝策略
    private RejectPolicy rejectPolicy;

    public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit,int queueCapcity,RejectPolicy rejectPolicy) {
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.taskQueue = new BlockQueue<>(queueCapcity);
        this.rejectPolicy = rejectPolicy;
    }

    public void execute(Runnable task){
        //当任务数没有超过coreSize时,直接交给worker执行
        //如果任务数超过coreSize时,加入任务队列暂存
        synchronized (workers){
            if (workers.size() 

来一个任务,通过线程池执行的整体流程:
如果核心线程数大于 正在执行的线程数,则直接用一个线程去执行任务,并且把这个线程加入保存正在执行线程的数据结构
如果核心线程数小于等于 正在运行的线程数,尝试把任务放入阻塞任务队列
如果阻塞任务队列没满,直接把任务加入队列
如果阻塞任务队列满了,执行拒绝策略(一定要记住,拒绝策略是针对阻塞任务队列满了的时候放入不了任务时该怎么办)
当线程将当前任务执行完时,会自动获取任务队列里的任务,如果获取超时,则直接将该线程移出保存正在执行线程的数据结构

拒绝策略函数式接口:

@FunctionalInterface
interface RejectPolicy{
    void reject(BlockQueue queue,T taks);
}

拒绝策略具体实现,这里是直接在初始化线程池时实现:
我们也可以定义一个枚举来封装拒绝策略,和jdk线程池一样

public class TestPool {
    public static void main(String[] args) {
        ThreadPool threadPool = new ThreadPool(1,1000,TimeUnit.MILLISECONDS,1,(queue,task)->{
            //1.死等
            //queue.put(task);

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

            //3.调用者自己放弃
            //System.out.println("放弃"+task);

            //4.调用者抛出异常
            //throw new RuntimeException("任务加入阻塞任务队列失败"+task);

            //5.调用者自己执行任务
            task.run();
        });
        for (int i=0;i<4;i++){
            int t = i;
            threadPool.execute(()->{
                try {
                    Thread.sleep(1000L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(t);
            });
        }
    }
}
转载请注明:文章转载自 www.mshxw.com
本文地址:https://www.mshxw.com/it/760032.html
我们一直用心在做
关于我们 文章归档 网站地图 联系我们

版权所有 (c)2021-2022 MSHXW.COM

ICP备案号:晋ICP备2021003244-6号