首页 > 代码库 > Java线程池技术以及实现

Java线程池技术以及实现

2024-10-08 21:57:40   216人阅读

　　对于服务端而言，经常面对的是客户端传入的短小任务，需要服务端快速处理并返回结果。如果服务端每次接受一个客户端请求都创建一个线程然后处理请求返回数据，这在请求客户端数量少的阶段看起来是一个不错的选择，但是面对成千上万的请求在某一时段同时到达服务器时，如果还是采用这种方式，那么将会创建数以万计的线程，暂且不讨论线程是否会达到操作系统上限，单单操作系统频繁的进行线程的上下文切换就是一个巨大的开销，无故的增加的系统的负载，而线程的创建和消亡都是需要耗费系统资源的，也无疑造成了资源的浪费。

　　线程池技术就能很好的解决这个问题，它事先创建好了若干个处理请求任务的线程，线程的数量并不受请求客户端的控制，在这前提下使用固定或者较为固定数目的线程来完成请求任务的处理，消除了频繁创建和消亡线程的系统资源开销。

　　

　　下面定义一个简单的线程池接口 ：

public interface ThreadPool<Task extends Runnable> {
    
    //执行一个Task，这个Task需要实现Runnable
    void execute(Task task) ;
    
    //关闭线程池
    void shutdown() ;
    
    //增加工作线程
    void addWorker(int num ) ;
    
    //减少工作线程
    void removeWorker(int num);
    
    //得到正在执行的任务数量
    int getTaskSize() ;
    
}

　　客户端可以通过execute(Task)方法将Task提交入线程池执行，而客户端自身不用等待Task的执行完成。线程池还提供了增加/减小工作线程以及关闭线程池的方法，这里的工作线程就代表了一个重复执行Task的线程，而每个由客户端提交的Task都将进入到一个等待队列中等待工作线程处理。

　　接下来是线程池接口的默认实现 ：

public class DefaultThreadPool<Task extends Runnable> implements ThreadPool<Task> {
    
    //线程池最大限制数
    private static final int MAX_WORKER_NUMBERS = 10;
    //线程池默认的数量
    private static final int DEFAULT_WORKER_NUMBERS= 5;
    //线程池最小数量
    private static final int MIN_WORKER_NUMBERS = 1 ;
    //这是一个任务列表，将会向里面插入任务
    private final LinkedList<Task> jobs = new LinkedList<Task>() ;
    // 工作者列表
    private final List<Worker> workers = Collections.synchronizedList(new ArrayList<Worker>()) ;
    //工作者线程的数量
    private int workerNum = DEFAULT_WORKER_NUMBERS ;
    //线程编号生成
    private AtomicLong threadNum = new AtomicLong() ;
    
    public DefaultThreadPool(){
        initializeWorkers(DEFAULT_WORKER_NUMBERS);
    }
    public DefaultThreadPool(int initSize){
        workerNum = initSize > MAX_WORKER_NUMBERS ? MAX_WORKER_NUMBERS : 
            initSize < MIN_WORKER_NUMBERS ?MIN_WORKER_NUMBERS :  initSize ;
        initializeWorkers(workerNum);
    }
    
    private void initializeWorkers(int num){
        for(int i = 0 ; i < num ; i++){
            Worker worker = new Worker() ;
            workers.add(worker) ;
            Thread t = new Thread(worker, "ThreadPool-worker-"+threadNum.incrementAndGet()) ;
            t.start();
        }
    }

    @Override
    public void execute(Task task) {
        if(task!=null){
            synchronized(jobs){
                jobs.addLast(task);
                //通知
                jobs.notify();
            }
        }
    }

    @Override
    public void shutdown() {
        for(Worker worker : workers){
            worker.shutdown();
        }
    }
    
    @Override
    public void addWorker(int num) {
        //这里是否应该选择workers为监视器锁
        synchronized(workers){
            //判断新增的数量是否超出线程池的内置规则
            if(num+this.workerNum > MAX_WORKER_NUMBERS){
                num = MAX_WORKER_NUMBERS - this.workerNum ;
            }
            initializeWorkers(num);
            this.workerNum += num ;
        }
    }

    @Override
    public void removeWorker(int num) {
        synchronized(workers){
            if(num >= this.workerNum){
                throw new IllegalArgumentException("beyond workNum") ;
            }
            
            for(int i = 0 ; i < num ; i++){
                Worker worker = workers.get(i);
                if(workers.remove(worker)){
                    worker.shutdown();
                }
            }
            this.workerNum -=num ;
        }
    }

    @Override
    public int getTaskSize() {
        
        return jobs.size();
    }
    
    class Worker implements Runnable{
        private volatile boolean running ;
        @Override
        public void run() {
            while(running){
                
                Task job = null ;
                synchronized(jobs){
                    //如果任务列表是空的，那么就wait
                    if(jobs.isEmpty()){
                        try {
                            jobs.wait();
                        } catch (InterruptedException e) {
                            //感知到外部对WorderThread的中断操作。返回
                            Thread.currentThread().interrupt();
                            return ;
                        }
                    }
                    job = jobs.removeFirst() ;
                }
                
                if(job!=null){
                    try {
                        job.run();
                    } catch (Exception e) {
                        //忽略job执行中的exception
                    }
                }
            }//while end
        }//run end
        public void shutdown(){
            running = false ;
        }
    }
}

 

Java线程池技术以及实现


联系
我们