理解线程池,自己实现一个线程池

线程池本质是一个生产者-消费者模式,一边维护一些线程执行任务,一边由主线程添加一些任务。现在我们抛弃源码中一些繁杂的状态判断,自己写一个线程池。

public class poolT {   //可能频繁增删任务,链表队列效率较高
    private final BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<Runnable>();
    private final HashSet<Work> workers = new HashSet<Work>();
    private static int num = 3;

    public poolT(int num) {
        this.num = num;
        for (int i = 0; i < num; i++) {
            Work w = new Work();
            w.start();
            workers.add(w);
        }
    }

    public void addWork(Runnable r) {
        workQueue.add(r);
    }

    public void close() throws Exception {

        while (!workQueue.isEmpty()) {
            Thread.sleep(500);
        }
        for (Work work : workers) {
            // 通知正在运行的结束
            work.setDrop();
            // 强制结束还在等待的
            if (work.getState() == Thread.State.WAITING) {
                work.interrupt();
            }
        }
        Thread.sleep(2000);
        for (Work work : workers) {
            System.out.println(work.getName() + "状态:" + work.getState());
        }
    }

    // 内部线程封装
    private class Work extends Thread {
        Runnable r = null;
        // 结束线程标志位
        private boolean hasRunning = true;

        public void setDrop() {
            this.hasRunning = false;
        }

        public void run() {
            try {
                while (hasRunning || !workQueue.isEmpty()) {
                    // 阻塞线程执行
                    r = workQueue.take();
                    if (r != null) {
                        r.run();
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws Exception {
        poolT p = new poolT(4);
        for (int i = 0; i < 2; i++) {
            Runnable newRun = new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(1000);
                        System.out.println(Thread.currentThread().getName() + "运行任务;");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
            p.addWork(newRun);
        }
        p.close();
        System.out.println("主程序完毕");
    }
}

这里面我使用了一个阻塞队列,当任务添加时,由队列随机选取一个空闲线程进行处理,没有任务时,进行阻塞。

当然也可以不用阻塞队列,不过需要自己进行同步

public class MyThreadPool {

    List<Runnable> taskList = new LinkedList<Runnable>();

    private List<MyThread> threadList = new LinkedList<MyThread>();

    private static MyThreadPool threadPool;

    public MyThreadPool(int num) {
        for (int i = 0; i < num; i++) {
            threadList.add(new MyThread());
        }
        for (MyThread thread : threadList) {
            thread.start();
        }
    }

      public void destroy() {
            while (!taskList.isEmpty()) {// 如果还有任务没执行完成,就先睡会吧
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            // 工作线程停止工作,且置为null
            for (MyThread thread : threadList) {
                thread.setDistroy();
            }
        }  

    public void execute(Runnable run) {

        synchronized (taskList) {
            taskList.add(run);
            taskList.notify();
        }
    }

    private class MyThread extends Thread {
        public boolean hasRun = true;

        private void setDistroy() {
            this.hasRun = false;
        }

        @Override
        public void run() {
            while (hasRun) {
                Runnable r = null;
                System.out.println(Thread.currentThread().getName() + "is running");
                synchronized (taskList) {
                    if (taskList.isEmpty() && hasRun) {
                        try {
                            taskList.wait(20);
                        } catch (InterruptedException e) {

                            e.printStackTrace();
                        }
                    } else {
                        r = taskList.remove(0);
                    }
                }
                if (r != null) {
                    r.run();
                }
            }
        }
    }

    public static void main(String[] args) throws Exception {
//         ExecutorService excutor=Executors.newFixedThreadPool(3);
        MyThreadPool pool =new MyThreadPool(4);
        pool.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(500);
                    System.out.println("任务一");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
        });
        pool.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(500);
                    System.out.println("任务贰");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
        });

        System.out.println("End");
        pool.destroy();
    }
}

参考:http://blog.csdn.net/hsuxu/article/details/8985931

时间: 2024-11-07 14:05:00

理解线程池,自己实现一个线程池的相关文章

线程问题-一个线程怎么调用另一个线程的结果(2)

1 public class Job1{ 2 public static void main(String[] args){ 3 //ThreadClass t=new ThreadClass(); 4 //RunnableClass r=new RunnableClass(); 5 Sums s=new Sums(); 6 ThreadClass t1=new ThreadClass(s); 7 //Thread t1=new Thread(new ThreadClass(s)); 8 Thr

关于多个异步任务线程如何取消上一个线程的问题

app中遇到这个问题,当用户点击某个标签时,用异步任务开子线程向服务器请求数据并刷新显示.如图: 但难免有用户喜欢点着玩,而且使劲点,作为程序员,我们得宽容用户.所以,我们采用延迟发消息的机制,若用户点击的时间在上一个点击的延迟时间内,即上一个消息还未发出,又来一个发向handler的消息,我们就可以将上一个消息remove掉. 代码如下: handler1.removeMessages(1000); handler1.sendEmptyMessageDelayed(1000, 500); 发送

线程问题—一个线程怎么调出另外一个线程的结果。

问题:建两线程,线程1进行计算1*1+~+99*99,线程2打印出线程1名字,每隔段时间读取一次线程1的计算结果? 1 public class Other{ 2 public static volatile int sum; 3 public static void main(String[] args){ 4 Thread t1=new ThreadClass(); 5 t1.setName("我是线程1"); 6 Thread t2=new Thread(new Runnable

python创建一个线程和一个线程池

创建一个线程 1.示例代码 import time import threading def task(arg): time.sleep(2) while True: num = input('>>>') t = threading.Thread(target=task.args=(num,)) t.start() 创建一个线程池 1.示例代码 import time from concurrent.futures import ThreadPoolExecutor def task(m

线程池异常处理之重启线程处理任务

线程池异常处理之重启线程处理任务 本文记录一下在使用线程池过程中,如何处理 while(true)循环长期运行的任务,在业务处理逻辑中,如果抛出了运行时异常时怎样重新提交任务. 这种情形在Kafka消费者中遇到,当为每个Consumer开启一个线程时, 在线程的run方法中会有while(true)循环中消费Topic数据. 本文会借助Google Guava包中的com.google.common.util.concurrent.ThreadFactoryBuilder类创建线程工厂,因为它能

池化技术——自定义线程池

目录 池化技术--自定义线程池 1.为什么要使用线程池? 1.1.池化技术的特点: 1.2.线程池的好处: 1.3.如何自定义一个线程池 2.三大方法 2.1.单个线程的线程池方法 2.2.固定的线程池的大小的方法 2.3.可伸缩的线程池的方法 2.4.完整的测试代码为: 3.为什么要自定义线程池?三大方法创建线程池的弊端分析 4.七大参数 5.如何手动的去创建一个线程池 6.四种拒绝策略 6.1.会抛出异常的拒绝策略 6.2.哪来的去哪里拒绝策略 6.3.丢掉任务拒绝策略 6.4.尝试竞争拒绝

C#线程系列讲座(3):线程池和文件下载服务器

如果设计一个服务器程序,每当处理用户请求时,都开始一个线程,将会在一定程序上消耗服务器的资源.为此,一个最好的解决方法就是在服务器启动之前,事先创建一些线程对象,然后,当处理客户端请求时,就从这些建好的线程中获得线程对象,并处理请求.保存这些线程对象的结构就叫做线程池. 在C#中可以通过System.Threading.ThreadPool类来实现,在默认情况下,ThreadPool最大可建立500个工作线程和1000个I/O线程(根据机器CPU个数和.net framework版本的不同,这些

Android怎样保证一个线程最多仅仅能有一个Looper?

1. 怎样创建Looper? Looper的构造方法为private,所以不能直接使用其构造方法创建. private Looper(boolean quitAllowed) { mQueue = new MessageQueue(quitAllowed); mThread = Thread.currentThread(); } 要想在当前线程创建Looper.需使用Looper的prepare方法,Looper.prepare(). 假设如今要我们来实现Looper.prepare()这种方法

线程系列02,多个线程同时处理一个耗时较长的任务以节省时间

当面对一个耗时较长的任务时,我们可以把这个任务切分成多个部分,然后同时交给多个线程处理. □ 统计字节数组一个比较耗时的方式 以下来统计一个字节数组的大小. class Program { static byte[] values = new byte[500000000]; static void Main(string[] args) { GenerateByteArray(); Console.WriteLine("正在统计字节数"); Stopwatch watch = new

linux程序设计——取消一个线程(第十二章)

12.7    取消一个线程 有时,想让一个线程能够要求还有一个线程终止,就像给它发送一个信号一样. 线程有方法能够做到这一点,与与信号处理一样.线程能够被要求终止时改变其行为. pthread_cancel是用于请求一个线程终止的函数: #inlude <pthread.h> int pthread_cancel(pthread_t thread); 这个函数提供一个线程标识符就能够发送请求来取消它. 线程能够用pthread_setcancelstate设置线程的取消状态 #include