workflowsessionmanager.java

公司自己开发的工作流引擎

package cn.com.iaspec.workflow.util;

import java.util.*;
import java.util.Map.*;
import org.apache.log4j.*;
import cn.com.iaspec.workflow.exception.*;
import cn.com.iaspec.workflow.vo.workflow.*;
import com.sunyard.sunflow.client.*;

/**
 *
 * <p>Title: 工作流线程池管理</p>
 * <p>Description: </p>
 * <p>Copyright: Copyright (c) 2005</p>
 * <p>Company: IASPEC Technologies</p>
 * @author xiesonglin
 * @version 1.0
 */
public class WorkflowSessionManager{
  private static Logger logger=Logger.getLogger(WorkflowSessionManager.class);
  private static WorkflowSessionManager instance=null;
  private static HashMap usedSessionPool=new HashMap();
  private static HashMap freeSessionPool=new HashMap();

  //连接超期时间
  private static long TIME_OUT=3000; //3second

  //session失效时间
  private static long SESSION_TIME_OUT=1800000; //30minute

  //最大线程数
  private static long MAX_SESSION=4;
  private static long workflowFreeCount=0;
  private static long workflowConnectCount=0;

  private WorkflowSessionManager(){
  }

  public static WorkflowSessionManager getInstance(){
    if(instance==null){
      instance=new WorkflowSessionManager();
      //创建session服务线程
      new WorkflowSessionThread().start();
    }
    return instance;
  }

  /**
   * 检查session是否已经失效，如果最后的使用时间超过30分钟，则将该session删除。
   */
  public synchronized void checkSessionTimeOut(){
    //logger.info("begin checkSessionTimeOut...");
    Set entry=freeSessionPool.entrySet();
    Iterator it=entry.iterator();
    while(it.hasNext()){
      Entry me=(Entry)it.next();
      WorkflowSessionClient client=(WorkflowSessionClient)freeSessionPool.get(
          me.getKey());
      if(client!=null&&
          System.currentTimeMillis()-client.getLastFreeTime()>=SESSION_TIME_OUT){
        //断开连接,删除引用
        logger.info("session thread task begin delete timeout session...");
        this.disconnectWorkflowClient(client.getSessionId());
      }
    }
  }

  /**
   * 取得工作流session
   * @param userInfo UserInfo
   * @throws Exception
   * @return SunflowClient
   */
  private SunflowClient getWorkflowClientEx(UserInfo userInfo)
      throws Exception{
    logger.info("begin getWorkflowClientEx...");
    SunflowClient sunClient=null;
    if(freeSessionPool.containsKey(userInfo.getSessionId())){
      if(freeSessionPool.get(userInfo.getSessionId())!=null){
        WorkflowSessionClient client=(WorkflowSessionClient)freeSessionPool.get(
            userInfo.getSessionId());
        sunClient=client.getSunflowClient();
      }
    }
    else{
      //客户端数已达最大值
      if(freeSessionPool.size()+usedSessionPool.size()>=MAX_SESSION){
        //将最早闲置的session释放，并建立新的session
        sunClient=getReleaseAndNewSession(userInfo);
      }
      else{
        sunClient=connectWorkflowClient(userInfo);
      }
    }
    return sunClient;
  }

  /**
   * 将最早闲置的session释放，并建立新的session
   * @return String
   */
  private SunflowClient getReleaseAndNewSession(UserInfo userInfo)
      throws Exception{
    logger.info("begin getReleaseAndNewSession...");
    SunflowClient sunClient=null;
    String sessionId="";
    long startTime=System.currentTimeMillis();
    long count=0;
    Set entry=freeSessionPool.entrySet();
    Iterator it=entry.iterator();
    while(it.hasNext()){
      Entry me=(Entry)it.next();
      WorkflowSessionClient client=(WorkflowSessionClient)freeSessionPool.get(
          me.getKey());
      if(client!=null&&startTime-client.getLastFreeTime()>count){
        sessionId=client.getSessionId();
        count=startTime-client.getLastFreeTime();
      }
    }
    if(sessionId!=null&&!sessionId.equals("")){
      //断开连接
      disconnectWorkflowClient(sessionId);
      //创建新的连接
      sunClient=connectWorkflowClient(userInfo);
    }
    return sunClient;
  }

  /**
   * 取得工作流session
   * @param userInfo UserInfo
   * @throws Exception
   * @return SunflowClient
   */
  public synchronized SunflowClient getWorkflowClient(UserInfo userInfo)
      throws Exception{
    logger.info("begin getWorkflowClient...");
    long startTime=System.currentTimeMillis();
    SunflowClient client=null;
    while(client==null&&System.currentTimeMillis()-startTime<TIME_OUT){
      client=getWorkflowClientEx(userInfo);
    }
    if(client==null){
      throw new GetWorkflowSessionException("不能连接到工作流引擎，工作流引擎客户端数量已达最大值。");
    }
    this.workflowConnectCount++;
    logger.info("workflowConnectCount is:"+workflowConnectCount);
    logger.info("Execution getWorkflowClient time: "+(System.currentTimeMillis()-
        startTime)+" ms.");
    return client;
  }

  /**
   * 释放工作流线程，将线程加入闲置的线程池
   * @param userInfo UserInfo
   */
  public synchronized void freeWorkflowClient(UserInfo userInfo){
    long startTime=System.currentTimeMillis();
    logger.info("begin freeWorkflowClient...");
    WorkflowSessionClient client=(WorkflowSessionClient)usedSessionPool.get(
        userInfo.getSessionId());
    if(client!=null){
      client.setLastFreeTime(System.currentTimeMillis());
      //加入闲置的线程池
      freeSessionPool.put(userInfo.getSessionId(),client);
      //从使用的线程池中删除
      usedSessionPool.remove(userInfo.getSessionId());
      this.workflowFreeCount++;
      logger.info("workflowFreeCount is:"+workflowFreeCount);
    }
    logger.info("Execution freeWorkflowClient time: "+(System.currentTimeMillis()-
        startTime)+" ms.");
  }

  /**
   * 新建工作流session
   * @param userInfo UserInfo
   * @throws Exception
   * @return SunflowClient
   */
  private synchronized SunflowClient connectWorkflowClient(UserInfo userInfo)
      throws Exception{
    logger.info("begin connectWorkflowClient...");
    long startTime=System.currentTimeMillis();
    if(freeSessionPool.size()+usedSessionPool.size()>=MAX_SESSION){
      return null;
    }
    SunflowClient flowClient=null;
    try{
      flowClient=WorkflowUtil.getSunflowClient();
      long startTime1=System.currentTimeMillis();
      flowClient.connect(userInfo.getLoginName(),userInfo.getPassword());
      logger.info("Execution connect WorkflowClient use time:"+
          (System.currentTimeMillis()-startTime1)+"ms.");
      WorkflowSessionClient wfClient=new WorkflowSessionClient();
      wfClient.setSessionId(userInfo.getSessionId());
      wfClient.setSunflowClient(flowClient);
      wfClient.setLastFreeTime(System.currentTimeMillis());
      //加入已用线程池
      usedSessionPool.put(userInfo.getSessionId(),wfClient);
      logger.info("success connectWorkflowClient use time:"+
          (System.currentTimeMillis()-startTime)+"ms.");
    }
    catch(Exception e){
      e.printStackTrace();
      throw new CannotLinkToSunflowServerException("不能连接到工作流引擎。");
    }
    logger.info("Execution "+userInfo.getLoginName()+
        "'s connSunflowClient time: "+(System.currentTimeMillis()-startTime)+
        " ms.");
    return flowClient;
  }

  /**
   * 断开工作流线程
   * @param sessionId String
   * @throws Exception
   */
  private synchronized void disconnectWorkflowClient(String sessionId){
    long startTime=System.currentTimeMillis();
    logger.info("begin disconnectWorkflowClient...");
    if(sessionId!=null&&!sessionId.equals("")){
      if(freeSessionPool.containsKey(sessionId)){
        WorkflowSessionClient client=(WorkflowSessionClient)freeSessionPool.get(
            sessionId);
        try{
          client.getSunflowClient().disconnect();
        }
        catch(Exception ex){
          ex.printStackTrace();
          logger.info("disconnect workflow client error...");
        }
        freeSessionPool.remove(sessionId);
        logger.info("sessionId:"+sessionId+
            " success disconnectWorkflowClient use time:"+
            (System.currentTimeMillis()-startTime)+"ms.");
      }
    }
  }

  /**
   * 将session断开连接，从线程池中删除
   * @param userInfo UserInfo
   */
  public synchronized void logoutWorkflowClient(UserInfo userInfo){
    long startTime=System.currentTimeMillis();
    logger.info("begin logoutWorkflowClient...");
    this.disconnectWorkflowClient(userInfo.getSessionId());
    logger.info("Execution logoutWorkflowClient time: "+
        (System.currentTimeMillis()-startTime)+" ms.");
  }

}