Java利用线程工厂监控线程池的实现示例

public interface ThreadFactory {
    Thread newThread(Runnable r);
}

public class ThreadfactoryText {
    public static void main(String[] args) {
        Runnable runnable=new Runnable() {
            @Override
            public void run() {
                int num=new Random().nextInt(10);
                System.out.println(Thread.currentThread().getId()+"--"+System.currentTimeMillis()+"--睡眠"+num);
                try {
                    TimeUnit.SECONDS.sleep(num);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        //创建线程池 使用自定义线程工厂 采用默认的拒绝策略
        ExecutorService executorService=new ThreadPoolExecutor(5, 5, 0, TimeUnit.SECONDS, new SynchronousQueue<>(), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread t=new Thread(r);
                t.setDaemon(true);//设置为守护线程，当主线程运行结束，线程池中线程也会被释放
                System.out.println("创建了线程"+t);
                return t;
            }
        });
        //提交五个任务
        for (int i = 0; i < 5; i++) {
            executorService.submit(runnable);
        }
    }
}

int getActiveCount()//获得线程池只当前的获得线程数量
long getCompletedTaskCount()//返回线程池完成任务数量
int getCorePoolSize()//线程池中核心任务数量
int getLargestPoolSize() //返回线程池中曾经达到线程的最大数
int getMaximumPoolSize()//返回线程池的最大容量
int getPoolSize()//返回线程大小
BlockingQueue<Runnable> getQueue()//返回阻塞队列
long getTaskCount()//返回线程池收到任务总数

public class Text {
    public static void main(String[] args) throws InterruptedException {
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                System.out.println(Thread.currentThread().getId() + "线程开始执行--" + System.currentTimeMillis());
                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        //创建线程池 使用默认线程工厂 有界队列  采用DiscardPolicy策略
        ThreadPoolExecutor executorService = new ThreadPoolExecutor(2, 5, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(5),Executors.defaultThreadFactory(),new ThreadPoolExecutor.DiscardPolicy());
        //提交五个任务
        for (int i = 0; i < 30; i++) {
            executorService.submit(runnable);
            System.out.println("当前线程核心线程数"+executorService.getCorePoolSize()+",最大线程数："+executorService.getMaximumPoolSize()+",当前线程池大小："+executorService.getPoolSize()+"活动线程数："+executorService.getActiveCount()+",收到任务："+executorService.getTaskCount()+"完成任务数："+executorService.getCompletedTaskCount()+"等待任务数："+executorService.getQueue().size());
            TimeUnit.MILLISECONDS.sleep(500);
        }
        System.out.println("-------------------");
        while (executorService.getActiveCount()>=0)//继续对线程池进行检测
        {
          System.out.println("当前线程核心线程数"+executorService.getCorePoolSize()+",最大线程数："+executorService.getMaximumPoolSize()+",当前线程池大小："+executorService.getPoolSize()+"活动线程数："+executorService.getActiveCount()+",收到任务："+executorService.getTaskCount()+"完成任务数："+executorService.getCompletedTaskCount()+"等待任务数："+executorService.getQueue().size());
            Thread.sleep(1000);//每1秒检测一次
        }

    }
}

protected void beforeExecute(Thread t, Runnable r) { }
protected void afterExecute(Runnable r, Throwable t) { }

public void run() {
            runWorker(this);
}
final void runWorker(Worker w) {
                try {
                    beforeExecute(wt, task);
                    try {
                        task.run();
                        afterExecute(task, null);
                    } catch (Throwable ex) {
                        afterExecute(task, ex);
                        throw ex;
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
    }      

package com;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class Text07 {
    public static void main(String[] args) {

        //定义扩展线程池 定义线程池类继承ThreadPoolExecutor，然后重写其他方法
        ExecutorService threadPoolExecutor=
 new ThreadPoolExecutor(5,5,0, TimeUnit.SECONDS,new LinkedBlockingDeque<>()){
     //在内部类重写开始方法
     @Override
     protected void beforeExecute(Thread t, Runnable r) {
         System.out.println(t.getId()+"线程准备执行任务"+((Mytask)r).name);
     }
     //在内部类重写结束方法
     @Override
     protected void afterExecute(Runnable r, Throwable t) {
         System.out.println(((Mytask)r).name+"执行完成");
     }
     //线程池退出
     @Override
     protected void terminated() {
         System.out.println("线程池退出");
     }
 };
        for (int i = 0; i < 5; i++) {
            Mytask mytask=new Mytask("Thread"+i);
            threadPoolExecutor.execute(mytask);
        }
    }
    private  static  class  Mytask implements Runnable
    {
        private  String name;

        public  Mytask(String name)
        {
            this.name=name;
        }
        @Override
        public void run() {
            System.out.println(name+"正在被执行"+Thread.currentThread().getId());
            try {
                Thread.sleep(1000);//模拟任务时长
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class Text09 {
    public static void main(String[] args) {
        //创建线程池
        ExecutorService executorService=new ThreadPoolExecutor(5,5,0, TimeUnit.SECONDS,new SynchronousQueue<>());
        //向线程池中添加两个数相处计算的任务
        for (int i = 0; i <5 ; i++) {
            executorService.submit(new Text(10,i));
        }

    }
    private  static class  Text implements  Runnable
    {
        private  int x;
        private  int y;
        public  Text(int x,int y)
        {
            this.x=x;
            this.y=y;
        }
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName()+"线程x/y结果的为"+x+"/"+y+"="+(x/y));
        }
    }
}

package com;

import java.util.concurrent.*;

public class Text09 {
    public static void main(String[] args) {
        //创建线程池  使用自定义的线程池
        ExecutorService executorService=new TranceThreadPoorExcuter(5,5,0, TimeUnit.SECONDS,new SynchronousQueue<>());
        //向线程池中添加两个数相处计算的任务
        for (int i = 0; i <5 ; i++) {
            executorService.submit(new Text(10,i));
        }

    }
    public  static class  Text implements  Runnable
    {
        public  int x;
        public  int y;
        public  Text(int x,int y)
        {
            this.x=x;
            this.y=y;
        }

        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName()+"线程x/y结果的为"+x+"/"+y+"="+(x/y));
        }
    }
    //自定义线程池类 对TranceThreadPoorExcuter进行扩展
    private  static  class  TranceThreadPoorExcuter extends  ThreadPoolExecutor
    {

        public TranceThreadPoorExcuter(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        }
        //定义一个方法用于传入两个参数 第一个是要接受的任务 第二个是Exception
        public  Runnable warp(Runnable r,Exception e)
        {
            return new Runnable() {
                @Override
                public void run() {

                    try {
                        r.run();
                    }
                    catch (Exception e1)
                    {
                        e.printStackTrace();
                        throw e1;
                    }
                }
            };
        }
        //重写submit方法
        @Override
        public Future<?> submit(Runnable task) {
            return super.submit(warp(task,new Exception("客户跟踪异常")));
        }
        //还可以重写excute方法
    }
}