如何使用rabbitmq隊列從django視圖設置芹菜延遲任務

Question

我必須在django項目中使用rabbitmq在芹菜中創建任務。基本上這個項目需要發送推送通知。有兩個可選的管理人員發送推送通知

1）立即：當他們點擊推送通知發送即時

2）發送後：在這種情況下，系統管理員所設定的日期和時間並且我需要 僅在該日期和時間發送推送通知。

我tasks.py是

from celery.task import task 
from celery.task.schedules import crontab 
from datetime import timedelta 
from celery.decorators import periodic_task 

#@task 
@periodic_task(run_every=(crontab(minute='*/1'))) 
def multiply(x, y): 
    multiplication = x * y 
    print('print me here') 
    return multiplication 

celery.py是

from __future__ import absolute_import 
import os 
from celery import Celery 
from django.conf import settings 

# set the default Django settings module for the 'celery' program. 
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'mysite.settings') 
app = Celery('mazda') 

# Using a string here means the worker will not have to 
# pickle the object when using Windows. 
app.config_from_object('django.conf:settings') 
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS) 


#s@app.task(bind=True) 
def debug_task(self): 
    print('Request: {0!r}'.format(self.request)) 

，我caling從視圖像result = multiply.delay(3,5)

任務，當我運行命令celery -A {myappname} worker -l info我可以看到在芹菜上創建了任務，另外一個隊列也在rabbitmq隊列上生成（在這裏，我無法檢查相同的任務是否創建或不）

現在我的問題是如何調用這個任務的時間和日期管理員選擇從管理端。所有文件都說延遲功能，沒有任何東西給我設置任務的確切延遲的選項。請幫助我。

settings.py文件是

import os 
#import djcelery 
#djcelery.setup_loader() 
# Build paths inside the project like this: os.path.join(BASE_DIR, ...) 
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) 


# Quick-start development settings - unsuitable for production 
# See https://docs.djangoproject.com/en/1.10/howto/deployment/checklist/ 

# SECURITY WARNING: keep the secret key used in production secret! 
SECRET_KEY = 'bjn4ngz(mk*$*(z12jk+ztf4b*-p@b42l#)rind!cl)$%z49h^' 

# SECURITY WARNING: don't run with debug turned on in production! 
DEBUG = True 

ALLOWED_HOSTS = [] 


# Application definition 

INSTALLED_APPS = [ 
    'mazda.apps.MazdaConfig', 
    'django.contrib.admin', 
    'django.contrib.auth', 
    'django.contrib.contenttypes', 
    'django.contrib.sessions', 
    'django.contrib.messages', 
    'django.contrib.staticfiles', 
    'push_notifications', 
    'djcelery' 
    #'djng', 
] 
PUSH_NOTIFICATIONS_SETTINGS = { 
     "GCM_API_KEY": "AIzaSyC4YNDJBVJK75QEVWuMKPXuTgiIEfa26Pw", 
     "APNS_CERTIFICATE": "pushLatestPushFile.pem", 
    # "WNS_PACKAGE_SECURITY_ID": "", 
    # "WNS_SECRET_KEY": "", 
} 
MIDDLEWARE = [ 
    'django.middleware.security.SecurityMiddleware', 
    'django.contrib.sessions.middleware.SessionMiddleware', 
    'django.middleware.common.CommonMiddleware', 
    'django.middleware.csrf.CsrfViewMiddleware', 
    'django.contrib.auth.middleware.AuthenticationMiddleware', 
    'django.contrib.messages.middleware.MessageMiddleware', 
    'django.middleware.clickjacking.XFrameOptionsMiddleware', 
] 

ROOT_URLCONF = 'mysite.urls' 

TEMPLATES = [ 
    { 
     'BACKEND': 'django.template.backends.django.DjangoTemplates', 
     'DIRS': [], 
     'APP_DIRS': True, 
     'OPTIONS': { 
      'context_processors': [ 
       'django.template.context_processors.debug', 
       'django.template.context_processors.request', 
       'django.contrib.auth.context_processors.auth', 
       'django.contrib.messages.context_processors.messages', 
      ], 
     }, 
    }, 
] 

WSGI_APPLICATION = 'mysite.wsgi.application' 


# Database 
# https://docs.djangoproject.com/en/1.10/ref/settings/#databases 

DATABASES = { 
    'default': { 
     'ENGINE': 'django.db.backends.postgresql', 
     'NAME': 'django', 
     'USER': 'postgres', 
     'PASSWORD': 'admin123', 
     'HOST': '', 
     'PORT': '', 
    } 
} 


# Password validation 
# https://docs.djangoproject.com/en/1.10/ref/settings/#auth-password-validators 

AUTH_PASSWORD_VALIDATORS = [ 
    { 
     'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator', 
    }, 
    { 
     'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator', 
    }, 
    { 
     'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator', 
    }, 
    { 
     'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator', 
    }, 
] 


# Internationalization 
# https://docs.djangoproject.com/en/1.10/topics/i18n/ 

LANGUAGE_CODE = 'en-us' 

TIME_ZONE = 'UTC' 

USE_I18N = True 

USE_L10N = True 

USE_TZ = True 


# Static files (CSS, JavaScript, Images) 
# https://docs.djangoproject.com/en/1.10/howto/static-files/ 

STATIC_URL = '/static/' 
LOGIN_URL = '/mazda/userlogin' 

BROKER_URL = 'amqp://guest:guest@localhost:5672//' 
CELERY_TIMEZONE = 'Europe/London' 

Answer 1

使用apply_async()與eta參數（ETA必須是日期時間對象，指定一個確切的日期和時間，包括毫秒精度和時區信息）：

>>> from datetime import datetime, timedelta 

>>> tomorrow = datetime.utcnow() + timedelta(days=1) 
>>> multiply.apply_async((3, 5), eta=tomorrow) 

芹菜文檔中的更多信息here。