템플릿 메서드 패턴이란

객체지향 디자인 패턴 중 하나, 기능의 뼈대(템플릿)과 실제 구현을 분리하는 패턴이다. 스프링으로 개발할때 몇번 공부했던 기본적인 패턴 중 하나로, 스프링에서는 보통 공통되는 기능에 대해 interface를 만든 뒤, 이를 구현하는 구현 클래스들을 여럿 생성한다. 실제 객체를 생성하는 부분에서는 코드 상 호출은 인터페이스를 호출하지만 실제 생성되는 객체는 여러 구현 클래스 중 필요한 클래스를 기반으로 객체를 생성한다.

상위 클래스와 하위 클래스로 나뉘는데, 이때 하위 클래스는 상위 클래스가 정의한 메서드만을 구현하게된다. 즉 어떤것을 구현하는지는 상위클래스에서 지정된다.

실제 내가 개발했던 코드를 보자.

이런 식으로 구현이 돼있다. 기존의 다른 코드의 모양새를 참고한 부분이고, 아직 네이버 외에는 개발이 안되어 있어서 if 문으로 naver의 경우에만 처리를 하고 다른 포탈의 경우에는 pass 하도록 되어있다.

생각해보면, 다른 포탈이 개발되어 else에 해당 포탈을 추가하게 되면 위의 코드를 똑같이 붙여넣기 해야하고 따라서 코드 중복의 문제가 생긴다. 이 때문에, 아직 개발이 되었든 되지 않았든 동일한 코드만 사용하도록 변경이 필요하다.

다음은 동료가 수정해준 코드이다.

MAPPER를 통해 어떤 포탈인지 가져오도록 하였다. 다른 포탈이 추가가 되면 이 mapper만 수정해주면 되며, 어떤 포탈이든 동일한 코드를 타게 됨으로써 코드의 중복이 확연하게 줄었다.

다음을 보자

NaverSearchingGateway는 KeywordSearchingBaseGateway를 상속한다. 상속을 통해 NaverSearchingGateway 객체가 생성된 곳에서 필요한 경우 KeywordSearchingBaseGateway의 메서드를 갖다 쓰는 방식으로 구현을 했는데.. 잘 보면 NaverSearchingGateway와 KeywordSearchingBaseGateway에 동일한 메서드가 전무하다. 즉 오버라이딩이 일절 되어있지 않다는것.

위에서 말했다시피 템플릿 메서드 패턴은 하위클래스는 상위클래스가 정의한 메서드를 쓰게 함으로써 상위클래스를 상속받는 다른 하위 클래스가 있을 경우 그 "템플릿"을 동일하게 가져가는것을 말하는데, 이것이 전혀 지켜지지 않은 구조라는 것.

수정된 코드를 보자

KeywordSearchingBaseGateway

NaverSearchingGateway

상위 클래스인 KeywordSearchingBaseGateway에서 선언만 되어있는 fetch_keyword_results,_get_header, _get_params 메서드를, 하위 클래스인 NaverSearchingGateway에서 구현을 해주었다.

이를 통해 하위클래스는 상위 클래스가 정의한 메서드를 쓰도록 구현을 해주었으며, 이후 이 상위클래스를 상속받는 하위클래스에서도 반드시 이 메서드들을 구현하도록 강제할 수 있다.

개인적으로 하나를 가지고 이리저리 짜맞추는걸 좋아해서 객체지향을 좋아하는게 있었는데...아직 한참 멀었구나 하는 생각이 들었다. 이게 다 피가되고 살이 되는거지

1. 한 라인에 하나의 모듈 import

• import는 항상 상단에 위치
• import는 여러 모듈을 콤마로 연결하지 않고 한 라인에 하나의 모듈을 import 한다

2. 모듈 import 순서

모듈 import 순서는 다음과 같다

1. 표준 라이브러리 모듈
2. 서드파티 모듈
3. 직접 만든 모듈

직접 만든 모듈은 제외하고, 표준 라이브러리 모듈과 서드 파티 모듈은 어떻게 구분할까. 표준 라이브러리 모듈은 python이나 django에서 지원하는 모듈일테고 서드파티 모듈은 팀에서 공통적으로 사용하도록 개발된 모듈일것이다.

위의 코드상으로 보면 표준 라이브러리 모듈 다음에 서드파티 모듈이 들어가도록 구현되어있다. 다만, naver_gateway.py를 보면 직접 만든 모듈이 서드파티 모듈 상단에 위치해있다.

이 부분을 이렇게 바꾸자.

3. 절대 경로 사용

pep8에서는 모듈 import시 절대 경로 사용을 권장한다. 사용할 모듈이 현재 모듈과 같은 패키지 상에 있더라도 절대경로를 사용하도록 하고 있다. 유지보수시 이슈가 되는 모듈을 빠르게 찾을 수 있기 위함인듯하다.

위 코드를 보면 상대경로를 매우 많이 쓰고 있다...기존 레거시 코드가 상대경로를 주로 사용하고 있었고 이에 익숙해진 탓이다. 이를 고쳐보자

훨씬 가독성이 나아졌다.

Pycon2019의 셀러리 핵심과 커스터마이제이션을 정리한 글입니다.

1. Celery

celery?

메세지 전달을 기반으로 한 비동기 task 큐

• 작업 : message로 표현됨
• Client : 작업을 요청하는 주체
• Worker : 작업을 수행하는 주체
• Broker : 클라이언트와 워커 사이에서 메세지를 전달함.

이러한 구조에서는 클라이언트와 워커 모두 scaling이 가능하다. 따라서 클라이언트는 불필요하게 무거운 작업으로부터 자유롭고, 워커는 필요에 따라 확장이 가능하다.

celery는 AMQP라는 프로토콜을 기반으로 만들어졌다.

메세지를 보낼때 최소한 한번은 반드시 전달된다!!

Producer가 브로커로 메세지를 보내면, 이 메세지는 Consumer로 전달된다. Consumer가 이 메세지를 Consume한 뒤 브로커로 다 처리했다는 의미의 "Ack"를 보내면, 브로커는 최초에 메세지를 보낸 Producer에게 "Confirm"을 보낸다.

만약 ack에서 문제가 생기면, 브로커는 메세지가 처리됐는지 알 수가 없어서, Consumer에게 메세지를 다시 보내게 됨.

따라서 AMQP 상에서는 이 메세지에 대한 처리를 idempotent하게 되어야 한다.

f(f(x)) = f(x) = y

message는 여러번 전달 될 수도 있지만 이때 message 소비는 idempotent 해야한다. 즉, 여러번 전달 되어도 동일하게 하나의 작업만 수행된다는것.(엘레베이터 닫힘 버튼을 여러번 누른다고 여러번 닫히지 않는것과 같음)

2. 안정적으로 완료하기

Late Ack'

• Why Late ACK? : 중요한 태스크를 실행을 해야하는데 실행이 되지 않았을 때. Ack는 기본적으로 워커가 _태스크를 실행하기 직전에 실행_이 된다. 워커에서 Ack'를 브로커로 보내면 큐에서는 해당 작업이 삭제되고 워커에 의해서 작업이 실행된다. 이렇게 실행이 되다가 Worker Crash가 발생하면? 큐에서도 사라져 있으므로, 작업을 다시 실행할 방법이 없다.

• Late Ack 를 사용하면? 태스크의 실행이 완료됐을 때 ack가 브로커로 전달된다. 따라서 워커에 의해 태스크가 실행중일 때 worker crash가 발생해도 아직 큐에 남아있으므로 다시 실행할 수 있다.

• 중복 실행될 수 있다! Late Ack를 쓰면 태스크가 중복 실행될 수 있으므로, 반드시 태스크가 Idempotentic 하게 작성되어야 한다.

Retry

• 예상 가능하지만 통제될 수 없는 상황에서 문제가 생길 경우(외부 API 호출시의 순단 문제 등), 다시 수행되도록함.

• 예상 가능한 예외에서만 적용이 되도록 해야하며, atomic하게 작동되어야 한다.

3. 효율적으로 처리하기

• Worker는 자신이 처리할 수 잇는 만큼만 처리한다. 따라서 처리하는 속도보다 일이 쌓이는 속도가 커도 _*일시적으로는 *_큰 문제가 되지 않는다.

• but, 이게 지속이 되면?? 브로커에 불필요한 부하가 가게 되고, 실제 필요한 작업이 진행이 되지 않을수가 있다.

• time Limit을 설정해야 태스크가 일정시간 이상 실행되면 종료

> celery -A <PRJ> worker -P <WORKER_POOL_NAME> -c <concurrency>

prefetch limit

• prefetch limit : ack 되지 않은 태스크의 개수를 worker가 얼마나 갖고 있을 수 있는가?

  prefetch limit = worker_prefetch_multifiler * concurrency

  worker_prefetch_multifiler의 값에 설정한 concurrency 값을 곱하면 이 prefetch limit 값이다.

  [worker_prefetch_multifiler = 0]이라면? prefetch limit에 대해 제한이 없으므로 메모리나 효율성을 고려하지 않고 작업을 실행하게 된다.

위 그림처럼 concurrency가 각각 1,2인 워커가 있고, message가 쌓여있다고 하자.

worker_prefetch_multiflier = 1, acks_late = False로 하게 되면 concurrency가 1,2 이므로 워커는 각각 prefetch_limit 이 1,2가 된다.

따라서 concurrency가 2인 워커는 (현재 수행중인 작업을 제외하고) 2개의 메세지를, concurrency가 1인 워커는 1개의 메세지를 prefetch 하게된다.

• 그럼 이 prefetch limit을 어떻게 쓸 수 있을까?

  긴 태스크에 대해서는 worker_prefetch_multiplier = 1 로 설정하면 긴 태스크 뒤에 짧은 태스크들이 불필요하게 실행되는 것을 막을 수 있다.

• acks_late = True로 설정하면?

  

  worker_prefetch_multiflier = 1이라면 실행중인 task만 prefetch하게 된다.

  짧은 태스크들의 경우 worker_prefetch_multiflier에 따라 prefetch를 하는데도 네트워크를 타기 때문에 worker_prefetch_multiflier를 높여주면 task를 더 빠르게 실행시킬 수 있다(????뭔소리야???)

  길고 짧은 태스크를 구분하여 워커를 지정하면서 이 옵션을 쓸 수 있다.

Prefork

• multiprocessing으로 구현이 돼있음.
• "-c N" 옵션으로 실행을 하면, 1개의 master process와 N개의 child process로 실행이 된다.
• master에서 task를 분배하고, 실제 처리는 child에서 이루어진다.
• -O fair 옵션 : master에서 child 로 task가 전달될 때 기본적으로는 pipe buffer(한방에 쓸 수 있는 양??)가 허용되는 만큼의 메세지를 전달한다. 하지만 "-O fair" 옵션을 주게 되면 실행 가능한 경우에만 메세지를 전달하게 된다.

prefetch_limit은 브로커에서 워커로의 메세지 전달을 통제, -O fair는 마스터 프로세스에서 child process로의 메세지 전달을 통제하는 것이다.

긴 작업과 짧은 작업이 섞여 있는 경우에 -O fair 옵션을 주게 되면 성능 향상 및 예상 가능한 동작을 기대할 수 있다.(한줄서기를 할 때 줄이 더 빨리 줄어드는 것과 같은 원리)

작업의 성질에 따라 적절히 다르게 처리해야 한다.

• IO/CPU
• 중요도
• 수행시간
• 실행 빈도

4. customization

Global Rate Limit

kombu라는 셀러리가 이용하는 메세징 라이브러리가 있는데, kombu에서 TokenBucket을 이용해서 RateLimit을 구현하고 있다.

TokenBucket?

공식문서의 Routing tasks를 번역했습니다.

celery에서는 사용가능한 queue를 task_queues로 세팅한다.

default_exchange = Exchange('default', type='direct')
media_exchange = Exchange('media', type='direct')

app.conf.task_queues = (
    Queue('default', default_exchange, routing_key='default'),
    Queue('videos', media_exchange, routing_key='media.video'),
    Queue('images', media_exchange, routing_key='media.image')
)
app.conf.task_default_queue = 'default'
app.conf.task_default_exchange = 'default'
app.conf.task_default_routing_key = 'default'

task_default_queue는 명백한 루트가 지정되지 않은 태스크들을 라우팅할 때 쓰인다.

디폴트 exchange, exchange type, routing key는 태스크에 대한 디폴트 라우팅값과, task_queues에 들어가는 디폴트 값으로 쓰인다.

하나의 queue에 대한 여러개의 바인딩도 지원된다. 다음은 같은 queue에 대해 두개의 라우팅 키를 설정한 예시이다.

from kombu import Exchange, Queue, binding

media_exchange = Exchange('media', type='direct')

CELERY_QUEUES = (
    Queue('media', [
        binding(media_exchange, routing_key='media.video'),
        binding(media_exchange, routing_key='media.image'),
    ]),
)

celery의 AMQP Primer 를 번역했습니다.

Messages

message는 header와 body로 구성돼있다. Celery는 header를 message의 content type과 내용의 encoding을 저장하는데 쓴다. content type은 보통 message를 직렬화(serialize)하는데 쓰이는 직렬화 포맷(serialization format)이다. body는 실행될 task의 이름과 task의 id(uuid), task에 적용될 아규먼트들, 그리고 재시도 횟수나 ETA(??)같은 몇개의 추가적인 메타 데이터들을 갖고 있다.

아래는 Python dictionary 타입으로 된 task message 예제이다.

{'task': 'myapp.tasks.add',
 'id': '54086c5e-6193-4575-8308-dbab76798756',
 'args': [4, 4],
 'kwargs': {}}

Producers, consumers and brockers

message 발신자를 일반적으로 publisher나 producer라 부르며, message 수신자를 consumer라 한다.

message를 producer에서 consumer로 라우팅 해주는 message server를 broker라 한다.

exchanges, queues, rouing keys

1. message는 exchange들로 보내어진다.
2. 하나의 exchange가 message들을 하나 혹은 여러개의 queue로 보낸다. 몇개의 exchange 타입이 존재하며 각각 서로 다른 라우팅 방식을 제공하거나 혹은 서로 다른 라우팅 시나리오를 수행한다.
3. message들은 consume(메세지 수신 혹은 수행)되기 전 까지 queue 내부에서 대기한다.
4. message는 acknowledge되면 queue에서 삭제된다.

   consume은 실제로 message가 수행이 됐을때?아니면 consumer로 수신되었을때? acknowledge는??

message가 수신/발신되기 위해서는 아래와 같은 단계를 거친다.

1. exchange 생성
2. queue 생성
3. exchange와 queue 바인딩

Celery는 task_queues 속에 queue에 필요한 객체를 자동으로 생성해준다(queue의 auto_declare가 False인 경우에는 안됨)

3개의 queue에 대한 예제. 하나는 video를 위한 'video' queue, 하나는 image를 위한 'image' queue, 마지막으로 나머지 다른 모든것을 위한 'default' queue

from kombu import Exchange, Queue

app.conf.task_queues = (
    Queue('default', Exchange('default'), routing_key='default'),
    Queue('videos',  Exchange('media'),   routing_key='media.video'),
    Queue('images',  Exchange('media'),   routing_key='media.image'),
)
app.conf.task_default_queue = 'default'
app.conf.task_default_exchange_type = 'direct'
app.conf.task_default_routing_key = 'default'

exchange types

exchange type은 message가 exchange를 통해 queue로 라우팅 되는 방식을 정의한다. 표준적으로 정의된 exchange type은 direct, topic, fanout, headers가 있다. 또한 비표준 exchange타입도 플러그인으로써 Rabbit-MQ에 적용이 가능하다.

direct exchanges

Direct exchanges는 정확한 라우팅 키로 매칭되어, 만약 큐가 'video' 라는 라우팅 키에 바인딩 되어 있을 때, 그 queue는 오직 'video' 라우팅 키를 갖는 messge만 받는다.

topic exchanges

topic exchange는 라우팅 키들을 "."으로 분리된 단어들과 와일드 카드 문자들(* - 한개의 단어, # - 0 혹은 더 많은 단어들)로 매칭한다.

"usa.news, usa.weather, norway.news, norway.weather" 와 같은 단어들 뿐만 아니라, *.news (모든 뉴스), usa.# (USA에 관련된 모든 것들), usa.weather (모든 USA 날씨) 도 바인딩 될 수 있다.

exchange와 관련된 api 명령들

exchange.declare(exchange_name, type, passive,
durable, auto_delete, internal)

exchange 선언 - passive :exchange가 생성되지 않음. 또 이를 통해 exchange가 이미 존재하는지의 여부를 확인할 수 있다. - durable : durable(지속적인)한 exchange는 영속적이다. 예를들어, 브로커가 재시작될 때에도 지속된다.

• auto_delete : 이 exchange를 사용하는 queue가 더 없을 경우, 이 exchange는 자동으로 삭제된다.

queue.declare(queue_name, passive, durable, exclusive, auto_delete)

queue 선언 - exclusive : exclusive한 queue는 오직 현재의 커넥션에서만 사용될 수 있다. exclusive 옵션이 설정되면 auto_delete 가 된다.

queue.bind(queue_name, exchange_name, routing_key)

라우팅 키를 통해 queue와 exchange를 바인딩함. 바인딩되지 않은 queue는 message를 받을 수 없으므로, bind가 반드시 필요하다.

queue.delete(name, if_unused=False, if_empty=False)

queue와 binding 설정을 삭제함.

exchange.delete(name, if_unused=False)

exchange 삭제

선언하는것이 반드시 "생성"을 의미하는것은 아니다. 선언을 할때 이는 객체가 존재하고 그것이 사용 가능하다고 주장하는것이다. 꼭 consumer나 producer가 제일 먼저 exchange/queue/binding을 생성해야 한다는 룰은 없다. 보통 그것들을 필요로 하는 쪽에서 먼저 그것들을 생성한다.

API 실습(hands-on with the api)

Celery에는 AMQP API에 대한 CLI 엑세스에 사용되는 celery amqp라는 도구가 함께 제공되어 queue 및 exchange의 생성/삭제, queue제거 또는 message 전송과 같은 관리작업에 엑세스 할 수 있다. AMQP가 아닌 브로커에도 사용 할 수 있지만, 이 경우에는 모든 커맨드의 사용이 제한 될 수 있다. (redis는 amqp가 아니므로 이건 다음에 기회가 될때 계속 번역함)

celery 공식문서의 Routing Tasks를 번역하였습니다

1. Basic

자동 라우팅

task를 라우팅하는 가장 쉬운 방법은 task_create_missing_queues 세팅을 사용하는 것이다.

이 세팅을 키면, 아직 task queues에 이름없는 queue가 생성된다. 이를 통해 task라우팅을 손쉽게 수행할 수 있다.

일반적인 task들을 다루는 두개의 서버 x, y와 feed 관련 작업만 다루는 z 서버가 있다 가정하자. 이러한 구성을 사용 할 수 있다.

task_routes = {'feed.tasks.import_feed': {'queue': 'feeds'}}

이를 활성화 하면, feed 가져오기 작업(import_feed)은 "feeds" 큐로 라우팅 되고, 다른 모든 작업들은 기본적인 queue로 라우팅된다.

혹은 feed.taks 네임스페이스에 있는 모든 태스크들을 매칭하기 위해, 전역 패턴 매칭 심지어 정규 표현식까지 사용할 수 있다.

app.conf.task_routes = {'feed.tasks.*': {'queue': 'feeds'}}

만약 매칭 패턴들의 순서가 중요하다면 items 포맷으로 라우터를 지정해줘야 한다.

task_routes = ([
    ('feed.tasks.*', {'queue': 'feeds'}),
    ('web.tasks.*', {'queue': 'web'}),
    (re.compile(r'(video|image)\.tasks\..*'), {'queue': 'media'}),
],)

task_routes 세팅은 딕셔너리 타입일수도 있고 리스트 타입일수도 있다. 이련 경우에는, 리스트나 딕셔너리를 튜플로 감싸줘야 한다.

라우터 설정이 끝난 뒤에, z 서버만을 feeds 큐를 처리하기 위해 구동시킬 수 있다.

user@z:/$ celery -A proj worker -Q feeds

원하는 만큼 큐를 지정할 수 있어서, 이 서버 프로세스를 아래와 같이 디폴트 큐로 지정할 수 있다.

user@z:/$ celery -A proj worker -Q feeds,celery

• 디폴트 큐의 이름 변경

  app.conf.task_default_queue = 'default'

• 큐 정의 방법 큐의 이름은 아래와 같이 지정된다.

  {'exchange': 'video',
  'exchange_type': 'direct',
  'routing_key': 'video'}

  AMQP가 아닌 REDIS나 SQS는 변경을 지원하지 않으므로, "exchange"와 큐의 이름을 동일하게 해주어야 한다.

수동 라우팅

다시 일반적인 task들을 다루는 두개의 서버 x, y와 feed 관련 작업만 다루는 z 서버가 있다 가정하자. 아래와 같은 구성을 사용할 수 있다.

from kombu import Queue

app.conf.task_default_queue = 'default'
app.conf.task_queues = (
    Queue('default',    routing_key='task.#'),
    Queue('feed_tasks', routing_key='feed.#'),
)
app.conf.task_default_exchange = 'tasks'
app.conf.task_default_exchange_type = 'topic'
app.conf.task_default_routing_key = 'task.default'

task_queues는 큐 객체들의 리스트이다. 만약 exchange나 echange_type값을 바꾸지 않았다면, 이 값들은 task_default_exchange와 task_default_exchange_type 세팅에서 가져오게 된다.

태스크를 feed_tasks 큐로 라우팅하기 위해, task_routes 세팅을 추가할 수있다.

task_routes = {
        'feeds.tasks.import_feed': {
            'queue': 'feed_tasks',
            'routing_key': 'feed.import',
        },
}

또한 routing_key아큐먼트를 사용해서 Task.apply_async() 나 send_task()로 오버라이딩 할 수도 있다.

>>> from feeds.tasks import import_feed
>>> import_feed.apply_async(args=['http://cnn.com/rss'],
                        queue='feed_tasks',
                        routing_key='feed.import')

z서버가 feed queue에서만 처리되도록 하기 위해 셀러리 구동시 celery worker -Q 옵션을 사용한다.

celery -A proj worker -Q feed_tasks --hostname=z@%h

서버 x와 y는 디폴트 큐를 사용하도록 반드시 표기해주어야 한다.

user@x:/$ celery -A proj worker -Q default --hostname=x@%h
user@y:/$ celery -A proj worker -Q default --hostname=y@%h

만약 다른 exchange에 있지만 추가하고 싶은 다른 queue가 있다면 exchange와 exchange type을 지정해주기만 하면 된다.

from kombu import Exchange, Queue

app.conf.task_queues = (
    Queue('feed_tasks',    routing_key='feed.#'),
    Queue('regular_tasks', routing_key='task.#'),
    Queue('image_tasks',   exchange=Exchange('mediatasks', type='direct'),
                           routing_key='image.compress'),
)

2. Special Routing Options

Redis 메세지 우선순위

Celerㅛ Redis 전송은 우선순위 필드를 존중(???) 하긴 하지만, 실제로 Redis에는 우선순위 개념이 없다. 따라서 Redis로 우선순위를 구현하기 위해서는 아래의 사항을 참고할 필요가 있다.

우선순위에 따라 작업을 예약하려먼 queue_order_strategy 전송 옵션을 구성해야한다.

app.conf.broker_transport_options = {
    'queue_order_strategy': 'priority',
}

우선순위는 각각의 queue에 대해 n개의 리스트를 생성함으로써 지원된다.비록 10개(0~9)의 우선순위 레벨이 있지만, 이 우선순위 레벨들은 자원을 세이브하기 위해 기본적으로 4개 수준으로 통합되었다는 뜻이다(???). 즉, celery라는 큐는 실제로는 4개의 큐로 분할됨을 뜻한다.

celery라 이름붙여진 큐가 가장 우선순위가 높으며, 다른 큐들은 분할자(기본적으로 x06x16)를 갖게 되고, 이 뒤에 우선순위 숫자가 붙게 된다.

['celery', 'celery\x06\x163', 'celery\x06\x166', 'celery\x06\x169']

-> celery + \06\16(분할자) + 3 (우선순위) 

만약 더 많은 우선순위 레벨이나 다른 분할자를 지정하고자 할 때에는 broker_transport_options에서 'prioty_steps'와 'sep'옵션을 지정할 수 있다.

app.conf.broker_transport_options = {
    'priority_steps': list(range(10)),
    'sep': ':',
    'queue_order_strategy': 'priority',
}

이렇게 세팅을 하면 아래와 같은 queue가 만들어진다.

['celery', 'celery:1', 'celery:2', 'celery:3', 'celery:4', 'celery:5', 'celery:6', 'celery:7', 'celery:8', 'celery:9']

이렇게 구현된 우선순위들은 절대 서버 수준에서 구현된 우선순위들만큼 좋을리는 없고, 기껏해야 근사치 정도일 수 있다. 다만, 어플리케이션 수준에서는 충분할 수 있다.

하나의 시스템이 제 시간에 처리할 수 있는 데이터의 양은 제한되어 있다. 만약 셀러리에서 하나의 태스크가 완료되는데 10분이 걸리고, 매 분마다 새로운 태스크들이 계속해서 유입된다면, 큐는 절대로 비는 일이 없을 것이다. 이것이 큐의 길이를 모니터링 하는 것이 중요한 그 이유이다.

큐를 모니터링 하는 방법으로는 Munin(현재 큐에 있는 태스크의 수를 그래프로 보여줌)을 사용하는 것이 있다. 만약 큐가 더이상 수용할수 없을 만큼 가득 차게 된다면 즉각적으로 확인할 수 있다. 이를 통해 새로운 worker node를 추가하거나, 불필요한 태스크를 제거하는 등의 조치를 취할 수 있다.

2. 일반적인 세팅

Broker Connection pool

Broker Connection pool은 2.5 버전부터 기본사양으로써 사용이 가능하다. Broker Connection Pool을 미세 조정함으로써 경쟁 상황을 최소화 할 수 있는데, 이 값들은 broker connection을 사용하는 활성화된 thread의 수에 기반해야 한다.

Using Transient Queues

기본적으로 Celery에 의해 만들어진 queue는 영구적이다. 이 말은, 브로커가 재시작한다 할지라도, 브로커는 태스크가 반드시 수행되도록 message를 디스크에 기록한다는 것이다.

하지만 어떤 경우에는, 이러한 메시지가 손실되어도 괜찮기 때문에 모든 task가 내구성을 가질 필요는 없다. 이러한 태스크에는 "임시 큐"를 생성하여 사용함으로써 퍼포먼스를 증대 시킬 수 있다.

from kombu import Exchange, Queue

task_queues = (
    Queue('celery', routing_key='celery'),
    Queue('transient', Exchange('transient', delivery_mode=1),
          routing_key='transient', durable=False),
)

혹은 task_routes를 사용한다

task_routes = {
    'proj.tasks.add': {'queue': 'celery', 'delivery_mode': 'transient'}
}

delery_mode는 queue로의 메세지 전달방식을 바꾼다. 1은 message가 반드시 디스크에 기록 되지 않아도 된다는걸 뜻하며, 2는 반드시 디스크에 기록되어야 함을 의미한다.

queue의 인자를 지정해줌으로써, 태스크를 새로 생성한 큐로 가도록 할 수 있다.

task.apply_async(args, queue='transient')

3.Worker Settings

Prefetch Limits

Prefetch limits는 하나의 worker가 처리할 수 있는 task(message)의 수이다. 만약 prefetch limit이 0일 경우, 작업자는 message를 더 빨리 처리할 수 있는 노드가 있거나 혹은 이 message가 메모리 크기에 맞지 않는 등의 문제를 고려하지 않고 계속해서 message(task)를 처리할 것이다.

worker의 기본 prefetch 카운트 수는, worker_prefetch_multiplier옵션을 통해 지정할 수 있다(기본값 4)

개별 작업의 처리 시간이 긴 경우 prefetch 수는 1로 해야 한다. 즉, 한번에 worker process 하나당 하나의 작업만 예약하도록 해야한다.(복잡한 수학 연산등이 이에 해당)

반면, 개별 작업의 처리시간은 짧지만 처리량이 많거나 왕복 대기시간이 긴 경우 prefetch count가 커야 한다. message들이 미리 가져와져서(prefetced) 메모리에 올라가면 worker는 초당 더 많은 작업을 수행할 수 있다. 이 적절한 값을 찾기 위해서는 여러 실험을 해봐야한다...이러한 상황에는 50 이나 150 등이 좋을수도?!

개별 작업의 처리시간이 긴 작업들과 짧은 작업들이 섞여 있을 경우, 가장 좋은 방법은 2개의 worker node를 사용해서 긴 작업과 짧은작업으로 나눠서 처리해주는 것이다.Routing Tasks

Memory Usage

만약 하나의 worker에서 높은 메모리 점유를 경험한다면, 가장 먼저 이 이슈가 Celery master process에서도 일어나는지 확인해야 한다. Celery master process의 메모리 점유는 구동 된 이후에 계속해서 극적으로 증가하면 안된다. 이러한 일이 발생한다면 이는 메모리 누수일 수 있다.

만약 자식 프로세스에만 메모리 사용량이 높을 경우 해당 task에 문제가 있다는 것이다.

파이썬 프로세스의 메모리 사용에는 "high watermark" 라는 것이 있어 자식 프로세스가 중단되기 전까지는 운영체제에게 메모리를 반환하지 않음을 명심하라. 이말은 즉, 하나의 메모리 사용량이 높은 작업이 있다면 자식프로세스를 재시작 하기 전까지는 영구적으로 자식 프로세스의 메모리 사용을 증가시킬 수 있다는 뜻이다. chunking 로직을 이러한 task에 추가하여 메모리 사용을 줄일 필요가 있다.

Celery worker는 "worker_max_tasks_per_child"와 "worker_max_memory_per_child"을 사용하여 자식 프로세스로부터의 메모리 누수를 줄일 수 있다.

이 세팅들을 너무 낮지 않도록 해야하며, 또한 worker가 자식 프로세스를 재시작 하는데 드는 시간이 task를 처리하는데 드는 시간보다 더 걸리지 않도록 주의해야 한다. 예를 들어 만약 " worker_max_tasks_per_child"의 값을 1로 사용하고, 자식 프로세스가 시작되는데 걸리는 시간이 1초라면, 자식프로세스는 분당 최대 60개의 task만을 처리할 수 있을 것이다. 비슷한 상황은 태스크의 수가 언제나 "worker_max_memory_per_child"보다 클 경우 발생할 수 있다.

Jenkins Pipeline (or simply "Pipeline" with a capital "P") is a suite of plugins which supports implementing and integrating continuous delivery pipelines into Jenkins. 젠킨스 파이프라인은 지속적인 배포 파이프라인을 구현하고, 젠킨스에 통합하는것을 지원하는 플러그인이다.

2. Declarative vs Scripted 파이프라인 문법

3. 왜 pipeline을 사용하는가?

4. pipeline의 컨셉들

2. 관리자 권한으로 젠킨스에 들어가준다

3.openjdk 설치

4. jdk가 설치된 경로를 찾는다(env)

env를 입력해주면 jenkins 컨테이너의 환경변수가 주르륵 나온다. 여기에서 JAVA_HOME이 해당 경로이다.

JAVA_HOME=/opt/java/openjdk 확인

5. 위 경로를 jenkins-web의 jdk 경로로 지정해준다.

6. 빌드여부 확인

너무 잘된다 ㅎㅋㅎ

2. Docker의 jenkins 컨테이너에 터널링으로 접속

- docker exec -it jenkins-server bash

3. jenkins 컨테이너 내부에서 내 프로젝트 찾기

- continuous integration : 여러 개발자들의 코드베이스를 계속해서 통합함.
- continuous delivery : 서비스를 배달함. 코드 베이스가 항상 배포 가능한 상태로 유지하는 것.
- continuous deployment : 코드베이스를 사용자가 사용가능한 환경으로 배포하는것을 자동화함.

- 코드가 지속적으로 합쳐짐 : 개발자와 사용자 사이의 격차를 없임
- 이러한 과정에는 코드 빌드+테스트+배포 가 포함됨.

- ci/cd가 없으면 코드가 통합이 안돼어 있어서 여러 개발자가 개발한것을 합칠때 문제가 됨.
- 가능한 코드를 만들자마자 바로바로 합치자는 것.

- ci : 가능한 최대한 많이 빨리 내 코드를 코드 베이스에 안착시킴
- cd : 

2. Jenkins의 기본 개념과 동작 방식

- 이러한 귀찮은것들을 대신해주는것이 젠킨스
- java runtime 위에서 동작한다. 아니면 도커에서 하든가
- 다양한 플러그인을 활용해서 각종 자동화 작업을 처리함. 하나의 플러그인으로 모듈화함.
- 일련의 “자동화 작업의 순서들의 집합”인 파이프라인을 통해 ci/cd 파이프라인을 구축함.
- 두가지의 pipeline syntax가 존재 : declarative, scripted —> 얼리페이에서는 declarative pipeline syntax를 사용함.

- 대표적인 플러그인
    a. Credentials plugin : 젠킨스는 단지 서버임. 배포에 필요한 각종 리소스에 접근하기 위해서는 중요 정보들을 저장해야 한다. 
                    이러한 중요 정보(was token, git access token 등등)을 저장해주는 플러그인.

    b. Pipeline plugin : 파이프라인이란 ci/cd 파이프라인을 젠킨스에 구현하기 위한 “일련의 플러그인의 집합이자 구성”,
                  여러 플러그인들을 이 파이프라인에서 용도에 맞게 사용하고 정의함으로써 파이프라인을 통해 서비스가 배포됨.
                일종의 작업 명세서/레시피.
            1) section : 대 카테고리, 누가 어떤 일을 할 것인가.
                - agent section : 여러 slave node를 두고 일을 시킬 수 있는데, 이처럼 어떤 젠킨스가 일을 하게 할것인지를 지정한다.
                - post section : 각 순서(stage)마다 이후의 결과에 따라 후속 조치를 할 수 있다.(성공시에 성공 이메일 등등)
                - stage section : 어떤 일들을 처리할 것인지 일련의 stage(순서)를 정의함.
                - step section : 한 스테이지 안에서의 단계로, 일련의 스텝을 보여준다.

            2) declarative : 각 스테이지 안에서 어떤 일을 처리할 것인지를 정의함.
                - when : 언제 실행되는가
                - trigger : 어떤 형태로 트리거가 되는가, 이 파이프라인이 어떤 주기로 실행이 되는가
                - environment : 환경 변수

            3) steps : 실행 가능한 여러 작업들, 플러그인을 설치하면 그 플러그인에 맞는 스텝들을 쓸 수 있다.

    c. Docker plugin