前面有一篇文章中，博主为大家介绍了**xxchat系统线程监控方案。今天博主为大家分享的是**微xxxx系统三大队列任务执行时所用到的线程池的不停服务器监控、暂停、启用、更改参数等等操作。

当然，在大多数情况下，小伙伴们可能都遇到的是允许线上停机改配置重启的操作。但是，如果你们的业务有很多人在用，老板要求不宕机的情况下对队列线程池进行动态的修改、重启、监控，那么此时我们就需要使用一些技术（Jmx、http等）来远程操控。博主今天主要分享的使用jmx来进行动态操作线程池的方案，在分享之前我们首先得了解jmx这项技术是什么？有什么用途？在什么时候用?下面博主就开始带大家一起去领略jmx高级监控技术的世界。

Jmx（Java Managerment Extensions，即"Java管理扩展"）,是一个为应用程序、设备、系统等植入管理功能的框架。Jmx可以跨越一系列异构操作系统平台、系统体系结构和网络传输协议，灵活的开发无缝隙集成的系统、网络和服务管理应用。 Jmx在Java编程语言中定义了应用程序以及网络管理和监控的体系结构、设计模式、应用程序接口以及服务。通常使用Jmx来监控系统的运行状态或管理系统的某些方面，比如清空缓存、重新加载配置文件等。

此处，**系统主要使用Jmx来注册自定义线程池组件，线程池组件上暴露了一系列的方法给Jmx；它们包括：线程池的停止，修改线程池的核心线程数、最大线程数、超时时间、队列大小等等参数，重启线程池等等方法。然后，客户可以使用Jmx管理工具（如：jconsole/ jvisualvm/程序Agent等等）连接上组件暴露的Jmx端口，然后调用其暴露的方法以实现具体的业务功能。

以下是**项目的自定义线程池组件代码：

1.Jmx组件接口CommonTaskManagerMBean

package com.xxxxxx.job.queue;
/*** * 类CommonTaskManagerMBean.java的实现描述：jmx组件接口* @author arron 2018年x月2x日 下午x:x2:x9*/
public interface CommonTaskManagerMBean {int getPageSize();void setPageSize(int pageSize);int getCorePoolSize();int getQueueSize();void setQueueSize(int queueSize);void setCorePoolSize(int corePoolSize);int getMaximumPoolSize() ;void setMaximumPoolSize(int maximumPoolSize);boolean isKill() ;public boolean isStop();/*** 立刻停掉* @return*/String stopTaskNow();/*** 优雅停掉* @return*/String stopTaskDely();/*** 开启任务线程池* @return*/String startTask();
}

2.Jmx组件接口CommonTaskManagerMBean的具体实现类，也是线程池组件实现类；       CommonTaskManager

package com.xxxxxx.job.queue;import java.lang.management.ManagementFactory;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;import javax.management.MBeanServer;
import javax.management.ObjectName;import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;/*** * 类CommonTaskManager.java的实现描述：通用线程池管理组件* * @author arron 2018年x月2x日 下午x:x2:x9*/
public class CommonTaskManager implements InitializingBean,CommonTaskManagerMBean {private static final Logger      logger        = LoggerFactory.getLogger(CommonTaskManager.class);private int                      corePoolSize=10;private int                      maximumPoolSize=20;private int                      pageSize=1000;private int                      queueSize=1000;//queue size 是否生效private boolean              iseffect=true;//第几次设置queuesizeprivate int                      opcount=0;//之前生效的queuesize大小private int                      oldQueueSize=0;//任务管理器名称private String                 taskManagerName="未命名"; private ThreadPoolExecutor executor=null;private volatile boolean kill=false;private volatile boolean stop=false;public boolean isKill() {return kill;}public void setKill(boolean kill) {this.kill = kill;}public boolean isStop() {return stop;}public String getTaskManagerName() {return taskManagerName;}public void setTaskManagerName(String taskManagerName) {if(StringUtils.isNotBlank(taskManagerName)){this.taskManagerName = taskManagerName;}else{logger.error("任务线程池管理器[{"+taskManagerName+"}]--setTaskManagerName()->设置线程池设置任务管理器名称执行失败,任务管理器名称不能为空");}}public int getPageSize() {return pageSize;}public void setPageSize(int pageSize) {if(pageSize>0){this.pageSize = pageSize;}else{logger.error("任务线程池管理器[{"+taskManagerName+"}]--setPageSize()->设置线程池分页处理size数执行失败,分页处理size数必须大于0");}}public int getCorePoolSize() {return corePoolSize;}public int getQueueSize() {return queueSize;}public void setQueueSize(int queueSize) {if(queueSize>0){this.queueSize = queueSize;}else{logger.error("任务线程池管理器[{"+taskManagerName+"}]--setQueueSize()->设置线程池队列size数执行失败,队列size数必须大于0");}opcount=opcount+1;if(opcount>1){iseffect=false;}}public void setCorePoolSize(int corePoolSize) {if(corePoolSize>=0){this.corePoolSize = corePoolSize;if(executor!=null&&!executor.isShutdown()){try{executor.setCorePoolSize(corePoolSize);}catch(Exception e){logger.error("任务线程池管理器[{"+taskManagerName+"}]--setCorePoolSize()->立刻设置线程池核心线程数执行失败",e);}               }}else{logger.error("任务线程池管理器[{"+taskManagerName+"}]--setCorePoolSize()->立刻设置线程池核心线程数执行失败,最大线程数必须大于或者等于0");}}public int getMaximumPoolSize() {return maximumPoolSize;}public void setMaximumPoolSize(int maximumPoolSize) {if(maximumPoolSize>0){this.maximumPoolSize = maximumPoolSize;if(executor!=null&&!executor.isShutdown()){if(maximumPoolSize>0){try{executor.setMaximumPoolSize(maximumPoolSize);}catch(Exception e){logger.error("任务线程池管理器[{"+taskManagerName+"}]--setMaximumPoolSize()->立刻设置线程池最大线程数执行失败",e);}     }}}else{logger.error("任务线程池管理器[{"+taskManagerName+"}]--setMaximumPoolSize()->立刻设置线程池最大线程数执行失败,最大线程数必须大于0");}}protected ThreadPoolExecutor getExecutor() {return executor;}/*** 当spring xml中定义的属性初始化完成后,执行该方法*/@Overridepublic void afterPropertiesSet() throws Exception {logger.warn("任务线程池管理器[{"+taskManagerName+"}]--执行初始化消息管理组件CommonTaskManager: init-afterPropertiesSet");//初始化线程池executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, 0L, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<Runnable>(queueSize));oldQueueSize=queueSize;}private ExecutorService pool;/*** 初始化加载上次未完成发送的消息继续发送;*/public void initMethod() {logger.warn("任务线程池管理器[{"+taskManagerName+"}]--执行初始化消息管理组件CommonTaskManager[{}]: init-method",taskManagerName);//注册到jmx管理MBeanServer beanServer = ManagementFactory.getPlatformMBeanServer();  try {ObjectName objectName = new ObjectName("CommonTaskManager:type=CommonTaskManager-"+taskManagerName);  beanServer.registerMBean(this, objectName);  pool = Executors.newSingleThreadExecutor();//创建实现了Runnable接口对象，Thread对象当然也实现了Runnable接口pool.submit(new Runnable() {@Overridepublic void run() {while (true) {try {logger.warn("任务线程池管理器[{"+taskManagerName+"}]--任务管理器中参数详情:corePoolSize:{},maximumPoolSize:{},pageSize:{},queueSize:{}",corePoolSize ,maximumPoolSize,pageSize,queueSize);if(executor!=null&&!executor.isShutdown()){if(!iseffect){logger.warn("任务线程池管理器[{"+taskManagerName+"}]--任务管理器中线程池的queueSize设置后还未重启，当前处于未生效状态,");}logger.warn("任务线程池管理器[{"+taskManagerName+"}]--任务管理器中线程池中参数详情:corePoolSize:{},maximumPoolSize:{},pageSize:{},queueSize:{},queue队列中任务数:{}",executor.getCorePoolSize() ,executor.getMaximumPoolSize(),pageSize,oldQueueSize,executor.getQueue().size());}else{logger.error("任务线程池管理器[{"+taskManagerName+"}]--任务管理器中线程池处于非工作状态!");}//休眠5秒钟Thread.sleep(30000l);} catch (InterruptedException e) {e.printStackTrace();}}}});} catch (Exception e) {logger.error("任务线程池管理器[{"+taskManagerName+"}]--initMethod()->立刻停止线程池执行失败",e);}}/*** 执行销毁线程池管理组件代码;*/public void destroyMethod() {logger.warn("任务线程池管理器[{"+taskManagerName+"}]--执行销毁消息管理组件destroy-method");shutdownAndAwaitTermination(executor);pool.shutdownNow();logger.warn("任务线程池管理器[{"+taskManagerName+"}]--执行销毁消息管理组件destroy-method...任务管理器完成销毁动作...");}/*** 线程池暴力销毁* @param pool*/private void shutdownNow() {try{executor.shutdownNow();logger.warn("任务线程池管理器[{"+taskManagerName+"}]--shutdownNow()->立刻停止线程池执行成功");}catch(Exception e){logger.error("任务线程池管理器[{"+taskManagerName+"}]--shutdownNow()->立刻停止线程池执行失败",e);}executor=null;}/*** 执行销毁线程池管理组件代码;*/private void shuntdownDely() {logger.warn("任务线程池管理器[{"+taskManagerName+"}]--执行销毁消息管理组件shuntdownDely-method");shutdownAndAwaitTermination(executor);logger.warn("任务线程池管理器[{"+taskManagerName+"}]--执行销毁消息管理组件shuntdownDely-method...任务管理器完成销毁动作...");}/*** 线程池优雅销毁* * @param pool*/private void shutdownAndAwaitTermination(ExecutorService pool) {try {if(pool.isShutdown()){logger.warn("任务线程池管理器[{"+taskManagerName+"}]--pool has been shutdown");}else{//让线程池无法提交新的任务pool.shutdown(); // Disable new tasks from being submitted// Wait a while for existing tasks to terminate//等待之前的存在的任务执行60s，然后停止if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {//立即停止正在执行的任务pool.shutdownNow(); // Cancel currently executing tasks// Wait a while for tasks to respond to being cancelledif (!pool.awaitTermination(60, TimeUnit.SECONDS)) {logger.warn("任务线程池管理器[{"+taskManagerName+"}]--Pool did not terminate");}}}} catch (InterruptedException ie) {// (Re-)Cancel if current thread also interruptedlogger.error("任务线程池管理器[{"+taskManagerName+"}]-停止时发生异常",ie);try{pool.shutdownNow();// Preserve interrupt status//Thread.currentThread().interrupt();}catch(Exception e){logger.error("任务线程池管理器[{"+taskManagerName+"}]-停止时发生异常",e);}}pool=null;}/*** 发起任务请求 submit Callable task request* @param <V>*/public <V> Future<V> submitCallableTaskRequest(Callable<V> callTask) {Future<V> result=null;try{result = submitTask(callTask);logger.warn("任务线程池管理器[{"+taskManagerName+"}]--submitCallableTaskRequest()->任务提交成功");}catch(Exception e){logger.error("任务线程池管理器[{"+taskManagerName+"}]--submitCallableTaskRequest()->任务提交失败",e);}return result;}/*** 发起任务请求 submit Runnable task request* @param <V>*/public void submitRunableTaskRequest(Runnable runTask) {try{excuteTask(runTask);logger.warn("任务线程池管理器[{"+taskManagerName+"}]--submitRunableTaskRequest()->任务提交成功");}catch(Exception e){logger.error("任务线程池管理器[{"+taskManagerName+"}]--submitRunableTaskRequest()->任务提交失败",e);}}protected <V> Future<V> submitTask(Callable<V> callTask) {Future<V> rs= executor.submit(callTask);return rs;}protected void excuteTask(Runnable runTask) {executor.execute(runTask);}@Overridepublic String stopTaskNow() {if(!this.kill){this.kill=true;}stop=true;shutdownNow();return  "success";}@Overridepublic String stopTaskDely() {if(!this.kill){this.kill=true;}stop=true;shuntdownDely();return "success";}@Overridepublic String startTask() {String rs="success";if(executor!=null&&!executor.isShutdown()){rs ="fail";logger.warn("任务线程池管理器[{"+taskManagerName+"}]--startTask()->线程池开启失败");}else{//初始化线程池executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, 0L, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<Runnable>(queueSize));stop=false;iseffect=true;oldQueueSize=queueSize;logger.warn("任务线程池管理器[{"+taskManagerName+"}]--startTask()->线程池开启成功");}return  rs;}}

3.spring Jmx线程池组件配置文件（可定义多个实例，进行不同的操作）

<bean id="synMemberTaskManager" class="com.xxxxxx.job.queue.CommonTaskManager" init-method="initMethod" destroy-method="destroyMethod"><property name="corePoolSize"  value="10" /><property name="maximumPoolSize" value="20" /><property name="pageSize" value="1000" /><property name="queueSize" value="1000"  /><property name="taskManagerName" value="member-taskManager"  />
</bean>
<bean id="qrcodeTaskManager" class="com.xxxxxxx.job.queue.CommonTaskManager" init-method="initMethod" destroy-method="destroyMethod"><property name="corePoolSize"  value="10" /><property name="maximumPoolSize" value="20" /><property name="pageSize" value="1000" /><property name="queueSize" value="1000"  /><property name="taskManagerName" value="qrcode-taskManager"  />
</bean>
<bean id="templateMsgTaskManager" class="com.xxxxxx.job.queue.CommonTaskManager" init-method="initMethod" destroy-method="destroyMethod"><property name="corePoolSize"  value="10" /><property name="maximumPoolSize" value="20" /><property name="pageSize" value="1000" /><property name="queueSize" value="1000"  /><property name="taskManagerName" value="templateMsg-taskManager"  />
</bean> 

4.定时任务队列中使用该线程池组件的案例SynMember

package com.xxxxxx.jobImpl;import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;import javax.annotation.Resource;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;import com.xxxxxx.job.queue.CommonTaskManager;
import com.xxxxxx.model.ThreadLastRuntime;
import com.xxxxxx.model.WccMemberCommon;
import com.xxxxxx.utils.HessianHelper;/*** 定时同步中间表的信息到会员表*
*/
@Lazy(false)
@Component
public class SynMember {private static final Logger logger = LoggerFactory.getLogger(SynMember.class);@Resourceprivate CommonTaskManager synMemberTaskManager;private  static ArrayBlockingQueue queue=  new ArrayBlockingQueue(1);//初始化线程池//每1分钟执行一次@Async@Scheduled(cron = "0 0/1 * * * ? ")public  void synMember(){//获取中间表未处理的数据if(queue.size()>0){logger.warn("前一次任务未执行完成跳过本次任务");return;}synMemberTaskManager.setKill(false);if(synMemberTaskManager.isStop()){logger.warn("会员同步线程池为启动,跳过本次任务");return;}try{queue.add(new Object());updateLastThreaRunTime();Integer size=synMemberTaskManager.getPageSize();List<WccMemberCommon> memberCommons = HessianHelper.getAppService().findMemberCommon(size);boolean rssiz=memberCommons != null && memberCommons.size() > 0;logger.warn("会员同步中间表未处理的大小==="+memberCommons.size());while(rssiz){updateLastThreaRunTime();if(synMemberTaskManager.isKill()||synMemberTaskManager.isStop()){logger.warn("synMember()-会员同步处理任务退出本次任务,");queue.clear();return ;}List<Future<Boolean>>  futureList=new ArrayList<Future<Boolean>>();logger.warn("synMember()-会员同步处理开始提交任务");for(WccMemberCommon memberCommon:memberCommons){if(synMemberTaskManager.isKill()||synMemberTaskManager.isStop()){break ;}Future<Boolean>  future=synMemberTaskManager.submitCallableTaskRequest(new SynMemberCallableThread(memberCommon));if(future!=null){futureList.add( future);}}updateLastThreaRunTime();logger.warn("synMember()-会员同步处理提交任务-结束-开始获取执行结果");if(futureList.size()>0){for(Future<Boolean>  item:futureList){try {if(synMemberTaskManager.isKill()||synMemberTaskManager.isStop()){logger.warn("synMember()-会员同步处理任务退出本次任务,");queue.clear();return ;}if(item!=null){if(item.isCancelled()){queue.clear();return;}if(item.isDone()){//如果已经完成,就不做任何事情}else{boolean r=item.get(30, TimeUnit.SECONDS);}}} catch (InterruptedException | ExecutionException e) {e.printStackTrace();logger.error("synMember--会员同步出现错误",e);}catch (Exception e) {logger.error("synMember--会员同步任务出现错误",e);}}}else{logger.warn("synMember()-会员同步处理提交任务成功数为0,线程池拒绝了任务");break;}updateLastThreaRunTime();memberCommons = HessianHelper.getAppService().findMemberCommon(size);logger.warn("synMember()-会员同步处理获取执行结果结束");logger.warn("synMember()-会员同步处理完成一轮,剩余size大小"+memberCommons.size());rssiz=memberCommons != null && memberCommons.size() > 0;}logger.warn("会员同步中间表未处理的大小==="+memberCommons.size()+",任务结束");}catch(Exception e){logger.error("会员同步中间表任务报错",e);}queue.clear();}public  void updateLastThreaRunTime(){int rscount= HessianHelper.getAppService().updateByThreadType("memberSync");ThreadLastRuntime threadLastRuntime=new ThreadLastRuntime();threadLastRuntime.setThreadType("memberSync");threadLastRuntime.setLastThreadRuntime(new Date());if(rscount==0){HessianHelper.getAppService().save(threadLastRuntime);}else if (rscount>1){HessianHelper.getAppService().deleteByThreadType("memberSync");HessianHelper.getAppService().save(threadLastRuntime);}}
}

5.Jmx配置（此处服务器是windows server 2012,所以配置是在tomcat注册文件中修改）

-Dcom.sun.management.jmxremote
-Dcom.sun.management.jmxremote.port=333111
-Dcom.sun.management.jmxremote.ssl=false
-Dcom.sun.management.jmxremote.authenticate=false

通过注册表修改
win+r   -->  regedit 打开注册表【有道深入理解jvm虚拟机】
打开 注册表HKEY_LOCAL_MACHINE\SOFTWARE\Wow6432Node\Apache Software Foundation\Procrun 2.0\Tomcat6\Parameters\Java(路径可能有一点点差别)中的Options。

6.项目组件启动效果图

7.Jmx连接效果图（此处我就用本地截图了）

最后，以上就是**项目系统监控操作线程池的全过程了，如果小伙伴们对Jmx这块的知识感兴趣，可以联系博主或者留言。后期博主将推出Jmx的整个技术系列的教程，敬请期待。