content: package com.xxl.job.admin.core.thread;
  
  import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
  import com.xxl.job.admin.core.model.XxlJobGroup;
  import com.xxl.job.admin.core.model.XxlJobRegistry;
  import com.xxl.job.core.biz.model.RegistryParam;
  import com.xxl.job.core.biz.model.ReturnT;
  import com.xxl.job.core.enums.RegistryConfig;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  import org.springframework.util.StringUtils;
  
  import java.util.*;
  import java.util.concurrent.*;
  
  /**
   * job registry instance
   *
   * @author xuxueli 2016-10-02 19:10:24
   */
  public class JobRegistryHelper {
  
  	private static Logger logger = LoggerFactory.getLogger(JobRegistryHelper.class);
  
  	private static JobRegistryHelper instance = new JobRegistryHelper();
  
  	public static JobRegistryHelper getInstance() {
  		return instance;
  	}
  
  	private ThreadPoolExecutor registryOrRemoveThreadPool = null;
  
  	private Thread registryMonitorThread;
  
  	private volatile boolean toStop = false;
  
  	public void start() {
  
  		// for registry or remove
  		registryOrRemoveThreadPool = new ThreadPoolExecutor(2, 10, 30L, TimeUnit.SECONDS,
  				new LinkedBlockingQueue<Runnable>(2000), new ThreadFactory() {
  					@Override
  					public Thread newThread(Runnable r) {
  						return new Thread(r,
  								"xxl-job, admin JobRegistryMonitorHelper-registryOrRemoveThreadPool-" + r.hashCode());
  					}
  				}, new RejectedExecutionHandler() {
  					@Override
  					public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
  						r.run();
  						logger.warn(
  								">>>>>>>>>>> xxl-job, registry or remove too fast, match threadpool rejected handler(run now).");
  					}
  				});
  
  		// for monitor
  		registryMonitorThread = new Thread(new Runnable() {
  			@Override
  			public void run() {
  				while (!toStop) {
  					try {
  						// auto registry group
  						List<XxlJobGroup> groupList = XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao()
  								.findByAddressType(0);
  						if (groupList != null && !groupList.isEmpty()) {
  
  							// remove dead address (admin/executor)
  							List<Integer> ids = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao()
  									.findDead(RegistryConfig.DEAD_TIMEOUT, new Date());
  							if (ids != null && ids.size() > 0) {
  								XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().removeDead(ids);
  							}
  
  							// fresh online address (admin/executor)
  							HashMap<String, List<String>> appAddressMap = new HashMap<String, List<String>>();
  							List<XxlJobRegistry> list = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao()
  									.findAll(RegistryConfig.DEAD_TIMEOUT, new Date());
  							if (list != null) {
  								for (XxlJobRegistry item : list) {
  									if (RegistryConfig.RegistType.EXECUTOR.name().equals(item.getRegistryGroup())) {
  										String appname = item.getRegistryKey();
  										List<String> registryList = appAddressMap.get(appname);
  										if (registryList == null) {
  											registryList = new ArrayList<String>();
  										}
  
  										if (!registryList.contains(item.getRegistryValue())) {
  											registryList.add(item.getRegistryValue());
  										}
  										appAddressMap.put(appname, registryList);
  									}
  								}
  							}
  
  							// fresh group address
  							for (XxlJobGroup group : groupList) {
  								List<String> registryList = appAddressMap.get(group.getAppname());
  								String addressListStr = null;
  								if (registryList != null && !registryList.isEmpty()) {
  									Collections.sort(registryList);
  									StringBuilder addressListSB = new StringBuilder();
  									for (String item : registryList) {
  										addressListSB.append(item).append(",");
  									}
  									addressListStr = addressListSB.toString();
  									addressListStr = addressListStr.substring(0, addressListStr.length() - 1);
  								}
  								group.setAddressList(addressListStr);
  								group.setUpdateTime(new Date());
  
  								XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().update(group);
  							}
  						}
  					}
  					catch (Exception e) {
  						if (!toStop) {
  							logger.error(">>>>>>>>>>> xxl-job, job registry monitor thread error:{}", e);
  						}
  					}
  					try {
  						TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);
  					}
  					catch (InterruptedException e) {
  						if (!toStop) {
  							logger.error(">>>>>>>>>>> xxl-job, job registry monitor thread error:{}", e);
  						}
  					}
  				}
  				logger.info(">>>>>>>>>>> xxl-job, job registry monitor thread stop");
  			}
  		});
  		registryMonitorThread.setDaemon(true);
  		registryMonitorThread.setName("xxl-job, admin JobRegistryMonitorHelper-registryMonitorThread");
  		registryMonitorThread.start();
  	}
  
  	public void toStop() {
  		toStop = true;
  
  		// stop registryOrRemoveThreadPool
  		registryOrRemoveThreadPool.shutdownNow();
  
  		// stop monitir (interrupt and wait)
  		registryMonitorThread.interrupt();
  		try {
  			registryMonitorThread.join();
  		}
  		catch (InterruptedException e) {
  			logger.error(e.getMessage(), e);
  		}
  	}
  
  	// ---------------------- helper ----------------------
  
  	public ReturnT<String> registry(RegistryParam registryParam) {
  
  		// valid
  		if (!StringUtils.hasText(registryParam.getRegistryGroup())
  				|| !StringUtils.hasText(registryParam.getRegistryKey())
  				|| !StringUtils.hasText(registryParam.getRegistryValue())) {
  			return new ReturnT<String>(ReturnT.FAIL_CODE, "Illegal Argument.");
  		}
  
  		// async execute
  		registryOrRemoveThreadPool.execute(new Runnable() {
  			@Override
  			public void run() {
  				int ret = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().registryUpdate(
  						registryParam.getRegistryGroup(), registryParam.getRegistryKey(),
  						registryParam.getRegistryValue(), new Date());
  				if (ret < 1) {
  					XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().registrySave(
  							registryParam.getRegistryGroup(), registryParam.getRegistryKey(),
  							registryParam.getRegistryValue(), new Date());
  
  					// fresh
  					freshGroupRegistryInfo(registryParam);
  				}
  			}
  		});
  
  		return ReturnT.SUCCESS;
  	}
  
  	public ReturnT<String> registryRemove(RegistryParam registryParam) {
  
  		// valid
  		if (!StringUtils.hasText(registryParam.getRegistryGroup())
  				|| !StringUtils.hasText(registryParam.getRegistryKey())
  				|| !StringUtils.hasText(registryParam.getRegistryValue())) {
  			return new ReturnT<String>(ReturnT.FAIL_CODE, "Illegal Argument.");
  		}
  
  		// async execute
  		registryOrRemoveThreadPool.execute(new Runnable() {
  			@Override
  			public void run() {
  				int ret = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().registryDelete(
  						registryParam.getRegistryGroup(), registryParam.getRegistryKey(),
  						registryParam.getRegistryValue());
  				if (ret > 0) {
  					// fresh
  					freshGroupRegistryInfo(registryParam);
  				}
  			}
  		});
  
  		return ReturnT.SUCCESS;
  	}
  
  	private void freshGroupRegistryInfo(RegistryParam registryParam) {
  		// Under consideration, prevent affecting core tables
  	}
  
  }
  