由于工作的需要，写了一份异步调用的小框架，分享出来。。。
启动类：
/**
* 线程启动
* @author yfguopeng
*/
public class ThreadExecutorListener implements ServletContextListener{
private final static Log log = LogFactory.getLog(ThreadExecutorListener.class);
@SuppressWarnings("unchecked")
public void contextInitialized(ServletContextEvent sce) {
ServletContext servletContext = sce.getServletContext();
WebApplicationContext wac = WebApplicationContextUtils.getRequiredWebApplicationContext(servletContext);
List<ThreadConfigBean> worders = (List<ThreadConfigBean>) wac.getBean("workers");
log.info("=====================初始化线程池========================");
//创建线程组
SecurityManager s = System.getSecurityManager();
ThreadGroup father = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
ThreadGroup group = new ThreadGroup(father, "root-threadgroup");
for (ThreadConfigBean configBean : worders) {
//设置排队队列大小
ArrayBlockingQueue<Runnable> taskQueue = new ArrayBlockingQueue<Runnable>(configBean.getQueueCapacity());
//设置线程工厂
ThreadFactory threadFactory = new DecorateThreadFactory(new ThreadGroup(group,configBean.getBusinessId()),configBean.getBusinessId());
ThreadPoolExecutor worker = new ThreadPoolExecutor(configBean.getBusinessId(),configBean.getMin(), configBean.getMax(), configBean.getKeepAliveTime(),
TimeUnit.SECONDS, taskQueue, threadFactory, configBean.getRejectHandler());
ThreadGroupUtil.addThreadWorker(configBean.getBusinessId(), worker);
}
log.info("=====================线程池初始化完毕========================");
log.info("=====================初始化监控线程========================");
ThreadGroupUtil.monitorThreadStart(group,2000l);
log.info("=====================监控线程初始化完毕========================");
}
public void contextDestroyed(ServletContextEvent sce) {
}
}
业务监控配置：
<bean id="xxxIndex" class="xxx.xxx.xxx.xxx.web.thread.ThreadConfigBean">
<property name="businessId" value="xxxIndex"></property><!-- 业务ID，唯一 -->
<property name="max" value="40"></property><!-- 最好为请求线程的倍数 -->
<property name="min" value="10"></property><!-- 最好为请求线程的倍数 -->
<property name="queueCapacity" value="80"></property><!-- 最好为请求线程的倍数 -->
<property name="keepAliveTime" value="600"></property><!-- 线程空闲保存时间 -->
<property name="rejectHandler" ><!--任务拒绝处理策略 -->
<bean class="com.jd.m.pay.web.thread.RejectedPolicyHandler" >
<property name="bizName" value="pay-index"></property><!-- 业务ID，唯一 -->
</bean>
</property>
</bean>
<bean id="workers" class="java.util.ArrayList">
<constructor-arg >
<list>
<ref bean="xxxIndex"/>
</list>
</constructor-arg>
</bean>
线程工厂：
/**
* 线程工厂， 加入了线程名的业务描述
*
* @User: guopeng
* @Date: 2013-02-28
*/
public class DecorateThreadFactory implements ThreadFactory {
static final AtomicInteger poolNumber = new AtomicInteger(1);
final ThreadGroup group;
final AtomicInteger threadNumber = new AtomicInteger(1);
final String namePrefix;
public DecorateThreadFactory(final ThreadGroup group,final String bizName) {
this.group = group;
namePrefix = bizName + "-pool-" + poolNumber.getAndIncrement() + "-thread-";
}
public Thread newThread(Runnable r) {
Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
if (t.isDaemon())
t.setDaemon(false);
if (t.getPriority() != Thread.NORM_PRIORITY)
t.setPriority(Thread.NORM_PRIORITY);
return t;
}
}
监控线程：
/**
* 监控业务线程池运行情况
* @author yfguopeng
* @Date 2013-02-28
*/
public class MonitorThread implements Runnable {
private final static Log log = LogFactory.getLog(MonitorThread.class);
private final ThreadGroup group;
public MonitorThread(ThreadGroup group) {
this.group = group;
}
public void run() {
Map<String, ThreadPoolExecutor>workers = ThreadGroupUtil.getThreadWorkers();
Iterator<String> iterator = workers.keySet().iterator();
log.info("total threadpools:[ "+workers.size()+" ],total threads:[ "+group.activeCount()+" ]");
while(iterator.hasNext()) {
ThreadPoolExecutor worker = ThreadGroupUtil.getThreadWorker(iterator.next());
RejectedExecutionHandler handler = worker.getRejectedExecutionHandler();
String rejectedSize = "";
if (RejectedPolicyHandlerInteface.class.isAssignableFrom(handler.getClass())) {
rejectedSize = " ],rejected threads:[ "+((RejectedPolicyHandlerInteface) handler).getRejectedSize();
}
log.info("business name:[ "+worker.getBizName()+" ]" +
", core threads:[ "+worker.getCorePoolSize()+" ], max threads:[ "+worker.getMaximumPoolSize()+" ]" +
", queue capacitys:[ "+worker.getQueue().size()+" ], running threads:[ "+worker.getActiveCount()+"] " +
", reject threads:[ "+rejectedSize+" ], largest threads:[ "+worker.getLargestPoolSize()+" ], complete threads:[ "+worker.getCompletedTaskCount()+" ]");
}
}
}
线程拒绝处理器：
/**
* 线程拒绝执行控制球
* @author yfguopeng
* @Date 2013-02-28
*/
public class RejectedPolicyHandler extends ThreadPoolExecutor.AbortPolicy implements RejectedPolicyHandlerInteface{
private final static Log log = LogFactory.getLog(RejectedPolicyHandler.class);
private static AtomicLong totals = new AtomicLong(0l);
private String bizName;
public RejectedPolicyHandler(){}
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
String tip = "["+bizName+"] 线程忙，请求被拒绝.max: "+executor.getMaximumPoolSize()+", queue: "+executor.getQueue().size();
log.info(tip);
//业务报警 TODO
totals.addAndGet(1);
super.rejectedExecution(r, executor);
}
public String getBizName() {
return bizName;
}
public void setBizName(String bizName) {
this.bizName = bizName;
}
public long getRejectedSize() {
return totals.get();
}
}
import java.util.concurrent.RejectedExecutionHandler;
public interface RejectedPolicyHandlerInteface extends RejectedExecutionHandler{
public long getRejectedSize() ;
}
线程配置bean：
@SuppressWarnings("serial")
public class ThreadConfigBean implements Serializable{
/**
* 业务ID
*/
private String businessId;
/**
* 任务队列最大线程数
* 默认：80
*/
private Integer max = 160;
/**
* 任务队列最小线程数
* 默认：40
*/
private Integer min = 80;
/**
* 等待队列请求数
* 默认：300
*/
private Integer queueCapacity = 300;
/**
* 空闲线程存活时间
* 默认：3分钟
*/
private Long keepAliveTime = 3 * 60l;
/**
* 线程拒绝策略
*/
private RejectedExecutionHandler rejectHandler = new ThreadPoolExecutor.AbortPolicy();
public ThreadConfigBean() {
super();
}
public Integer getMax() {
return max;
}
public void setMax(Integer max) {
this.max = max;
}
public Integer getMin() {
return min;
}
public void setMin(Integer min) {
this.min = min;
}
public Integer getQueueCapacity() {
return queueCapacity;
}
public void setQueueCapacity(Integer queueCapacity) {
this.queueCapacity = queueCapacity;
}
public Long getKeepAliveTime() {
return keepAliveTime;
}
public void setKeepAliveTime(Long keepAliveTime) {
this.keepAliveTime = keepAliveTime;
}
public RejectedExecutionHandler getRejectHandler() {
return rejectHandler;
}
public void setRejectHandler(RejectedExecutionHandler rejectHandler) {
this.rejectHandler = rejectHandler;
}
public String getBusinessId() {
return businessId;
}
public void setBusinessId(String businessId) {
this.businessId = businessId;
}
}
线程组：
/**
* 各个业务获取响应线程池
* @author yfguopeng
*/
public class ThreadGroupUtil {
private static Map<String, ThreadPoolExecutor> threadworkers;
private static ScheduledExecutorService monitorThread;//监视线程
private static final long delay = 200l;
private static long cycle_default = 5000l;
static {
threadworkers = new ConcurrentHashMap<String, ThreadPoolExecutor>();
monitorThread = Executors.newScheduledThreadPool(1);
}
public static void addThreadWorker(String bizName,ThreadPoolExecutor executor){
threadworkers.put(bizName, executor);
}
public static ThreadPoolExecutor getThreadWorker(String bizName) {
return threadworkers.get(bizName);
}
public static Map<String, ThreadPoolExecutor> getThreadWorkers(){
return threadworkers;
}
public static ScheduledExecutorService getMonitorThread() {
return monitorThread;
}
public static void setMonitorThread(ScheduledExecutorService monitorThread) {
ThreadGroupUtil.monitorThread = monitorThread;
}
public static void monitorThreadClosed(){
if (monitorThread != null)
if (!monitorThread.isTerminated())
monitorThread.shutdown();
}
public static void monitorThreadStart(ThreadGroup group,Long cycle){
MonitorThread monitor = new MonitorThread(group);
if (cycle > 0l) {
try {
cycle_default = cycle;
} catch (Exception e) {
}
}
monitorThread.scheduleWithFixedDelay(monitor, delay, cycle_default, TimeUnit.MILLISECONDS);
}
}
线程池实现类：
/**
* 线程池
* @author yfguopeng
*
*/
public class ThreadPoolExecutor extends java.util.concurrent.ThreadPoolExecutor {
private String bizName;
public ThreadPoolExecutor(String bizName,int corePoolSize, int maximumPoolSize,
long keepAliveTime, TimeUnit unit,
BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
threadFactory, handler);
this.bizName = bizName;
}
public String getBizName() {
return bizName;
}
public void setBizName(String bizName) {
this.bizName = bizName;
}
}
web.xml配置：
<listener> <listener-class>xxx.xx.xx.xxx.web.thread.ThreadExecutorListener</listener-class> </listener>
调用：
ThreadPoolExecutor exc = ThreadGroupUtil.getThreadWorker("xxxIndex");
String payOrgInfo = null;
String cards = null;
Future<String> xxxFuture = null;
Future<String> yyyFuture = null;
long start = System.currentTimeMillis();
xxxTask xxxTask = new xxxTask(//参数);
yyyTask yyyTask = new yyyTask(//参数);
System.out.println("开始......");
xxxFuture = exc.submit(xxxTask );
yyyFuture = exc.submit(yyyTask );
try {
xxx= xxxFuture .get();
yyy= yyyFuture .get();
System.out.println(xxx);
System.out.println(yyy);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (ExecutionException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
long end = System.currentTimeMillis();
System.out.println("结束...... "+(end-start));
return "";
相关推荐
Spring Boot 自定义异步线程池的两种方式 Demo 下载，教程：https://www.ydyno.com/archives/1199.html
主要介绍了Django异步任务线程池实现原理,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
本程序详细介绍了线程和线程池的用法，使用多线程进行和异步编程实现数据库操作和日志的记录
1. SpringBoot 自定义线程池以及多线程间的异步调用（@Async、@EnableAsync） 2.Java多线程之定时任务 以及 SpringBoot多线程实现定时任务 3.@EnableScheduling 与 @Scheduled
这是一个图片加载器，使用线程池替代异步任务
主要给大家介绍了关于Spring Boot利用@Async如何实现异步调用：自定义线程池的相关资料，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友可以参考下
主要介绍了Spring Boot使用Spring的异步线程池的实现，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
这个小程序只是简单的使用线程池实现了libcurl的异步请求功能，很多地方还需要优化，之所有厚颜上传到这里，主要有有三个目的。 一个是回报这些天在CSDN上得到的帮助，希望可以帮助到后来的人； 二是感于libcurl编译...
异步+线程池+线程+加载网络图片.zip
这个类没有实现异步调用，只是一个同步操作。 2.也可以用ThreadPoolTaskExecutor结合FutureTask做到同步。 3.2. SyncTaskExecutor与ThreadPoolTaskExecutor区别 前者是同步执行器，执行任务同步，后者是线程池，执行...
andorid异步加载网络图片，在操作UI的的时候，不会影响UI阻塞，是系统运行看起来流畅
我们将在这里进一步讨论一些.NET类,以及他们在多线程编程中扮演的角色和怎么编程。它们是: System.Threading.ThreadPool 类 System.Threading.Timer 类 如果线程的数目并不是很多,而且你想控制每个线程的...
Java多线程实现数据切割批量执行，实现限流操作。 java线程池Executors实现数据批量操作。 批量异步Executors处理数据，实现限流操作，QPS限流。 线程池调用第三方接口限流实现逻辑。 案例适合： 1.批量处理大数据。...
定义全局线程池，将用户的请求放入自定义队列中，排队等候线程调用，等待超时则自动取消该任务，实现超时可取消的异步任务
本篇文章主要介绍了浅谈Spring @Async异步线程池用法总结，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
主要介绍了Python实现可设置持续运行时间、线程数及时间间隔的多线程异步post请求功能,涉及Python网络请求的创建、发送、响应、处理等相关操作技巧,需要的朋友可以参考下
c# 自定义线程事件 异步挂接主线程,支持单线程状态下的开始，暂停，继续，停止事件. 挂接主线程Progress 进度条同步定位. 实用版本. 有问题进群交流：52433292
为了给用户提供更好的体验，读取数据库等程序需要长时间来执行的操作，需要弹出一个等待界面。可以一边执行需要等待的操作，一边...本资源使用VS2017软件，VB.NET编程，异步委托和多线程编程实现背景透明的等待窗体。
主要介绍了spring boot使用自定义配置的线程池执行Async异步任务，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧