Celery

瀹樻柟

Celery 瀹樼綉锛歨ttp://HdhCmsTestceleryproject.org/

Celery 瀹樻柟鏂囨。鑻辨枃鐗堬細http://docs.celeryproject.org/en/latest/index.html

Celery 瀹樻柟鏂囨。涓枃鐗堬細http://docs.jinkan.org/docs/celery/

Celery鏋舵瀯

Celery鐨勬灦鏋勭敱涓夐儴鍒嗙粍鎴愶紝娑堟伅涓棿浠讹紙message broker锛夈€佷换鍔℃墽琛屽崟鍏冿紙worker锛夊拰 浠诲姟鎵ц缁撴灉瀛樺偍锛坱ask result store锛夌粍鎴愩€?/p>

浠诲姟鎵ц鍗曞厓

Worker鏄疌elery鎻愪緵鐨勪换鍔℃墽琛岀殑鍗曞厓锛寃orker骞跺彂鐨勮繍琛屽湪鍒嗗竷寮忕殑绯荤粺鑺傜偣涓€?/p>

浠诲姟缁撴灉瀛樺偍

Task result store鐢ㄦ潵瀛樺偍Worker鎵ц鐨勪换鍔＄殑缁撴灉锛孋elery鏀寔浠ヤ笉鍚屾柟寮忓瓨鍌ㄤ换鍔＄殑缁撴灉锛屽寘鎷珹MQP, redis绛?/p>

浣跨敤鍦烘櫙

寮傛浠诲姟锛氬皢鑰楁椂鎿嶄綔浠诲姟鎻愪氦缁機elery鍘诲紓姝ユ墽琛岋紝姣斿鍙戦€佺煭淇?閭欢銆佹秷鎭帹閫併€侀煶瑙嗛澶勭悊绛夌瓑

瀹氭椂浠诲姟锛氬畾鏃舵墽琛屾煇浠朵簨鎯咃紝姣斿姣忓ぉ鏁版嵁缁熻

Celery鎵ц寮傛浠诲姟

project 鈹溾攢鈹€ celery_task # celery鍖? 鈹? 鈹溾攢鈹€ __init__.py # 鍖呮枃浠? 鈹? 鈹溾攢鈹€ celery.py # celery杩炴帴鍜岄厤缃浉鍏虫枃浠讹紝涓斿悕瀛楀繀椤讳氦celery.py 鈹? 鈹斺攢鈹€ tasks.py # 鎵€鏈変换鍔″嚱鏁? 鈹溾攢鈹€ add_task.py # 娣诲姞浠诲姟 鈹斺攢鈹€ get_result.py # 鑾峰彇缁撴灉

鍩烘湰浣跨敤

celery.py

# 1锛夊垱寤篴pp + 浠诲姟

# 2锛夊惎鍔╟elery(app)鏈嶅姟锛?# 闈瀢indows
# 鍛戒护锛歝elery worker -A celery_task -l info
# windows锛?# pip3 install eventlet
# celery worker -A celery_task -l info -P eventlet

# 3锛夋坊鍔犱换鍔★細鎵嬪姩娣诲姞锛岃鑷畾涔夋坊鍔犱换鍔＄殑鑴氭湰锛屽彸閿墽琛岃剼鏈?
# 4锛夎幏鍙栫粨鏋滐細鎵嬪姩鑾峰彇锛岃鑷畾涔夎幏鍙栦换鍔＄殑鑴氭湰锛屽彸閿墽琛岃剼鏈?

from celery import Celery

broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
app = Celery(broker=broker, backend=backend, include=['celery_task.tasks'])

tasks.py

from .celery import app
import time
@app.task
def add(n, m):
    print(n)
    print(m)
    time.sleep(10)
    print('n+m鐨勭粨鏋滐細%s' % (n + m))
    return n + m

@app.task
def low(n, m):
    print(n)
    print(m)
    print('n-m鐨勭粨鏋滐細%s' % (n - m))
    return n - m

add_task.py

from celery_task import tasks

# 娣诲姞绔嬪嵆鎵ц浠诲姟
t1 = tasks.add.delay(10, 20)
t2 = tasks.low.delay(100, 50)
print(t1.id)


# 娣诲姞寤惰繜浠诲姟
from datetime import datetime, timedelta
def eta_second(second):
    ctime = datetime.now()
    utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())
    time_delay = timedelta(seconds=second)
    return utc_ctime + time_delay

tasks.low.apply_async(args=(200, 50), eta=eta_second(10))

get_result.py

from celery_task.celery import app

from celery.result import AsyncResult

id = '21325a40-9d32-44b5-a701-9a31cc3c74b5'
if __name__ == '__main__':
    async = AsyncResult(id=id, app=app)
    if async.successful():
        result = async.get()
        print(result)
    elif async.failed():
        print('浠诲姟澶辫触')
    elif async.status == 'PENDING':
        print('浠诲姟绛夊緟涓鎵ц')
    elif async.status == 'RETRY':
        print('浠诲姟寮傚父鍚庢鍦ㄩ噸璇?)
    elif async.status == 'STARTED':
        print('浠诲姟宸茬粡寮€濮嬭鎵ц')

楂樼骇浣跨敤

celery.py

# 1锛夊垱寤篴pp + 浠诲姟

# 2锛夊惎鍔╟elery(app)鏈嶅姟锛?# 闈瀢indows
# 鍛戒护锛歝elery worker -A celery_task -l info
# windows锛?# pip3 install eventlet
# celery worker -A celery_task -l info -P eventlet

# 3锛夋坊鍔犱换鍔★細鑷姩娣诲姞浠诲姟锛屾墍浠ヨ鍚姩涓€涓坊鍔犱换鍔＄殑鏈嶅姟
# 鍛戒护锛歝elery beat -A celery_task -l info

# 4锛夎幏鍙栫粨鏋?

from celery import Celery

broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
app = Celery(broker=broker, backend=backend, include=['celery_task.tasks'])


# 鏃跺尯
app.conf.timezone = 'Asia/Shanghai'
# 鏄惁浣跨敤UTC
app.conf.enable_utc = False

# 浠诲姟鐨勫畾鏃堕厤缃?from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
    'low-task': {
        'task': 'celery_task.tasks.low',
        'schedule': timedelta(seconds=3),
        # 'schedule': crontab(hour=8, day_of_week=1),  # 姣忓懆涓€鏃╁叓鐐?        'args': (300, 150),
    }
}

tasks.py

from .celery import app

import time
@app.task
def add(n, m):
    print(n)
    print(m)
    time.sleep(10)
    print('n+m鐨勭粨鏋滐細%s' % (n + m))
    return n + m


@app.task
def low(n, m):
    print(n)
    print(m)
    print('n-m鐨勭粨鏋滐細%s' % (n - m))
    return n - m

get_result.py

from celery_task.celery import app

from celery.result import AsyncResult

id = '21325a40-9d32-44b5-a701-9a31cc3c74b5'
if __name__ == '__main__':
    async = AsyncResult(id=id, app=app)
    if async.successful():
        result = async.get()
        print(result)
    elif async.failed():
        print('浠诲姟澶辫触')
    elif async.status == 'PENDING':
        print('浠诲姟绛夊緟涓鎵ц')
    elif async.status == 'RETRY':
        print('浠诲姟寮傚父鍚庢鍦ㄩ噸璇?)
    elif async.status == 'STARTED':
        print('浠诲姟宸茬粡寮€濮嬭鎵ц')

celery.py

# 閲嶇偣锛氳灏?椤圭洰鍚?settings 鎵€鍗犵殑鏂囦欢澶规坊鍔犲埌鐜鍙橀噺
# import sys
# sys.path.append(r'椤圭洰缁濆璺緞')

# 寮€鍚痙jango鏀寔
import os
os.environ.setdefault('DJANGO_SETTINGS_MODULE', '椤圭洰鍚?settings')
import django
django.setup()



# 1锛夊垱寤篴pp + 浠诲姟

# 2锛夊惎鍔╟elery(app)鏈嶅姟锛?# 闈瀢indows
# 鍛戒护锛歝elery worker -A celery_task -l info
# windows锛?# pip3 install eventlet
# celery worker -A celery_task -l info -P eventlet

# 3锛夋坊鍔犱换鍔★細鑷姩娣诲姞浠诲姟锛屾墍浠ヨ鍚姩涓€涓坊鍔犱换鍔＄殑鏈嶅姟
# 鍛戒护锛歝elery beat -A celery_task -l info

# 4锛夎幏鍙栫粨鏋?

from celery import Celery

broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
app = Celery(broker=broker, backend=backend, include=['celery_task.tasks'])


# 鏃跺尯
app.conf.timezone = 'Asia/Shanghai'
# 鏄惁浣跨敤UTC
app.conf.enable_utc = False

# 浠诲姟鐨勫畾鏃堕厤缃?from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
    'django-task': {
        'task': 'celery_task.tasks.test_django_celery',
        'schedule': timedelta(seconds=3),
        'args': (),
    }
}

tasks.py

from .celery import app
# 鑾峰彇椤圭洰涓殑妯″瀷绫?from api.models import Banner
@app.task
def test_django_celery():
    banner_query = Banner.objects.filter(is_delete=False).all()
    print(banner_query)

查看更多关于Django 澶х鎵嬫妸鎵嬪甫浣犱笂璺?~ celery绯诲垪的详细内容...