自定义线程池,自定义拒绝策略,线程池参数解释代码实现

线程池状态

ThreadPoolExecutor

ThreadPoolExecutor 使用int的高三位来表示线程池状态,低29位表示线程数量

线程池参数解释

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) 

corePoolSize

• 核心线程数(执行任务的线程数)

maximumPoolSize

• 最大线程数目
• 解释: 比如核心线程设置为2,最大线程设置为3,那除了两个核心线程还会创建一个救急线程,当核心线程满了,阻塞队列满了,就会塞给救急线程

keepAliveTime与unit

• 生存时间-针对救急线程

BlockingQueue

• 阻塞队列
• 就是存放任务的地方,可以无限长的队列,也可以传个指定长度的队列

threadFactory

• 自我理解其实就是给线程创建时取个名字
• 比如: ThreadFactory factory = r -> new Thread(r, “俺是一个好名字”);

RejectedExecutionHandler

• 拒绝策略
• 核心线程满了,阻塞队列满了,救急线程也满了,才会执行拒绝策略

四种默认拒绝策略与著名框架实现的拒绝策略

• AbortPolicy 让调用者抛出RejectExecutionException异常,这是默认策略
• CallerRunsPolicy 让调用者运行任务
• DiscardPolicy 放弃本次任务
• DiscardOldestPolicy 放弃队列中最早的任务,本任务取而代之
• Dubbo的实现, 在抛出RejectExecutionException异常之前会记录日志,并dump线程栈信息,方便定位问题
• Netty的实现,是创建一个新的线程来执行任务
• ActiceMq的实现,带超时等待60s尝试放入队列

自定义一个拒绝策略

public class MyRejected implements RejectedExecutionHandler { 
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { 
        System.out.println("我是自定义的");
    }
}
//使用
/** * 自定义拒绝策略 */
public class TestThreadPool { 
    public static void main(String[] args) { 

        LinkedBlockingQueue<Runnable> runnables= new LinkedBlockingQueue<Runnable>();
        ThreadFactory factory = r -> new Thread(r, "test-thread-pool");

        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 1, 0l, TimeUnit.SECONDS,
                runnables,
                factory,
                new MyRejected()
        );
    }
}

自定义线程池

package cn.test;

import cn.test.config.MyRejected;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import static cn.itcast.util.Sleeper.sleep;

@Slf4j(topic = "c.TestPool2")
public class TestPool2 { 
    public static void main(String[] args) { 

        ThreadPool threadPool = new ThreadPool(1, 1, TimeUnit.SECONDS, 1,(queue,task)->{ 
             //死等
            //queue.put(task);
            //带超时的等待 任务2超时了不会执行
            //queue.offer(task, 3, TimeUnit.SECONDS);
            //调用者放弃执行
            //log.debug("不做任何操作 我放弃");
            //抛出异常
            //throw new RuntimeException("执行任务失败"+task);
            //让调用者自己执行
            task.run();
        });
        for (int i = 0; i < 4; i++) { 
            int j =i;
            threadPool.execute(()->{ 

                log.debug("执行任务 {}",j);
                sleep(1);
            });
        }
    }

}

@FunctionalInterface
interface  RejectPolicy<T> { 
    void reject(BlockingQueue<T> queue,T task);
}
@Slf4j(topic = "c.ThreadPool")
class ThreadPool { 

    //任务队列
    private BlockingQueue<Runnable> taskQueue;
    //线程集合
    private HashSet<Work> works = new HashSet<>();
    //核心数
    private int coreSize;
    //获取任务超时时间
    private long time;

    private TimeUnit timeUnit;
    //线程满了时拒绝策略
    private RejectPolicy<Runnable> rejectPolicy;

    public void execute(Runnable runnable) { 
        if (works.size() >= coreSize) { 

            //taskQueue.put(runnable);
            taskQueue.tryPut(rejectPolicy,runnable);
        } else { 
            Work work = new Work(runnable);
            works.add(work);
            log.debug("{}开始执行任务",runnable);
            work.start();
        }

    }

    public ThreadPool(int coreSize, long time, TimeUnit timeUnit, int capcity,RejectPolicy<Runnable> rejectPolicy) { 
        this.coreSize = coreSize;
        this.time = time;
        this.timeUnit = timeUnit;
        this.rejectPolicy=rejectPolicy;
        taskQueue = new BlockingQueue<>(capcity);
    }


    class Work extends Thread { 
        private Runnable task;

        public Work(Runnable task) { 
            this.task = task;
        }

        @Override
        public void run() { 

            while (task != null || (task = taskQueue.poll(time,timeUnit)) != null) { 
                try { 
                    log.debug("执行任务....");
                    task.run();
                } catch (Exception ex) { 
                    ex.printStackTrace();
                } finally { 
                  task=null;
                }
            }
           synchronized (works){ 
               log.debug("删除线程....");
                works.remove(task);
           }

        }
    }

}

@Slf4j(topic = "c.BlockingQueue")
class BlockingQueue<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 BlockingQueue(int capcity) { 
        this.capcity = capcity;
    }

    //带超时的阻塞获取
    public T poll(long time, TimeUnit timeUnit) { 
        lock.lock();
        long nanos = timeUnit.toNanos(time);
        try { 
            while (queue.isEmpty()) { 
                if (nanos <= 0) { 
                    return null;
                }
                try { 
                    nanos = emptyWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
            T t = queue.removeFirst();
            fullWaitSet.signal();
            return t;
        } finally { 
            lock.unlock();
        }
    }

    //阻塞获取
    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 { 
                    fullWaitSet.await();
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
            queue.addLast(task);
            emptyWaitSet.signal();
        } finally { 
            lock.unlock();
        }
    }
    //阻塞生产
    public Boolean offer(T element,long timeout,TimeUnit timeUnit) { 
        lock.lock();
        long nanos = timeUnit.toNanos(timeout);
        try { 
            while (queue.size() == capcity) { 
                if (nanos<=0){ 
                    log.debug("{}时间到了",element);
                    return false;
                }
                try { 
                    log.debug("{}等待中...",element);
                 nanos= fullWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
            log.debug("{}加入阻塞队列",element);
            queue.addLast(element);
            emptyWaitSet.signal();
            return true;
        } finally { 
            lock.unlock();
        }
    }
    public int getSize() { 
        lock.lock();
        try { 
            return queue.size();
        } finally { 
            lock.unlock();
        }

    }


    public void tryPut(RejectPolicy<T> rejectPolicy, T element) { 

        lock.lock();
        try { 
            if (queue.size()==capcity){ //如果队列满了

                rejectPolicy.reject(this,element);

            }else {   //如果队列没满
                log.debug("{}加入阻塞队列",element);
                queue.addLast(element);
                emptyWaitSet.signal();
            }

        }finally { 
            lock.unlock();
        }
    }
}

注:作者比较垃圾,写的可能都是错的,看看就好