blackmatrix7/celery-demo

# 等价于直接定义一个broker，如
# broker = 'amqp://user:[email protected]:5672//'
broker = current_config.CELERY_BROKER_URL
# 将config模块的配置转换成dict，等价于直接通过dict定义一组celery配置
celery_conf = {k: v for k, v in current_config.items()}

# 创建celery实例
celery = Celery('demo',  broker=broker)
# 载入celery配置
celery.config_from_object(celery_conf)

celery = Celery('demo',  broker=broker)
celery.config_from_object(celery_conf)
# argv中传入启动celery所需的参数
celery.start(argv=['celery', 'worker', '-l', 'info', '-f', 'logs/celery.log'])

celery.start(argv=['celery', 'worker', '-Q', 'message_queue', '-l', 'info', '-f', 'logs/message.log'])

# 导入manage.py中创建的celery实例
from manage import celery

# 需要在celery中异步执行的函数，增加celery.task装饰器
@celery.task
def async_send_email(send_from, send_to, subject, content):
    """
    忽略方法实现
    """
    pass

async_push_message(send_to=['张三', '李四'], content='老板喊你来背锅了')
async_push_message(send_to=['刘五', '赵六'], content='老板喊你来领奖金了')

async_push_message.delay(send_to=['张三', '李四'], content='老板喊你来背锅了')
async_push_message.delay(send_to=['刘五', '赵六'], content='老板喊你来领奖金了')

# 导入Queue
from kombu import Queue
# 导入Task所在的模块，所有使用celery.task装饰器装饰过的函数，所需要把所在的模块导入
# 我们之前创建的几个测试用函数，都在handlers.async_tasks和handlers.schedules中
# 所以在这里需要导入这两个模块，以str表示模块的位置，模块组成tuple后赋值给CELERY_IMPORTS
# 这样Celery在启动时，会自动找到这些模块，并导入模块内的task
CELERY_IMPORTS = ('handlers.async_tasks', 'handlers.schedules')
# 为Celery设定多个队列，CELERY_QUEUES是个tuple，每个tuple的元素都是由一个Queue的实例组成
# 创建Queue的实例时，传入name和routing_key，name即队列名称
CELERY_QUEUES = (
    Queue(name='email_queue', routing_key='email_router'),
    Queue(name='message_queue', routing_key='message_router'),
    Queue(name='schedules_queue', routing_key='schedules_router'),
)
# 最后，为不同的task指派不同的队列
# 将所有的task组成dict，key为task的名称，即task所在的模块，及函数名
# 如async_send_email所在的模块为handlers.async_tasks
# 那么task名称就是handlers.async_tasks.async_send_email
# 每个task的value值也是为dict，设定需要指派的队列name，及对应的routing_key
# 这里的name和routing_key需要和CELERY_QUEUES设定的完全一致
CELERY_ROUTES = {
    'handlers.async_tasks.async_send_email': {
        'queue': 'email_queue',
        'routing_key': 'vcan.email_router',
    },
    'handlers.async_tasks.async_push_message': {
        'queue': 'message_queue',
        'routing_key': 'message_router',
    },
    'handlers.schedules.every_30_seconds': {
        'queue': 'schedules_queue',
        'routing_key': 'schedules_router',
    }
}

# 设定 Celery 时区
CELERY_TIMEZONE = 'Asia/Shanghai'

from datetime import timedelta
from celery.schedules import crontab

CELERYBEAT_SCHEDULE = {
	# 给计划任务取一个独一无二的名字吧
    'every-30-seconds': {
    	# task就是需要执行计划任务的函数
         'task': 'handlers.schedules.every_30_seconds',
         # 配置计划任务的执行时间，这里是每30秒执行一次
         'schedule': timedelta(seconds=30),
         # 传入给计划任务函数的参数
         'args': {'value': '2333333'}
    }
}

'push_occupancy_rates': {
            'task': 'handlers.schedules.test_func_b',
            # 每天中午12点执行
            'schedule': crontab(hour='12', minute='0'),
            'args': None
        },