实现线程池及其调用

小咪咪 2023-10-09 19:28 16阅读 0赞

### 实现线程池及其调用 ###

## 1、线程池工具类 ##

import java.util.ResourceBundle;
    import java.util.concurrent.*;
    
    /**
     * @description: 线程管理工具类
     * @description: 默认核心线程数9
     * @description: 最大线程数20
     * @description: 等待线程优先队列20超过继续等待排队
     * @description: 60秒回收非核心线程
     * @param:
     * @return:
     **/
    public class ThreadPoolUtils {
        
    	/**默认核心数*/
    	private static int minPoolSize = 9;
    	/**最大线程数*/
    	private static int maxPoolSize = 20;
    	/**等待线程数*/
    	private  static int waitSize = 20;
    
    	private static ExecutorService executor;
    
    	private static BlockingQueue<Runnable> queue;
    	private static RejectedExecutionHandler handler;
    
    	private static ThreadFactory threadFactory;
    
    	static{
        
    		ResourceBundle reb = ResourceBundle.getBundle(Constans.CONFIG_FILE);
    		if(!StringUtils.isEmpty(reb.getString(Constans.MIN_POOLS_IZE))){
        
    			minPoolSize = Integer.valueOf(reb.getString(Constans.MIN_POOLS_IZE));
    		}
    		if(!StringUtils.isEmpty(reb.getString(Constans.MAX_POOL_SIZE))){
        
    			maxPoolSize = Integer.valueOf(reb.getString(Constans.MAX_POOL_SIZE));
    		}
    		if(!StringUtils.isEmpty(reb.getString(Constans.WAIT_SIZE))){
        
    			waitSize = Integer.valueOf(reb.getString(Constans.WAIT_SIZE));
    		}
    		threadFactory = new CustomThreadFactory(reb.getString(Constans.NAME));
    		queue = new ArrayBlockingQueue<Runnable>(waitSize);
    		handler = new RejectedExecutionHandler() {
        
    			public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        
                    if (!executor.isShutdown()){
        
    
                        r.run();
                    }
                }
            };
    		executor = new MyThreadPool(minPoolSize,maxPoolSize,60L,TimeUnit.MILLISECONDS ,queue,threadFactory,handler);
    	}
    	/**
    	 * 线程工具类
    	 * @param command
    	 */
    	public static void execute(Runnable command){
        
    		executor.execute(command);
    	}
    
    	public static <T> Future<T> submit(Callable<T> task) {
        
    		return executor.submit(task);
    	}
    	/**
    	 * 线程工具类关闭线程
    	 */
    	public static void shutdown(){
        
    		executor.shutdown();
    	}
    
    	private class Constans{
        
    		public static final String CONFIG_FILE = "config/threadPool";
    		public static final String MIN_POOLS_IZE = "minPoolSize";
    		public static final String MAX_POOL_SIZE = "maxPoolSize";
    		public static final String WAIT_SIZE = "waitSize";
    		public static final String NAME = "name";
    	}
    }

import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.util.concurrent.*;
    import java.util.concurrent.atomic.AtomicLong;
    
    /**
     * @description: 自定义实现线程池工具类
     * @param:
     * @return:
     **/
    public class MyThreadPool extends ThreadPoolExecutor{
        
    	/**日志记录*/
    	private final Logger log = LoggerFactory.getLogger(MyThreadPool.class);
    	/**记录每个线程的开始时间*/
    	private final ThreadLocal<Long> startTime = new ThreadLocal<Long>();
    	/**原子锁记录完成的线程数*/
        private final AtomicLong numTasks = new AtomicLong();  
        /**原子锁记录总时间*/
        private final AtomicLong totalTime = new AtomicLong();  
        
        /**
         * 构造方法
         * @param corePoolSize 核心线程数
         * @param maximumPoolSize 最大线程数
         * @param keepAliveTime 非核心线程的闲置超时时间，超过这个时间就会被回收
         * @param unit 指定keepAliveTime的单位
         * @param workQueue 线程池中的任务队列
         * @param handler 线程池中的资源已经全部使用，添加新线程被拒绝时处理规则
         */
    	public MyThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                            BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        
    		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory,handler);
    	}
    	
    	@Override
    	protected void beforeExecute(Thread paramThread, Runnable paramRunnable){
        
    		super.beforeExecute(paramThread, paramRunnable);
            startTime.set(System.currentTimeMillis()); 
    		log.info(String.format("Thread %s: start %s", Thread.currentThread().getName(), startTime.get()));  
    	}
    	
    	@Override
    	protected void afterExecute(Runnable r, Throwable t) {
        	
    		 super.afterExecute(r, t);  
    		 long endTime = System.currentTimeMillis();  
             long taskTime = endTime-startTime.get(); 
             startTime.remove();
             numTasks.incrementAndGet();  
             totalTime.addAndGet(taskTime);  
            
             log.info(String.format("Thread %s: Throwable %s, time=%dms 线程池相关数据，核心数 :%d 最大数量 :%d,活动总数：%d,队列大小：%d，完成数大小：%d", 
            		 Thread.currentThread().getName(),t,taskTime,this.getCorePoolSize(),this.getMaximumPoolSize(), this.getActiveCount(),
            		 this.getQueue().size(),this.getCompletedTaskCount()));
    	}
    	@Override
    	protected void terminated() {
        
    		log.info(String.format("Thread %s: terminated 线程池相关数据，核心数 :%d 最大数量 :%d,活动总数：%d,队列大小：%d，完成数大小：%d,time=%dms", 
    				Thread.currentThread().getName(),this.getCorePoolSize(),this.getMaximumPoolSize(), this.getActiveCount(),
            		 this.getQueue().size(),this.getCompletedTaskCount(),totalTime.get()/numTasks.get()));
    	}
    	
    }

import java.util.concurrent.ThreadFactory;
    import java.util.concurrent.atomic.AtomicInteger;
    
    public class CustomThreadFactory implements ThreadFactory {
        
    		// 计数
    		private AtomicInteger count = new AtomicInteger(0);
    		// 名字
    		private String name;
    		
    		public CustomThreadFactory(String name) {
        
    			if(StringUtils.isEmpty(name)) {
        
    				this.name =  MyThreadPool.class.getSimpleName();
    			}else {
        
    				this.name = name;
    			}
    		}
    		
    		@Override
    		public Thread newThread(Runnable r) {
        
    			Thread t = new Thread(r);
    			String threadName = name +"-"+ count.addAndGet(1);
    			t.setName(threadName);
    			return t;
    		}
    }

## 2、配置文件 ##

resources文件夹下新建config文件夹  
然后新增文件threadPool.properties，内容如下

minPoolSize=20
    maxPoolSize=25
    waitSize=1024
    name=fraginsClaimThread

## 3、业务逻辑(子线程处理的业务逻辑) ##

/**
     * @Param
     * @return
     **/
    public class TestThreadService implements Runnable{
        
    
    
    	private static final Logger logger = LoggerFactory.getLogger(TestThreadService.class);
    
    	private String rid;
    
    	// 构造方法传参
    	public PayResultSendMqThreadService(String rid
    	) {
        
    		super();
    		this.rid = rid;
    
    	}
    
    	@Override
    	public void run() {
        
    		try{
        
    			// 业务逻辑
    
    		} catch (Exception e){
        } finally {
        }
    	}
    
    
    
    	private void handlerErrInfo(boolean errFlag, StringBuilder errBuild) {
        
    		if(errFlag) {
        
    			throw new BusinessException(errBuild.toString());
    		}
    	}
    
    }

## 4、调用点 ##

String rid = "";
    TestThreadService testThreadService =
    		new TestThreadService(
    				rid
    		);
    ThreadPoolUtils.execute(testThreadService );