Python 환경에서 비동기 작업 대기열 패키지 Celery 설치 및 사용

1. 소개

셀러리(celery)는 분산 메시징 기반의 비동기 작업 대기열/작업 대기열입니다. 실시간 운영에 중점을 두고 있지만 일정 지원도 잘 되어 있습니다.
셀러리는 생산 시스템에서 매일 수백만 개의 작업을 처리하는 데 사용됩니다.
셀러리는 Python으로 작성되었지만 프로토콜은 모든 언어로 구현될 수 있습니다. 웹훅을 통해 다른 언어로도 구현 가능합니다.
권장되는 메시지 브로커는 RabbitMQ이지만 Redis, Beanstalk, MongoDB, CouchDB 및 데이터베이스(SQLAlchemy 또는 Django ORM 사용)에 대한 지원은 제한적으로 제공됩니다.
Celery는 Django, Pylons 및 Flask와 쉽게 통합할 수 있습니다. django-celery, celery-pylons 및 Flask-Celery 추가 기능 패키지를 사용하기만 하면 됩니다.

2. 설치
위 개념에 따라 여러 가지를 설치해야 합니다: RabbitMQ, SQLAlchemy, Celery
설치 방법도 매우 간단합니다: RabbitMQ:
Mac:

brew install rabbitmq

linux:

sudo apt-get install rabbitmq-server

나머지 두 개는 모두 Python입니다. MySQL 드라이버를 설치한 적이 없는 학생은 MySQL-python을 설치해야 할 수도 있습니다.
설치가 완료된 후 서비스를 시작하세요:

$ rabbitmq-server[回车]

시작한 후 창을 닫지 말고 다음 작업을 수행하여 새로 만듭니다. 창(탭)

3. 간단한 경우
이전 RabbitMQ가 시작되었는지 확인하세요.
공식 웹사이트의 예를 그대로 사용하여 다음 내용이 포함된 task.py 파일을 임의의 디렉터리에 생성합니다:

from celery import Celery

app = Celery('tasks', broker='amqp://guest@localhost//')

@app.task
def add(x, y):
  return x + y

동일한 디렉터리 :

$ celery -A tasks worker --loglevel=info

이 명령은 작업자를 시작하고 작업을 작업에 추가하고(add(x,y)) 작업을 넣는 것을 의미합니다. 대기열 중간에 들어갑니다.
창을 열어두고 새 창을 열어 대화형 모드, Python 또는 ipython으로 들어갑니다.

>>> from tasks import add
>>> add.delay(4, 4)

지금까지는 다음을 수행할 수 있습니다. 이미 셀러리를 사용하고 있습니다. 추가 작업은 위의 Python 대화형 모드에서 간단히 호출되고 4, 4 매개변수가 전달됩니다.
그런데 갑자기 이 작업의 실행 결과와 상태, 완료 여부를 알고 싶은 경우가 있습니다. 따라서 백엔드를 설정해야 합니다.
이전 task.py의 코드를 다음과 같이 수정합니다.

# coding:utf-8
import subprocess
from time import sleep

from celery import Celery

backend = 'db+mysql://root:@192.168.0.102/celery'
broker = 'amqp://guest@192.168.0.102:5672'

app = Celery('tasks', backend=backend, broker=broker)


@app.task
def add(x, y):
  sleep(10)
  return x + y


@app.task
def hostname():
  return subprocess.check_output(['hostname'])

백엔드 추가 외에도 who 메소드도 위에 추가됩니다. 다중 서버 작동을 테스트합니다. 수정이 완료된 후에도 이전과 같이 계속 시작됩니다.
비슷하게 Python 상호 작용 모델을 입력합니다.

>>> from tasks import add, hostname
>>> r = add.delay(4, 4)
>>> r.ready() # 10s内执行，会输出False，因为add中sleep了10s
>>>
>>> r = hostname.delay()
>>> r.result # 输出你的hostname

4. 여러 서버를 테스트합니다
위의 작업을 완료합니다. 테스트에서 Celery는 분산 작업 관리라고 하는데 분산 표현은 어디에 있습니까? 그 작업은 어떻게 수행됩니까? 어느 머신에서 실행되었나요?
현재 서버에서 셀러리 서비스를 종료하지 않고 같은 방법으로 다른 서버에 셀러리를 설치하고 시작합니다:

$ celery -A tasks worker --loglevel=info

서버가 RabbitMQ에 연결되어 있는 경우 이전 서버의 Celery 서비스가 방금 시작한 서버의 호스트 이름을 출력하는 것으로 나타났습니다.
그런 다음 Python 대화형 모드로 들어갑니다.

>>> from tasks import hostname
>>>
>>> for i in range(10):
...   r = hostname.delay()
...   print r.result # 输出你的hostname
>>>

입력한 내용을 살펴보고 두 서버 모두에서 셀러리 서비스를 시작할 때의 출력을 관찰합니다.

5. RabbitMQ 원격 연결 문제
초기 테스트 중 원격 서버가 로컬 RabbitMQ 서비스에 연결할 수 없는 것으로 나타났습니다. /usr/local/etc/rabbitmq에서 /rabbitmq-env.conf 파일에서 NODE_IP_ADDRESS=127.0.0.1의 IP를 0.0.0.0으로 수정합니다.

6. 요약
이 글에서는 여전히 분산 사용에 중점을 두고 Celery의 사용을 간략하게 소개합니다. 불편한 부분은 확장할 때 작업을 직접 공유하는 대신 코드(tasks.py)를 다시 배포해야 한다는 것입니다. 아마도 Celery는 작업을 사용하여 다른 작업자를 일치시킬 수 있을까요? 아직은 잘 모르는데, 좀 더 써본 후 얘기해보겠습니다.

Python 환경에서 비동기 작업 대기열 패키지 Celery를 설치하고 사용하는 방법에 대한 자세한 내용은 PHP 중국어 웹사이트를 참고하세요!