스케쥴러에 의한 대용량 처리

 열심히 적었던 글이 소스코드를 옮겨 넣는 와중에 고향 친구로부터 전화가 모르고 저장. 그대로 날라가 버렸다. 슬프다. ㅠㅠ

같은 말을 반복하는걸 매우 싫어하는 성격이므로 기억을 더듬어 대략 다음의 내용들을 적어본다.

 

1. 대용량 처리를 위해서는 작업 정보를 관리하고 정확 모니터링 할 수 있는 스케쥴러가 필요하다.

2. 스케쥴러 또한 하나 이상의 쓰레드로 구성되므로, 쓰레드의 특성에 대해 정확하게 알고 있어야 한다.

3. 쓰레드 사용시 야기되는 문제들 가운데 Race Condition에 따른 해결방법, 뮤텍스, 이벤트 세마포어, 크리티컬 섹션, 등을 이해하고 있어야 한다.

4. 자바의 최신 SDK에는 기존에 사용되던 suspand(), resume() 과 같은 메서드가 디프레취 되었고, 쓰레드를 중단하거나 재시작하는 것과 같은 생명주기를 개발자가 임의로 컨트롤하기 보다는 이전 생명주기를 완전히 종료시키고 다시 해당 쓰레드의 생명주기를 시작하는 방법을 더 권장하는것 같다.

5. 스케쥴러는 작업 도중 예기치 않은 오류를 복구하는 기능과, 실시간으로 서버 시스템의 자원(메모리 등)을 모니터링 할 수 있는 기능을 가져야 한다.

6. log4j와 같은 로그 툴을 통해 로그를 잘 남겨두어야 한다.

7. 작업 정보를 저장하는 데이터 구조인, Queue, Stack, List 등의 특성과 사용법을 잘 이해하고 있어야 한다.

8. 스케쥴러는 시스템에 과도한 부하를 주는것을 방지하고 시스템 성능등과 같은 이슈를 피하고 단위 작업에 대한 무결성을 위해 자신에게 할당된 작업을 분할 처리 할 수 있는 기능을 가져야 한다.

 

[code java] /**
 * <PRE>
 * Class/Interface Name : WebSchedulerINI
 * History
 *     1. 이지홍(hongsgo@gmail.com), 2009. 1. 8., 최초작성
 * </PRE>
 * @brief 색인 서버의 전체 작업을 관리하는 핵심 클래스.
 * @date 2009. 1. 8.
 * @version 1.0.0
 * @author 이지홍(hongsgo@gmail.com)
 * @warning
 */
public class WebSchedulerINI extends BaseService implements Runnable {
 
 //스케쥴러 쓰레드
 private Thread webjobs = null;
 //스케쥴러의 모니터링을  담당하는 클래스
 private MonitorService monitorService = null;
 //실패한 작업을 복구하는 클래스
 private JobRecoveryManager jobRecoveryManager = null;
 //스케쥴러의 작업을 비우는 플래그 클래스(최우선)
 private boolean job_clear=false;
 //서버 시작 간을 표시하기 위한 부분 .
 public Date run_time = null;  /**
  * <PRE>
  * Method Name : setSchedulerJobClearFlag
  * History
  *  1. 이지홍(hongsgo@gmail.com), 2009. 2. 18., 최초작성
  * </PRE>
  * @brief 다음 작업에 스케쥴러를 비우는 플래그
  * @date 2009. 2. 18.
  * @version 1.0.0 void
  */
 public void setSchedulerJobClearFlag(){
  this.job_clear=true;
 }  /**
  * <PRE>
  * Method Name : setSchedulerJobClear
  * History
  *  1. 이지홍(hongsgo@gmail.com), 2009. 2. 18., 최초작성
  * </PRE>
  * @brief  플래그에 따라 큐를 비우는 메소드
  * @date 2009. 2. 18.
  * @version 1.0.0 void
  */
 private void setSchedulerJobClear(){
  this.monitorService.getSchedulerMonitorInfo().setSplit_job_count(0L);
  this.que_webSchedulerJob.clear();
  this.job_clear=false;
 }  /**
  * 임시 작업결과 정보
  */
 private DBIndexJobInfo temp_dbjob = null;
 private OperateJobInfo temp_operator_job = null;
 private MemoryManager memoryManager = null;
 private DBIndexService dbIndexService = null;
 //색인을 처리하는 객체의 부모클래스
 private BaseIndexService baseIndexService = null;
 private String[] temp_arr_heapinfo = null;
 private boolean index_optimize = false;
 private int now_SCHEDULER_INDEX_OPTIMIZE_COUNT = 0;
 
 //스케쥴러 작업의 최상위 부모 클래스
 private BaseWebJobInfo temp_basejob= null;
 private Queue<BaseWebJobInfo> que_webSchedulerJob = null;
//
 private ArrayList<String[]> arr_indexdata= null;  /* (non-Javadoc)
  * @see com.jce.searchengine.common.service.BaseService#init()
  */
 public void init(){
  this.webjobs= new Thread(this);
  this.que_webSchedulerJob=new PriorityBlockingQueue<BaseWebJobInfo>(100,new JobtypeCompare());
  this.run_time=new Date();
  webjobs.start();
 }
 public int getQue_Count(){
  return this.que_webSchedulerJob.size();
 }     /**
     * <PRE>
     * Class/Interface Name : JobtypeCompare
     * History
     *     1. 이지홍(hongsgo@gmail.com), 2009. 2. 6., 최초작성
     * </PRE>
     * @brief 작업의 우선순위를 기준으로 비교하는 Sorter
     * @date 2009. 2. 6.
     * @version 1.0.0
     * @author 이지홍(hongsgo@gmail.com)
     * @warning
     */
    class JobtypeCompare implements Comparator<BaseWebJobInfo> {      public int compare(BaseWebJobInfo o1, BaseWebJobInfo o2) {
     if(o1 instanceof BaseWebJobInfo && o2 instanceof BaseWebJobInfo){
      BaseWebJobInfo s1 = (BaseWebJobInfo)o1;
      BaseWebJobInfo s2 = (BaseWebJobInfo)o2;
              return (s1.getPriority() < s2.getPriority())? 1 : (s1.getPriority() == s2.getPriority() ? 0 : -1);
        }
        return -1;
  }
 }         /** (non-Javadoc)
  * @see java.lang.Runnable#run()
  * 주기적으로 rss,atom,메인 플래쉬 상단 xml을 퍼블리싱 하는 메서드 (1분간격)
  */
 public void run(){
  while(true)
  {
    //꺼내기전에 현재 큐를 비우는 플래그나 특정 사이트의 작업을 지우라는 플래그가 있는지 확인한다.
    //새 작업을 꺼낸다.
    if(job_clear){
     this.setSchedulerJobClear();
    }
    else
    {      this.temp_basejob=this.getNextJob();
     if(this.temp_basejob!=null)
     {
      switch(this.temp_basejob.getWebSchedulerJobType()){
       case DBIndex: //디비인덱스 작업일 경우 수행
        try
        {
        
          //보다 명확한 작업객체의 유효성 검사.
          if(this.temp_basejob instanceof DBIndexJobInfo){
           temp_dbjob = (DBIndexJobInfo)this.temp_basejob;
           if(temp_dbjob != null)
           {
            //작업 처리 여부 플래그를 활성화 시켜두고
            //디비 인덱스 작업의 타입에 따라 디비기반 인덱싱 작업 수행
            //해당 색인경로에 락이 남아있다면 락을 제거하여 준다.
            //데이터베이스로 부터 데이터를 가져오기 전에 log4j를 통해 info레벨의 log를 남긴다.
            //메모리 매니저를 통해 현재 가용중인 메모리의 정보를 갱신한다.            }
           else{
            throw new NullPointerException(super.glbConstant.getSCHEDULER_ERROR_6());
           }
          }
          //instanceof가 null인지 아닌지 확인하는 로직.
          else{
            //작업정보에는 데이터베이스 작업이라고 했지만, 실제 형변환을 해보니 아니더라. 그래서 예외 던짐.
            throw new ClassCastException(super.glbConstant.getSCHEDULER_ERROR_1());
          }
        } catch (Exception e) {
         this.monitorService.getSchedulerMonitorInfo().setIncFailJobCount(); //실패작업 카운팅
         if(this.temp_dbjob.getJobtype().equals(DBIndexJobType.DBIndexMakeAll)){
          //현재 처리중인 작업이 일괄색인이었다면 하위 분할된 작업값들을 초기화
          this.monitorService.getSchedulerMonitorInfo().setSplit_job_count(0);
         }
         if(e instanceof SQLException){
          if(!(this.temp_dbjob.getJobtype().equals(DBIndexJobType.DBIndexMakeAll))){
           this.jobRecoveryManager.setReadyToFailedJobSave(temp_dbjob);
          }
          logger.error(super.glbConstant.getSCHEDULER_ERROR_5()+ e.getMessage());
         }
         else if(e instanceof IOException){
          if(e instanceof CorruptIndexException){
           logger.error(super.glbConstant.getSCHEDULER_ERROR_3()+ e.getMessage());
          }
          else if(e instanceof LockObtainFailedException){
           try {
            this.dbIndexService.setDelLock(temp_dbjob.getService_site_info().getArr_list_index_location().get(0));
           } catch (IOException e1) {
            logger.error(e1.getMessage());
           }
           logger.error(super.glbConstant.getSCHEDULER_ERROR_1()+ e.getMessage());
          }
          else{
           logger.error(super.glbConstant.getSCHEDULER_ERROR_2()+ e.getMessage());
          }           if(!(temp_dbjob.getJobtype().equals(DBIndexJobType.DBIndexMakeAll))){
           this.jobRecoveryManager.setReadyToFailedJobSave(temp_dbjob);
          }
         }
         else if(e instanceof NullPointerException){
          logger.error(super.glbConstant.getSCHEDULER_ERROR_4()+e.getMessage());
         }
         else if(e instanceof ClassCastException){
          logger.error(super.glbConstant.getSCHEDULER_ERROR_1()+ e.getMessage());
         }
         else{
          logger.error(super.glbConstant.getSCHEDULER_ERROR_5()+ e.getMessage());
         }
        } finally{
         //플래그를 바꾼다.
         temp_dbjob.getService_site_info().setJobflag(false);
         this.setIndex_optimize(false);
         this.monitorService.getSchedulerMonitorInfo().setIncMonitorCount(temp_dbjob.getJobtype());
        }
        break;
       case Operate:
        //운영 툴 작업과 관련된게 있으면 처리
         try {
           if(this.temp_basejob instanceof OperateJobInfo){
            this.temp_operator_job = (OperateJobInfo)this.temp_basejob;
           }
           if(this.temp_operator_job != null)
           {
            switch(this.temp_operator_job.getJobtype()){
             case Clear:
               //로케이션이 하나라고 생각하고 처리.
               this.baseIndexService.setClearIndex(super.getIndexInfoLoaderService().getIndexInfoXmlSaxHandler().getHash_table_content_service_site().get(this.temp_operator_job.getSitecode()).getArr_list_index_location().get(0));
             break;
             case Optimize:
               this.baseIndexService.setOptimizeIndex(super.getIndexInfoLoaderService().getIndexInfoXmlSaxHandler().getHash_table_content_service_site().get(this.temp_operator_job.getSitecode()).getArr_list_index_location().get(0));
             break;
            }
           }
           else{
            //작업정보에는 데이터베이스 작업이라고 했지만, 실제 형변환을 해보니 아니더라. 그래서 예외 던짐.
            throw new ClassCastException(super.glbConstant.getSCHEDULER_ERROR_1());
           }
         } catch (Exception e) {
          logger.error(e.getMessage());
         } finally{
          this.monitorService.getSchedulerMonitorInfo().setIncMonitorCount(this.temp_operator_job.getJobtype());
         }
        break;
       case FieldInex:
        //쿼리 인덱스 혹은 필드 인덱스 작업이 있으면 수행.
        break;
       default:
        break;
      }//switch 문 끝       //현재 끄집어 낸 작업을 지우고
      this.que_webSchedulerJob.remove();
      //작업 처리 횟수를 올린다.
      this.monitorService.getSchedulerMonitorInfo().setIncTotJobCount();
      if(this.arr_indexdata!=null){
       this.arr_indexdata.clear();
      }
      this.arr_indexdata=null;
      this.temp_arr_heapinfo = null;
      this.temp_dbjob = null;
      this.temp_basejob=null;
     }
     else{// 작업이 없다면 다음 실행.
      try {
       //idle 횟수를 기억해 두었다가 최적화를 수행함
       if(now_SCHEDULER_INDEX_OPTIMIZE_COUNT>Integer.parseInt(super.getGlbConstant().getSCHEDULER_INDEX_OPTIMIZE_COUNT()) && this.isIndex_optimize()==false){
        this.now_SCHEDULER_INDEX_OPTIMIZE_COUNT=0;
        this.index_optimize=true;
       }
       else{
        now_SCHEDULER_INDEX_OPTIMIZE_COUNT++;
       }
      } catch (Exception e) {
       logger.error(this.glbConstant.getSCHEDULER_ERROR_8()+e.getMessage());
      } finally{       }      }
     this.temp_basejob = null;
     //if-else 문 끝
     try {
      if(this.que_webSchedulerJob.isEmpty()){ //작업 큐가 비었다면 idle
       Thread.sleep(this.glbConstant.getSCHEDULER_RELOAD_TIME());
      }
     } catch (InterruptedException e) {
      e.printStackTrace();
     }
    }   } //while문 끝
 }  public BaseWebJobInfo getNextJob(){   return this.que_webSchedulerJob.peek();
 }   [/code]