线程池技术及其示例

以你之姓@ 2022-04-24 08:46 150阅读 0赞

### 线程池技术及其示例 ###

*  线程池技术及其示例

# 线程池技术及其示例 #

对于服务端的程序，经常面对成千上万的任务提交到服务器汇总，如果采用一个任务一个线程的方式，那么将会创建数以万计的线程，这不是一个好的选择。因为这会使操作系统频繁的进行线程上下文切换，无故增加系统的负载，而线程的创建和消亡都是需要耗费系统资源的，也无疑浪费了系统资源。

线程池技术能够很好的解决这个问题，它预先创建了若干数量的线程，并且不能由用户直接对线程的创建进行控制，在这个前提下重复使用固定或较为固定数目的线程来完成任务的执行。这样做的好处是，一方面，消除了频繁创建和消亡线程的系统资源开销，另一方面，面对过量任务的提交能够平缓的劣化。

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

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

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

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

import java.util.ArrayList;
    import java.util.Collections;
    import java.util.LinkedList;
    import java.util.List;
    import java.util.concurrent.atomic.AtomicLong;
    
    public class DefaultThreadPool<Job extends Runnable> implements ThreadPool<Job> { 
    
        // 线程池最大限制数
        private static final int MAX_WORKER_NUMBER = 10;
        // 线程池默认的数量
        private static final int DEFAULT_WORKER_NUMBERS = 5;
        // 线程池最小的数量
        private static final int MIN_WORKER_NUMBERS = 1;
        // 这是一个工作队列，将会向里面插入工作
        private final LinkedList<Job> jobs = new LinkedList<Job>();
        // 工作者列表 Woker是自定义的。。。在最后面
        private final List<Worker> workers = Collections.synchronizedList(new ArrayList<Worker>());
        // 工作者线程的数量
        private int workerNum = DEFAULT_WORKER_NUMBERS;
        // 线程编号生成
        private AtomicLong threadNum = new AtomicLong();
    
        public DefaultThreadPool(){ 
            initialWorkers(DEFAULT_WORKER_NUMBERS);
        }
    
        public DefaultThreadPool(int num){ 
            workerNum = num > MAX_WORKER_NUMBER ? MAX_WORKER_NUMBER : num < MIN_WORKER_NUMBERS ? MIN_WORKER_NUMBERS : num;
            initialWorkers(workerNum);
        }
    
        // 初始化线程工作者
        private void initialWorkers(int num){ 
            for (int i = 0 ; i < num ; i ++){ 
                Worker worker = new Worker();
                workers.add(worker);
                // 一个工作者对应一个线程 worker <-> thread 用来执行Job
                Thread thread = new Thread(worker, "ThreadPool-Worker-" + threadNum.incrementAndGet());
                thread.start();
                workers.add(worker);
            }
        }
    
        @Override
        public void execute(Job job) { 
            if (job != null){ 
                // 添加一个工作，然后进行通知
                synchronized (jobs){ 
                    jobs.addLast(job);
                    jobs.notify();
                }
            }
        }
    
        @Override
        public void shutdown() { 
            for (Worker worker : workers){ 
                // 可以理解为停止每个线程
                worker.shutdown();
            }
        }
    
        @Override
        public void addWorkers(int num) { 
            synchronized (jobs){ 
                // 限制新增的Worker数量不能超过最大值
                if (num + this.workerNum > MAX_WORKER_NUMBER){ 
                    num = MAX_WORKER_NUMBER - this.workerNum;
                }
                initialWorkers(num);
                this.workerNum += num;
            }
        }
    
        @Override
        public void removeWorker(int num) { 
            synchronized (jobs){ 
                if (num >= this.workerNum){ 
                    throw new IllegalArgumentException("beyond workNum");
                }
                // 按照给定的数量停止 Worker
                int count = 0;
                while (count < num){ 
                    Worker worker = workers.get(count);
                    if (workers.remove(worker)){ 
                        worker.shutdown();
                        count ++;
                    }
                }
                this.workerNum -= count;
            }
        }
    
        @Override
        public int getJobSize() { 
            return jobs.size();
        }
    
        // 工作者，负责消费任务
        class Worker implements Runnable{ 
    
            // 是否工作
            private volatile boolean running = true;
            @Override
            public void run() { 
                while (running){ 
                    Job job = null;
                    synchronized (jobs){ 
                        // 如果工作者列表是空的，那么就wait
                        while (jobs.isEmpty()){ 
                            try{ 
                                jobs.wait();
                            }catch (InterruptedException ex){ 
                                // 感知到外部对WorkerThread的中断操作，返回
                                Thread.currentThread().interrupt();
                                return;
                            }
                        }
                        // 取出一个Job
                        job = jobs.removeFirst();
                    }
                    if (job != null){ 
                        try { 
                            job.run();
                        }catch (Exception ex){ 
                            // 忽略 Job 执行中的 Exception
                        }
                    }
                }
            }
    
            public void shutdown(){ 
                running = false;
            }
        }
    }

测试类编写

public class Main { 
        public static void main(String[] args) { 
            ThreadPool tp = new DefaultThreadPool();
            int count = 10;
            for (int i = 0 ; i < count ; i++){ 
                tp.execute(new Task(i));
            }
        }
    }
    
    class Task implements Runnable{ 
    
        private int num;
    
        public Task(int num){ 
            this.num = num;
        }
    
        @Override
        public void run() { 
            System.out.println("Task" + num + " running...");
            try { 
                Thread.sleep((int)(Math.random()*5000) + 1000);
            }catch (InterruptedException e){ 
                e.printStackTrace();
            }
        }
    }

输出

Task0 running...
    Task1 running...
    Task2 running...
    Task3 running...
    Task4 running...
    Task5 running... // 从这开始是后面陆续出现的
    Task6 running...
    Task7 running...
    Task8 running...
    Task9 running...

因为核心线程池大小默认是5，所以开始有五个执行，又我们在执行方法中设定了线程睡眠时间1-6秒范围，所以后面的五个是陆续执行的。

从线程的实现可以看到，当客户端调用execute(Job)方法时，会不断地向任务列表jobs中添加Job，而每个工作者线程会不断地从jobs上取出一个Job进行执行，当jobs为空时，工作者线程进入等待状态。

添加一个Job后，对工作队列jobs调用了其notify()方法，而不是notifyAll()方法，因为能够确定有工作者线程被唤醒，这时使用notify()方法将会比notifyAll()方法获得更小的开销（避免将等待队列中的线程全部移动到阻塞队列中）。

可以看到，线程池的本质就是使用了一个线程安全的工作队列连接工作者线程和客户端线程，客户端线程将任务放入工作队列后便返回，而工作者线程则不断地从工作队列上取出工作并执行。当工作队列为空时，所有的工作者线程均等待在工作队列上，当有客户端提交了一个任务之后会通知任意一个工作者线程，随着大量的任务被提交，更多的工作者线程会被唤醒。