[ABAP] 병렬 처리(Parallel Processing)

안녕하세요.

 

CBO 프로그램을 개발하다보면 대량의 데이터를 처리해야하는 경우가 생깁니다. 

대량의 문서에 액션을 취한다거나 데이터를 가져온다거나 그러다보면 덤프가 발생하기도 하고 시간이 오래 걸리기도 하는데요. 이럴 때 사용할 수 있는 방법이 병렬처리입니다.

 

병렬처리 

병렬 처리는 하나의 메인 프로그램에서 서브 프로세스를 여러개 만들어서 동시에 실행하여 처리하는 방식입니다.

서브 프로세스가 여러개 실행이 된다는 것은 여러 사람이 동일 작업을 진행하는 것과 같다고 보시면 되는데요.

만약 10개의 서브 프로세스를 만들어 10개씩 할당한다면 10사람이 10개의 작업을 맡아 하는 것과 같은 효과를 볼 수 있습니다. 

 

병렬처리 주의점

 

주의해야할 점도 따로 있으니 꼭 확인해서 개발하세요!

1. 최대 프로세스를 제한해야합니다.

2. 한 서브 프로세스가 다른 서브 프로세스 로직에 영향을 주면 안됩니다. 

(하나는 생성하고 하나는 삭제하고 이러면 안된다는 의미!)

3. 각 프로세스에 데이터가 중복되지 않도록 합니다.

4. 특정 프로세스에 문제가 발생하더라도 확인할 수 있는 로직을 추가해야합니다.

 

병렬처리 참고 소스

 

TOP

DATA: GV_JOBS         TYPE I,  "Number of parallel jobs "<<<<<<<<
      GV_SND_JOBS     TYPE I VALUE 1,  "Work packets sent for processing
      GV_RCV_JOBS     TYPE I VALUE 1,  "Work packet replies received
      GV_EXCP_FLAG(1) TYPE C.  "RESOURCE_FAILURE

DATA: GV_TASKNAME(4) TYPE N.  "Task name

 

  PERFORM init_pbt_environment

FORM init_pbt_environment .


  gv_group = sy-sysid. 

  CLEAR: gv_wp_total, gv_wp_available.
  CALL FUNCTION 'SPBT_INITIALIZE'
    EXPORTING
      group_name                     = gv_group
    IMPORTING
      max_pbt_wps                    = gv_wp_total
      free_pbt_wps                   = gv_wp_available
    EXCEPTIONS
      invalid_group_name             = 1
      internal_error                 = 2
      pbt_env_already_initialized    = 3
      currently_no_resources_avail   = 4
      no_pbt_resources_found         = 5
      cant_init_different_pbt_groups = 6
      OTHERS                         = 7.

  CASE sy-subrc.
    WHEN 0.
      "Everything’s ok.
    WHEN 1.
      MESSAGE 'Paralell group not defined; See transaction RZ12'
      TYPE 'E'.
    WHEN 2.
      MESSAGE 'SAP system error: see the system log (transaction SM21)'
      TYPE 'E'.
    WHEN 3.
*     MESSAGE 'PBT environment was already initialized.' TYPE 'W'.
      CALL FUNCTION 'SPBT_GET_CURR_RESOURCE_INFO'
        IMPORTING
          max_pbt_wps                 = gv_wp_total
          free_pbt_wps                = gv_wp_available
        EXCEPTIONS
          internal_error              = 1
          pbt_env_not_initialized_yet = 2
          OTHERS                      = 3.
    WHEN 4.
      MESSAGE '현재 가용한 시스템 리소스가 없습니다.'
      TYPE 'E'.
    WHEN 5.
      MESSAGE 'Check your servers, network, operation modes.'
      TYPE 'E'.
    WHEN 6.
  ENDCASE.

*  GV_JOBS = GV_WP_AVAILABLE.

  IF gv_wp_available >= 5.      "5개만 사용하도록 설정 .
    gv_jobs = 5.
  ELSE.
    gv_jobs = gv_wp_available.
  ENDIF.

ENDFORM.

 

 

 

  PERFORM init_taskname.

FORM init_taskname .
  gv_taskname = '0001'.
ENDFORM.

 

 

PERFORM sapgui_progress_indicator

FORM sapgui_progress_indicator
  USING pv_text1
        pv_text2
        pv_text3
        pv_text4
        pv_text5
        pv_text6
        pv_text7.


  DATA: lv_text1 TYPE string,
        lv_text2 TYPE string,
        lv_text3 TYPE string,
        lv_text4 TYPE string,
        lv_text5 TYPE string,
        lv_text6 TYPE string,
        lv_text7 TYPE string,
        lv_text  TYPE string.

  lv_text1 = pv_text1.
  lv_text2 = pv_text2.
  lv_text3 = pv_text3.
  lv_text4 = pv_text4.
  lv_text5 = pv_text5.
  lv_text6 = pv_text6.
  lv_text7 = pv_text7.

  CONCATENATE lv_text1
              lv_text2
              lv_text3
              lv_text4
              lv_text5
              lv_text6
              lv_text7 INTO lv_text.

  CALL FUNCTION 'SAPGUI_PROGRESS_INDICATOR'
    EXPORTING
      text = lv_text.

ENDFORM.

perform paralell_receive

FORM paralell_receive USING pv_taskname.

  DATA : lt_temp LIKE TABLE OF gt_temp WITH HEADER LINE .

  RECEIVE RESULTS FROM FUNCTION '병렬 function'
    TABLES
      et_temp = lt_temp
    EXCEPTIONS
      communication_failure = 1
      system_failure        = 2.

  gv_rcv_jobs = gv_rcv_jobs + 1.  "Receiving data

  APPEND LINES OF lt_temp TO gt_temp[].

ENDFORM.

  PERFORM init_pbt_environment.
  PERFORM init_taskname.

  CLEAR: gv_snd_jobs,
         gv_rcv_jobs,
         gv_excp_flag.

*--------------------------------------------------------------------*

  DATA : lv_vbeln   TYPE vbeln,
         lv_type    TYPE bapi_mtype,
         lv_message TYPE bapi_msg.

  DATA : ls_input TYPE zsds1650,
         ls_item  TYPE zsds1651.
  DATA : lt_item  TYPE TABLE OF zsds1651.

** 처리할 DATA ITAB
  DATA : lt_temp     LIKE TABLE OF gt_temp WITH HEADER LINE .
  DATA : lt_temp_div LIKE TABLE OF gt_temp WITH HEADER LINE .


  " 병렬 라인수 제어
  "---------------------------------------------------------------------
  DATA: lv_tot_lines TYPE i.
  CLEAR lv_tot_lines.
  DESCRIBE TABLE lt_temp LINES lv_tot_lines.

  CHECK lv_tot_lines > 0.

  DATA: lv_lines_f TYPE f.
  DATA: lv_lines TYPE i.
  CLEAR lv_lines_f.
  lv_lines_f = lv_tot_lines / gv_jobs.

  CLEAR lv_lines.
  CALL FUNCTION 'ROUND'
    EXPORTING
      decimals = 0
      input    = lv_lines_f
      sign     = '+' "올림
    IMPORTING
      output   = lv_lines.

  DATA: lv_from_index TYPE i.
  DATA: lv_to_index TYPE i.

  CLEAR: lv_from_index, lv_to_index.
  lv_from_index = 1.
  IF lv_lines >= lv_tot_lines.
    lv_to_index = lv_tot_lines.
  ELSE.
    lv_to_index = lv_lines.
  ENDIF.

  "---------------------------------------------------------------------
  " 병렬 처리 시작
  "---------------------------------------------------------------------
  PERFORM sapgui_progress_indicator
    USING '병렬 프로세스를 시작합니다. (Avg. Lines per Job : '
          lv_lines
          ', Available Work Process : '
          gv_jobs
          ')'
          ''
          ''.

  DO.
    CLEAR: lt_temp_div[].
    APPEND LINES OF lt_temp FROM lv_from_index TO lv_to_index
                 TO lt_temp_div.
    CLEAR gv_msg.
    CALL FUNCTION ' 병렬 관련 펑션 ' "--- RFC 형태
      STARTING NEW TASK gv_taskname
      DESTINATION IN GROUP gv_group
      PERFORMING paralell_receive ON END OF TASK
      EXPORTING
        ....
      TABLES
       ....
      EXCEPTIONS
        communication_failure = 1 MESSAGE gv_msg
        system_failure        = 2 MESSAGE gv_msg
        resource_failure      = 3.

    CASE sy-subrc.
      WHEN 0.
        gv_taskname = gv_taskname + 1.
        gv_snd_jobs = gv_snd_jobs + 1.

        gv_jobs  = gv_jobs - 1. "Number of existing gv_jobs

        "---------------------------------------------------------------
        " 병렬 라인수 제어
        "---------------------------------------------------------------
        IF lv_to_index = lv_tot_lines.
          EXIT. "모든 라인에 대하여 병렬 호출 종료
        ENDIF.

        lv_from_index = lv_to_index + 1.
        lv_to_index = lv_lines + lv_from_index - 1.

        IF lv_to_index > lv_tot_lines     "TO_INDEX가 전체라인을 초과
        OR 0 = gv_jobs - 1.               "다음 DO문이 마지막인 경우
          lv_to_index = lv_tot_lines.
        ENDIF.
        "---------------------------------------------------------------

      WHEN 1.
        MESSAGE i001(zhco) WITH gv_msg ' (Communication Failure)'.

      WHEN 2.
        MESSAGE i001(zhco) WITH gv_msg ' (System Failure)'.

      WHEN 3.
        IF gv_excp_flag = space.
          gv_excp_flag = 'X'.
          "First attempt at RESOURCE_FAILURE handling.
*          MESSAGE I001(ZHCO)
*          WITH 'All servers currently busy (Resource Failure) : 1초 대기'.
          WAIT UNTIL gv_rcv_jobs >= gv_snd_jobs UP TO '1' SECONDS.

        ELSE.
          "Second attempt at RESOURCE_FAILURE handling
*          MESSAGE I001(ZHCO)
*          WITH 'All servers currently busy (Resource Failure) : 5초 대기'.
          WAIT UNTIL gv_rcv_jobs >= gv_snd_jobs UP TO '5' SECONDS.

          IF sy-subrc = 0.
            CLEAR gv_excp_flag.
          ELSE.
            "Endless loop handling.........................
          ENDIF.
        ENDIF.
    ENDCASE.
  ENDDO.

  DO.
    PERFORM sapgui_progress_indicator
      USING '병렬 프로세스 수행중... (Avg.Lines/Job : '
            lv_lines
            ', Sended : '
            gv_snd_jobs
            ', Received : '
            gv_rcv_jobs
            ')'.

    WAIT UNTIL gv_rcv_jobs >= gv_snd_jobs UP TO 1 SECONDS.

    IF sy-subrc = 0.
      EXIT.
    ENDIF.
  ENDDO.


*후처리----------------------------------------------------------------*

  ENDLOOP.

STARTING NEW TASK '태스크 명'	- 각 태스크 별 이름 다르게 처리 ( 변수로 설정 多 )
DESTINATION IN GROUP	- 어떤 RFC 서버 그룹에서 처리할 지 (DEFAULT 이거나 없는 경우 기본 시스템에서 처리) - T-CODE RZ12에서 관리함
PERFORMING [후속 함수] END OF TASK	- 하나의 프로세스가 종료된 후 후속 처리될 후속 함수 - 후속 함수에서 Importing을 수행