自己实现一个简单线程池

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了自己实现一个简单线程池相关的知识,希望对你有一定的参考价值。

 

先上原理图:

技术分享

 

上代码之前,要先补充一下线程池构造的核心几个点

  1. 线程池里的核心线程数与最大线程数
  2. 线程池里真正工作的线程worker
  3. 线程池里用来存取任务的队列BlockingQueue
  4. 线程中的任务task

本例实现简化了一些,只实现了BlockingQueue存放任务,然后每个worker取任务并执行,下面看代码
首先定义一个线程池ThreadExcutor

class ThreadExcutor{

    //创建
    private volatile boolean RUNNING = true;

    //所有任务都放队列中,让工作线程来消费
    private static BlockingQueue<Runnable> queue = null;

    private final HashSet<Worker> workers = new HashSet<Worker>();

    private final List<Thread> threadList = new ArrayList<Thread>();

    
    int poolSize = 0;     //池子大小
    
    int currentSize = 0;  //创建了多少个线程

    boolean shutdown = false;

    public ThreadExcutor(int poolSize){
        this.poolSize = poolSize;
        queue = new LinkedBlockingQueue<Runnable>(poolSize);
    }
    
    //提交任务给线程池
    public void exec(Runnable runnable) {
        if (runnable == null) throw new NullPointerException();
        if(currentSize < poolSize){     //当前工作线程数小于池子大小
            addThread(runnable);        //创建一个线程来执行任务
        }else{
            //System.out.println("offer" +  runnable.toString() + "   " + queue.size());
            try {
                queue.put(runnable);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    
    public void addThread(Runnable runnable){
        currentSize ++;
        Worker worker = new Worker(runnable);        //实例化一个Worker的同时,将runnable任务添加进队列
        workers.add(worker);                        
        Thread t = new Thread(worker);               //创建一个线程,worker本身实现了runnable接口
        threadList.add(t);
        try {
            t.start();                           //开启线程 执行worker的run方法
        }catch (Exception e){
            e.printStackTrace();
        }

    }

    public void shutdown() {
        RUNNING = false;
        if(!workers.isEmpty()){
            for (Worker worker : workers){
                worker.interruptIfIdle();              //调用worker成员方法,中断自身所在线程
            }
        }
        shutdown = true;
        Thread.currentThread().interrupt();
    }
   //这里留个位置放内部类Worker
    。。
 }

 

 

然后定义一个内部类Worker,这个内部类Worker是用来执行每个任务的,在创建线程池后,往线程里添加任务,每个任务都是由Worker一个一个来启动的。

//工作线程worker
class  Worker implements Runnable{

        public Worker(Runnable runnable){
            queue.offer(runnable);
        }
        
        // worker内部类实现了runnable接口,run方法循环从queue队列里取任务执行,直到池子shutdown
        @Override
        public void run() {
            while (RUNNING){
                Runnable task = null;
                if(shutdown == false){
                    try {
                        task = getTask();
                        task.run();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                
            }
        }

        public Runnable getTask() throws InterruptedException {
            return queue.take();
        }

        public void interruptIfIdle() {                            //idle 空闲的
            for (Thread thread :threadList) {
                System.out.println(thread.getName() + " interrupt");
                thread.interrupt();
            }
        }
    }

 

首先注意的一点,这个Worker是个内部类,是在线程池内声明的。

exec方法

 技术分享

 

Worker怎么工作

技术分享

 

这个工作线程实例化的时候就先加入一个任务到队列中,也就是说在实例化这个工作线程时,这个工作线程也是一个任务被加入到线程池中。然后就是run方法,这个run方法是线程调start方法生成的线程,而Worker调的run方法并没有生成新的线程。就是一个循环,一直在不停的从队列中取任务,然后执行。可以看到,取队列的方法是take(),这个方法意思如果队列为空了,取不到数据时就阻塞队列。

然后看shutdown()

技术分享

 

你每天辛勤的劳动着,突然接收到上面的命令,说活暂时不要接了,先停下来,当你还没搞清楚状况时,接着你的领导又把你开除了,说公司要倒了,你先下岗吧,一会我也得下岗了。这就是shutdown做的事,shutdown必须是主线程才能停止工作线程。
shutdown方法并不是用线程那种强制停止的搞法,而是先用一个标识符告诉工作线程,不要再接任务了。然后通知工作线程,你可以interrupt()了,当所有的线程停止后记得要把主线程也停掉,这样,一个简单任务的线程池就完成了。
让我们来测试一下:

public class Main{
     public static void main(String[] args){
         ThreadExcutor excutor = new ThreadExcutor(3);
         for (int i = 0; i < 3; i++) {
             excutor.exec(new Runnable() {
                 @Override
                 public void run() {
                     System.out.println("线程 " + Thread.currentThread().getName() + " 在帮我干活");
                 }
             });
         }
        excutor.shutdown();
         
     }
 }

 


 

 

 

 

import java.util.*;
import java.util.concurrent.*;



public class Main{
     public static void main(String[] args){
         ThreadExcutor excutor = new ThreadExcutor(3);
         for (int i = 0; i < 1000; i++) {
             excutor.exec(new Runnable() {
                 @Override
                 public void run() {
                     System.out.println("线程 " + Thread.currentThread().getName() + " 在帮我干活");
                 }
             });
         }
        excutor.shutdown();
         
     }
 }

class ThreadExcutor{

    //创建
    private volatile boolean RUNNING = true;

    //所有任务都放队列中,让工作线程来消费
    private static BlockingQueue<Runnable> queue = null;

    private final HashSet<Worker> workers = new HashSet<Worker>();

    private final List<Thread> threadList = new ArrayList<Thread>();

    
    int poolSize = 0;     //池子大小
    
    int currentSize = 0;  //创建了多少个线程

    boolean shutdown = false;

    public ThreadExcutor(int poolSize){
        this.poolSize = poolSize;
        queue = new LinkedBlockingQueue<Runnable>(poolSize);
    }
    
    //提交任务给线程池
    public void exec(Runnable runnable) {
        if (runnable == null) throw new NullPointerException();
        if(currentSize < poolSize){     //当前工作线程数小于池子大小
            addThread(runnable);        //创建一个线程来执行任务
        }else{
            //System.out.println("offer" +  runnable.toString() + "   " + queue.size());
            try {
                queue.put(runnable);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    
    public void addThread(Runnable runnable){
        currentSize ++;
        Worker worker = new Worker(runnable);        //实例化一个Worker的同时,将runnable任务添加进队列
        workers.add(worker);                        
        Thread t = new Thread(worker);               //创建一个线程,worker本身实现了runnable接口
        threadList.add(t);
        try {
            t.start();                           //开启线程 执行worker的run方法
        }catch (Exception e){
            e.printStackTrace();
        }

    }

    public void shutdown() {
        RUNNING = false;
        if(!workers.isEmpty()){
            for (Worker worker : workers){
                worker.interruptIfIdle();              //调用worker成员方法,中断自身所在线程
            }
        }
        shutdown = true;
        Thread.currentThread().interrupt();
    }
   //这里留个位置放内部类Worker
    class  Worker implements Runnable{

        public Worker(Runnable runnable){
            queue.offer(runnable);
        }
        
        // worker内部类实现了runnable接口,run方法循环从queue队列里取任务执行,直到池子shutdown
        @Override
        public void run() {
            while (RUNNING){
                Runnable task = null;
                if(shutdown == false){
                    try {
                        task = getTask();
                        task.run();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                
            }
        }

        public Runnable getTask() throws InterruptedException {
            return queue.take();
        }

        public void interruptIfIdle() {                            //idle 空闲的
            for (Thread thread :threadList) {
                System.out.println(thread.getName() + " interrupt");
                thread.interrupt();
            }
        }
    }
 }

 

 

http://www.cnblogs.com/wxwall/p/7050698.html




以上是关于自己实现一个简单线程池的主要内容,如果未能解决你的问题,请参考以下文章

Motan在服务provider端用于处理request的线程池

简单实现java线程池

初学线程池--1,自己实现一个线程池

线程池-实现一个取消选项

newCacheThreadPool()newFixedThreadPool()newScheduledThreadPool()newSingleThreadExecutor()自定义线程池(代码片段

一个简单的线程池实现