线程工具类

爱被打了一巴掌 2022-06-17 04:18 205阅读 0赞

public class XmlUtils \{  
/\*\*  
\* @param fileName 文件路径  
\* @return  
\*/  
public static Document getDocument(File file)  
\{  
if(null != file)  
\{  
InputStream is = null ;  
try   
\{  
is = new FileInputStream(file) ;  
return getDocument(is) ;  
\} catch (FileNotFoundException e)   
\{  
e.printStackTrace();  
\}  
\}  
  
return null ;  
\}  
  
/\*\*  
\* 从指定输入流当中解析得到xml配置  
\*   
\* @param is  
\* @return  
\* @author qin  
\*/  
public static Document getDocument(InputStream is)  
\{  
Document document = null;  
if(null != is)  
\{  
try   
\{  
SAXReader reader = new SAXReader();  
document = reader.read(is);  
\} catch( Exception e) \{  
e.printStackTrace();  
throw new RuntimeException(e.getMessage(), e);  
\} finally  
\{  
IOUtils.closeQuietly(is) ;  
\}  
\}  
  
        return document;  
\}  
  
/\*\*  
\* 通过文件名获取Document  
\* @param root  
\* @param fileName  
\* @return  
\*/  
public static Document getDocumentByFileName(String root,String fileName)\{  
File file = new File(root,fileName);  
return getDocument(file);  
\}  
/\*\*  
\* 获取根节点  
\* @param document  
\* @return  
\*/  
public static Element getRoot(Document document)\{  
return document.getRootElement();  
\}  
  
/\*\*  
\* 获取单一节点  
\* @param xpath  
\* @return  
\*/  
public static Node getSingleNode(Document document ,String xpath)\{  
return document.selectSingleNode(xpath);  
\}  
/\*\*  
\* 获取节点  
\* @param document  
\* @param xpath  
\* @return  
\*/  
@SuppressWarnings("unchecked")  
public static List<Node> getNodes(Document document ,String xpath)\{  
return document.selectNodes(xpath);  
\}  
/\*\*  
\* 获取节点的属性值  
\* @param node  
\* @param properties  
\* @return  
\*/  
public static String getValue(Node node,String properties)\{  
String properites = "@" + properties;  
return node.valueOf(properites);  
\}  
/\*\*  
\* 更改节点的属性值  
\* @param node  
\* @param name  
\* @param value  
\*/  
public static void setAttribute(Node node,String name,String value)\{  
Element element = (Element) node;  
element.addAttribute(name, value);  
\}  
/\*\*  
\* 设置text的值  
\* @param node  
\* @param value  
\*/  
public static void setText(Node node,String value)\{  
node.setText(value);  
\}  
/\*\*  
\* 保存文档  
\* @param document  
\*/  
public static void saveXml(Document document,String fileName)\{  
XMLWriter writer;  
try \{  
writer = new XMLWriter(new FileWriter(fileName));  
writer.write( document );  
writer.close();  
\} catch (IOException e) \{  
e.printStackTrace();  
throw new RuntimeException(e.getMessage(),e);  
\}

public class ThreadPoolManager \{  
  
  
    private static final Logger log = LoggerFactory.getLogger(ThreadPoolManager.class);  
  
  
  
  
private static ThreadPoolExecutor poolExecutor = null;  
  
private static final ReentrantLock serviceLock = new ReentrantLock();  
  
/\*\*  
\* 启动服务  
\*   
\*/  
public static void initialize(ThreadPoolConfig config) \{  
serviceLock.lock();  
try \{  
if (null == poolExecutor) \{  
ThreadPoolExecutor exector = config.getExecutor();  
if (null == exector) \{  
exector = new RetryThreadPoolExecutor( config.getMinPoolSize(), config.getMaxPoolSize(), config.getKeepAliveTime(),   
config.getWorkQueueSize(), config.getTimeout(), config.getRejectedExecutionHandler() );  
\}  
poolExecutor = exector;  
log.info("ThreadPool init success!!!");  
\}  
\} finally \{  
serviceLock.unlock();  
\}  
\}  
  
/\*\*  
\* 执行一个线程  
\*   
\* @param task 待执行的线程  
\* @return  
\*/  
public static boolean execute(Runnable task) \{  
if (null == task) \{  
throw new NullPointerException();  
\}  
  
// 线程池服务未启动  
if (null == poolExecutor) \{  
throw new NullPointerException("service is not started...");  
\}  
  
try \{  
poolExecutor.execute( task );  
  
return true;  
\} catch(java.util.concurrent.RejectedExecutionException rejectException) \{  
            RdtLogUtil.printStackMessage(log, rejectException);  
\}  
  
return false ;  
\}  
  
/\*\*  
\* 批量执行多个线程  
\*   
\* @param <T>  
\* @param tasks  
\* @return  
\* @throws InterruptedException  
\*/  
public static <T> List<Future<T>> batchExecute(Collection<? extends Callable<T>> tasks) throws InterruptedException \{  
if (null == tasks) \{  
throw new NullPointerException();  
\}  
  
// 线程池服务未启动  
if (null == poolExecutor) \{  
throw new NullPointerException( "thread pool is not started..." );  
\}  
  
  
if (0 < tasks.size()) \{  
return poolExecutor.invokeAll(tasks);  
\}  
  
return null;  
\}  
  
/\*\*  
\* 终止服务  
\*/  
public static void shutdown() throws Exception\{  
serviceLock.lock();  
try \{  
if(null != poolExecutor) \{  
poolExecutor.shutdown();  
poolExecutor = null;   
\}  
\} catch (Exception e) \{  
throw e;  
\}finally \{  
serviceLock.unlock();  
\}  
\}  
  
/\*\*  
\*   
 \* 新增，用来返回Callable线程的future  
 \* 2015.11.24  
 \*  
 \* @Title: submit  
 \* @Description:   
 \* @param task  
 \* @return  
\*/  
public static <T> Future<T> submit(Callable<T> task)\{  
return poolExecutor.submit(task) ;  
\}

public class ThreadPoolConfig \{  
  
private int minPoolSize;                 // 线程池保持的最小的线程数  
private long keepAliveTime;              // 线程池中大于minmumPoolSize的线程保持活动的最长时间(单位:毫秒)  
private int maxPoolSize;                 // 线程池中正在执行的最大线程数  
private int workQueueSize;               // 线程池缓冲队列中等待执行的线程数  
private long timeout;                    // 线程执行超时时间  
private ThreadPoolExecutor executor;     // 线程执行器  
private RejectedExecutionHandler rejectedExecutionHandler;    // 没有获得执行的线程捕捉器  
  
  
  
public ThreadPoolConfig() \{  
\}  
  
  
public int getMinPoolSize() \{  
return minPoolSize;  
\}  
  
  
public void setMinPoolSize(int \_minPoolSize) \{  
this.minPoolSize = \_minPoolSize;  
\}  
  
  
public int getMaxPoolSize() \{  
return maxPoolSize;  
\}  
  
  
public void setMaxPoolSize(int \_maxPoolSize) \{  
this.maxPoolSize = \_maxPoolSize;  
\}  
  
  
public long getKeepAliveTime() \{  
return keepAliveTime;  
\}  
  
  
public void setKeepAliveTime(long \_keepAliveTime) \{  
this.keepAliveTime = \_keepAliveTime;  
\}  
  
  
public int getWorkQueueSize() \{  
return workQueueSize;  
\}  
  
  
public void setWorkQueueSize(int \_workQueueSize) \{  
this.workQueueSize = \_workQueueSize;  
\}  
  
  
public long getTimeout() \{  
return timeout;  
\}  
  
  
public void setTimeout(long \_timeout) \{  
this.timeout = \_timeout;  
\}  
  
  
public ThreadPoolExecutor getExecutor() \{  
return executor;  
\}  
  
  
public void setExecutor(ThreadPoolExecutor \_executor) \{  
this.executor = \_executor;  
\}  
  
  
public RejectedExecutionHandler getRejectedExecutionHandler() \{  
return rejectedExecutionHandler;  
\}  
  
  
public void setRejectedExecutionHandler(  
RejectedExecutionHandler rejectedHandler) \{  
this.rejectedExecutionHandler = rejectedHandler;  
\}

public class RetryThreadPoolExecutor extends ThreadPoolExecutor \{  
  
private final ReentrantLock tokenLock = new ReentrantLock();   // 令牌锁  
private final Condition tocken = tokenLock.newCondition();  
private long timeout ; // 等待令牌超时时间  
private RejectedExecutionHandler rejectHandler;  // 线程拒绝处理器  
  
/\*\*  
\* 重试策略线程执行器  
\*   
\* @param corePoolSize  
\* @param maximumPoolSize  
\* @param keepAliveTime  
\* @param queueSize  
\* @param timeout  
\* @param rejectHandler  
\*/  
public RetryThreadPoolExecutor( int corePoolSize, int maximumPoolSize,long keepAliveTime, int queueSize, long timeout, RejectedExecutionHandler rejectHandler ) \{  
/\*\*  
\*  采用AbortPolicy策略  
\*  指定缓冲队列大小，FIFO策略  
\*/   
super(corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(queueSize), new ThreadPoolExecutor.AbortPolicy());  
this.timeout = timeout;  
this.rejectHandler = rejectHandler;  
\}  
  
  
@Override  
public void execute(Runnable task) \{  
try \{  
super.execute(task);  
  
  
\} catch(java.util.concurrent.RejectedExecutionException rejectException) \{  
tokenLock.lock();  
try \{  
// 等待令牌，如果超时，则放弃执行  
if ( tocken.await(timeout, TimeUnit.MILLISECONDS)) \{  
execute(task);  
\} else \{  
if (null != rejectHandler) \{  
rejectHandler.rejectedExecution(task, this);  
\}  
throw rejectException;  
\}  
\} catch (InterruptedException e) \{  
throw rejectException;  
\} finally \{  
tokenLock.unlock();  
\}  
\}  
\}  
  
  
@Override  
protected void afterExecute(Runnable task, Throwable t) \{  
super.afterExecute(task, t);  
  
// 如果有线程正在等待令牌  
tokenLock.lock();  
try \{  
//唤醒一个等待线程，交给令牌  
if(tokenLock.hasWaiters(tocken)) \{  
tocken.signal();  
\}  
\} finally \{  
tokenLock.unlock();  
\}  
\}  
  
@Override  
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException \{  
return super.invokeAll( tasks, timeout, TimeUnit.MILLISECONDS );  
\}  
  
  
public long getTimeout() \{  
return timeout;  
\}  
  
  
public void setTimeout(long timeout) \{  
this.timeout = timeout;  
\}  
  
  
public RejectedExecutionHandler getRejectHandler() \{  
return rejectHandler;  
\}  
  
  
public void setRejectHandler(RejectedExecutionHandler rejectHandler) \{  
this.rejectHandler = rejectHandler;  
\}

public class ParseTPoolConfig \{  
  
  
/\*\*  
\* 从配置文档中加载线程池配置  
\* @param document  
\* @return  
\* @throws Exception  
\*/  
public static ThreadPoolConfig getThreadPoolConfig(Document document) throws Exception \{  
ThreadPoolConfig config = new ThreadPoolConfig() ;  
  
// 最小连接数  
String param = XmlUtils.getSingleNode(document, "//threadPool/minPoolSize").getText() ;  
if (StringUtils.isNumeric(param)) \{  
config.setMinPoolSize(Integer.parseInt(param)) ;  
\}  
  
// 最大连接数  
param = XmlUtils.getSingleNode(document, "//threadPool/maxPoolSize").getText() ;  
if (StringUtils.isNumeric(param)) \{  
config.setMaxPoolSize(Integer.parseInt(param)) ;  
\}  
  
// 工作队列最大等待数  
param = XmlUtils.getSingleNode(document, "//threadPool/workQueueSize").getText() ;  
if (StringUtils.isNumeric(param)) \{  
config.setWorkQueueSize(Integer.parseInt(param)) ;  
\}  
  
// 超过最小连接数的连接最大活动时间  
param = XmlUtils.getSingleNode(document, "//threadPool/keepAliveTime").getText() ;  
if (StringUtils.isNumeric(param)) \{  
config.setKeepAliveTime(Long.parseLong(param)) ;  
\}  
  
// 连接超时时间  
param = XmlUtils.getSingleNode(document, "//threadPool/timeout").getText() ;  
if (StringUtils.isNumeric(param)) \{  
config.setTimeout(Long.parseLong(param)) ;  
\}  
  
// 任务执行失败捕捉器  
param = XmlUtils.getSingleNode(document, "//threadPool/rejectHandler").getText() ;  
  
if (StringUtils.isNotEmpty(param)) \{  
Class<?> executorClass =  Class.forName(param) ;  
RejectedExecutionHandler rejectHandler = (RejectedExecutionHandler) executorClass.newInstance() ;  
config.setRejectedExecutionHandler(rejectHandler) ;  
\}  
  
return config ;  
\}

public class ExecuteFailtureHandler implements RejectedExecutionHandler\{  
  
private final static Logger LOG = LoggerFactory.getLogger(ExecuteFailtureHandler.class);  
  
private static final long sleetTime = 500 ;  
  
private static final short retry = 3 ;  
  
  
@Override  
public void rejectedExecution(Runnable task, ThreadPoolExecutor executor) \{  
  
int batch = 0 ;  
while( batch < retry)  
\{  
try   
\{  
// 当前线程休眠一段时间，重新执行  
Thread.sleep(sleetTime) ;  
  
executor.execute(task) ;  
break ;  
  
\} catch (Exception e)  
\{  
batch ++ ;  
  
if( batch < retry )  
\{  
LOG.debug("----a task execute failure, retry \{\}", batch) ;  
\}else   
\{  
LOG.error("----a task re execute failure----") ;   
\}  
\}  
\}  
  
\}

document = RdtXmlUtil.loadXML("/config/threadPool.xml");  
            ThreadPoolManager.initialize(ParseTPoolConfig.getThreadPoolConfig(document));

public class RdtXmlUtil \{  
  
  
/\*\*  
\* 加载配置文件  
\* @param xmlFile  
\* @return  
\* @throws  
\*/  
public static Document loadXML(String xmlFile) throws Exception  
    \{  
if (StringUtils.isEmpty(xmlFile))  
        \{  
throw new Exception("xml config file is empty") ;  
\}  
  
Document document = null ;  
try \{  
// 加载配置  
File file = new File(RdtXmlUtil.class.getResource(xmlFile).toURI()) ;  
document = XmlUtils.getDocument(file) ;  
\} catch (Exception e)  
        \{  
throw e ;  
\}   
  
return document ;  
\}