支持停止、暂停、恢复的高级特性线程类

喜欢ヅ旅行 2022-05-27 05:26 192阅读 0赞

开发过程中，多线程处理比较耗时的程序，就会需要线程停止、暂停、恢复的功能。  
本人对Runnable实现了一个高级特性的线程类，其功能如下：  
（1）线程停止、暂停、恢复  
（2）支持无限循环

（3）支持有限循环

高级特性的线程类：

package com.thread;
    
    import java.util.List;
    import java.util.concurrent.atomic.AtomicInteger;
    
    import org.apache.log4j.Logger;
    
    import com.google.common.collect.Lists;
    
    /**
     * 高级特性的线程类
     * 功能如下：
     * 1、线程停止、暂停、恢复
     * 2、无限循环
     * 3、有限循环
     * @author changtan.sun
     *
     */
    public class AdvancedThreadRunnable implements Runnable {
    	private static final Logger logger = Logger.getLogger(AdvancedThreadRunnable.class);
    
    	private static final int MAX_NO_DATA_COUNTER = 6;
    
    	private RunnableJob job = null;
    
    	/**
    	 * 线程名称
    	 */
    	private String name = "";
    	/**
    	 * 线程停止标志
    	 */
    	private volatile boolean stopFlag = false;
    	/**
    	 * 线程暂停标识
    	 */
    	private volatile boolean suspendFlag = false;
    
    	/**
    	 * 线程是否Alive
    	 */
    	private volatile boolean aliveFlag = false;
    	/**
    	 * 没有数据可执行时的计数器，当计数器达到一定预置后，说明已经没有任何数据，则线程停止。
    	 */
    	private AtomicInteger nodataCounter = new AtomicInteger(0);
    
    	/**
    	 * 设置延迟间隔时间，默认1秒
    	 */
    	private volatile long delay = 1000;
    	/**
    	 * 是否启用延迟间隔时间
    	 */
    	private volatile boolean isEnableDelay = false;
    
    	private Object lockObj = new Object();
    
    	public AdvancedThreadRunnable() {
    	}
    	
    	public AdvancedThreadRunnable(String name) {
    		this.name = name;
    	}
    
    	public String getName() {
    		return name;
    	}
    
    	public void setName(String name) {
    		this.name = name;
    	}
    
    	public RunnableJob getJob() {
    		return job;
    	}
    
    	public void setJob(RunnableJob job) {
    		this.job = job;
    	}
    
    	@Override
    	public void run() {
    		if (job == null) {
    			return;
    		}
    
    		aliveFlag = true;
    		job.doPrepare(); //准备工作
    		try {
    			if (!job.isLimited()) { //无限循环
    				while (!stopFlag) {
    					synchronized (lockObj) {
    						if (suspendFlag) {
    							try {
    								lockObj.wait();
    							} catch (InterruptedException e) {
    								e.printStackTrace();
    							}
    						}
    					}
    
    					try {
    						RunnableJobStatus status = job.run(null);
    						if (status == RunnableJobStatus.NODATA) {
    							int counter = nodataCounter.getAndIncrement();
    							if (counter > MAX_NO_DATA_COUNTER) {
    								this.requestStop();
    							} else {
    								AdvancedThreadRunnable.sleep(1000);
    							}
    						} else {
    							nodataCounter.set(0);
    						}
    					} catch (Exception ex) {
    						job.doException(null, ex);
    						logger.error(ex);
    					}
    
    					if (this.isEnableDelay) {
    						AdvancedThreadRunnable.sleep(this.delay);
    					} else {
    						AdvancedThreadRunnable.sleep(100);
    					}
    				}
    			} else {//有限循环
    				List<?> workItems = job.getWorkItems();
    				if (workItems == null) {
    					return;
    				}
    				for (Object workItem : workItems) {
    					if (stopFlag) {
    						break;
    					}
    					synchronized (lockObj) {
    						if (suspendFlag) {
    							try {
    								lockObj.wait();
    							} catch (InterruptedException e) {
    								e.printStackTrace();
    							}
    						}
    					}
    
    					try {
    						job.run(workItem);
    					} catch (Exception ex) {
    						logger.error(ex);
    						job.doException(workItem, ex);
    					}
    
    					if (this.isEnableDelay) {
    						AdvancedThreadRunnable.sleep(this.delay);
    					} else {
    						AdvancedThreadRunnable.sleep(10);
    					}
    				}
    			}
    
    			job.doFinish(); //完成工作
    		} finally {
    			aliveFlag = false;
    
    			job.doDispose(); //销毁工作
    		}
    	}
    
    	public boolean isStop() {
    		return stopFlag;
    	}
    
    	public boolean isSuspend() {
    		return suspendFlag;
    	}
    
    	public boolean isAlive() {
    		return aliveFlag;
    	}
    
    	/**
    	 * 请求线程停止
    	 * @return
    	 */
    	public void requestStop() {
    		logger.info("线程停止....");
    		this.stopFlag = true;
    	}
    
    	/**
    	 * 请求线程暂停
    	 */
    	public void requestSuspend() {
    		logger.info("线程暂停....");
    		suspendFlag = true;
    	}
    
    	/**
    	 * 请求线程恢复
    	 */
    	public void requestResume() {
    		logger.info("线程恢复....");
    		if (suspendFlag) {
    			synchronized (lockObj) {
    				lockObj.notifyAll();
    			}
    			suspendFlag = false;
    		}
    	}
    
    	public long getDelay() {
    		return delay;
    	}
    
    	public void setDelay(long delay) {
    		this.delay = delay;
    	}
    
    	public boolean isEnableDelay() {
    		return isEnableDelay;
    	}
    
    	public void setEnableDelay(boolean isEnableDelay) {
    		this.isEnableDelay = isEnableDelay;
    	}
    
    	public static void sleep(long millis) {
    		try {
    			Thread.sleep(millis);
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    	}
    
    	/**
    	 * 线程具体执行的任务
    	 * @author changtan.sun
    	 *
    	 */
    	public static interface RunnableJob {
    		/**
    		 * 任务中任务工作项列表需要处理。
    		 * 比如 100条数据要在job中执行， 那么就有100个任务工作项
    		 * @return
    		 */
    		public List<?> getWorkItems();
    
    		/**
    		 * 是否有限循环
    		 * 比如：job中处理100条数据， 那么需要设置为true，表示有限循环，这时getWorkItems()就有意义了
    		 * 比如：不断轮询处理某数据，则需要设置为false，表示无限循环
    		 * @return
    		 */
    		public boolean isLimited();
    
    		/**
    		 * 执行任务工作项
    		 * @param workItem
    		 * @return
    		 */
    		public RunnableJobStatus run(Object workItem) throws ExecuteException;
    
    		/**
    		 * 异常处理
    		 * @param t
    		 */
    		public void doException(Object workItem, Throwable t);
    
    		/**
    		 * 准备工作的处理
    		 */
    		public void doPrepare();
    
    		/**
    		 * 完成工作处理
    		 */
    		public void doFinish();
    
    		/**
    		 * 销毁工作处理
    		 */
    		public void doDispose();
    	}
    	
    	/**
    	 * RunnableJob的空默认值实现
    	 * @author changtan.sun
    	 *
    	 */
    	public static class DefaultRunnableJob implements RunnableJob {
    
    		@Override
    		public List<?> getWorkItems() {
    			return null;
    		}
    
    		@Override
    		public boolean isLimited() {
    			return false;
    		}
    
    		@Override
    		public RunnableJobStatus run(Object workItem) throws ExecuteException {
    			return null;
    		}
    
    		@Override
    		public void doException(Object workItem, Throwable t) {
    		}
    
    		@Override
    		public void doPrepare() {
    		}
    
    		@Override
    		public void doFinish() {
    		}
    
    		@Override
    		public void doDispose() {
    			
    		}
    	}
    
    	/**
    	 * 任务执行返回的状态
    	 * @author changtan.sun
    	 *
    	 */
    	public static enum RunnableJobStatus {
    		/**
    		 * 执行成功
    		 */
    		SUCCESS,
    		/**
    		 * 执行失败
    		 */
    		FAILURE,
    		/**
    		 * 没有数据可执行
    		 */
    		NODATA
    	}
    }

异常类：

package com.thread;
    
    public class ExecuteException extends Exception {
    	private static final long serialVersionUID = -783262020204730527L;
    
    	public ExecuteException() {  
            super();  
        }  
      
        public ExecuteException(String message) {  
            super(message);  
        }  
      
        public ExecuteException(String message, Throwable cause) {  
            super(message, cause);  
        }  
      
        public ExecuteException(Throwable cause) {  
            super(cause);  
        }  
    }

测试代码如下：

package com.thread;
    
    import java.util.List;
    
    import com.google.common.collect.Lists;
    import com.thread.AdvancedThreadRunnable.DefaultRunnableJob;
    import com.thread.AdvancedThreadRunnable.RunnableJobStatus;
    
    public class AdvancedThreadRunnableTest {
    	public static void main(String[] args) {
    		AdvancedThreadRunnable runnable = new AdvancedThreadRunnable();
    		runnable.setName("线程1");
    		runnable.setJob(new DefaultRunnableJob() {
    			@Override
    			public RunnableJobStatus run(Object workItem) {
    				System.out.println(workItem);
    				return RunnableJobStatus.SUCCESS;
    			}
    
    			@Override
    			public List<Object> getWorkItems() {
    				List<Object> data = Lists.newArrayList();
    				for (int i = 0; i < 2000; i++) {
    					data.add("数据" + i);
    				}
    
    				return data;
    			}
    
    			@Override
    			public boolean isLimited() {
    				return true;
    			}
    		});
    
    		Thread t = new Thread(runnable);
    		t.start();
    
    		AdvancedThreadRunnable.sleep(4000);
    		runnable.requestSuspend();
    		AdvancedThreadRunnable.sleep(10000);
    
    		runnable.requestResume();
    
    		AdvancedThreadRunnable.sleep(10000);
    		runnable.requestStop();
    
    	}
    
    }