daemontaskexecutor.java.svn-base

监控整个SP系统的SMS信息通道的发送情况并且生成日志。 支持权限判断。

package com.infobank.superchannel.util;

import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

import org.apache.commons.configuration.Configuration;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.ConfigurationFactory;

import com.infobank.superchannel.daemon.GenericDaemon;
import com.infobank.superchannel.daemon.databasecache.DatabaseCacheDaemon;
import com.infobank.superchannel.daemon.messagelauncher.MessageLauncherDaemon;

public class DaemonTaskExecutor {
  private AtomicReference<ThreadPoolExecutor> taskExecutor=new AtomicReference<ThreadPoolExecutor>(null);

  public static boolean init(GenericDaemon daemon){
    if(instance.taskExecutor.get()!=null)
      return false;

        StringBuilder sb=new StringBuilder();
        if(daemon instanceof DatabaseCacheDaemon){
          sb.append("/daemon[@type='DatabaseCache']/task/executor/");
        }else{
          if(daemon instanceof MessageLauncherDaemon){
            String tag=((MessageLauncherDaemon)daemon).getTag();
            sb.append("/daemon[@type='MessageLauncher' and @tag='");
            sb.append(tag);
            sb.append("']/task/executor/");
          }
          else{
            throw new RuntimeException("invalid daemon type");
          }
        }

        String basePath=sb.toString();
    DaemonConfiguration config=DaemonConfiguration.getInstance();
    if(config==null)
      throw new RuntimeException("could not get task executor settings");

        sb=new StringBuilder();
        sb.append(basePath);
        sb.append("maxNumber");
    int maxExecutorNumber=config.getInt(sb.toString());

        sb=new StringBuilder();
        sb.append(basePath);
        sb.append("minNumber");
    int minExecutorNumber=config.getInt(sb.toString());

        sb=new StringBuilder();
        sb.append(basePath);
        sb.append("keepAliveTime");
    int keepAliveTime=config.getInt(sb.toString());

    instance.taskExecutor.set(new ThreadPoolExecutor(minExecutorNumber,
        maxExecutorNumber,keepAliveTime,
        TimeUnit.SECONDS,
        new LinkedBlockingQueue<Runnable>(),
        new ThreadPoolExecutor.CallerRunsPolicy())); 
    
    return true;
  }
  
  private static DaemonTaskExecutor instance=new DaemonTaskExecutor();
  
  public static DaemonTaskExecutor getInstance(){
    return instance;
  }
  
  public void addTask(Runnable task){
    taskExecutor.get().execute(task);
  }
  
  public void removeTask(Runnable task){
    taskExecutor.get().remove(task);
  }
  
  public long getTaskCount(){
    return taskExecutor.get().getTaskCount();
  }
  
  public static boolean fini(GenericDaemon daemon){
    instance.taskExecutor.get().shutdownNow();
    return true;
  }
}