python - 工作人员没有定期从 celery-beat 中挑选任务

Question

我正在尝试测试一个简单的 Celery 应用程序。

config.py 
extract/
       celery.py
       celeryconfig.py
       tasks.py
playlists/
         playlist.json 

config.py

from os import environ

# Configs
REDIS_HOST = "0.0.0.0"
REDIS_PORT = 6379
BROKER_URL = environ.get('REDIS_URL', "redis://{host}:{port}/0".format(
    host=REDIS_HOST, port=str(REDIS_PORT)))
CELERY_RESULT_BACKEND = BROKER_URL

celery.py

from __future__ import absolute_import    
from celery import Celery
from . import celeryconfig
   
app = Celery('extract')
app.config_from_object(celeryconfig)
app.autodiscover_tasks(['extract'], force=True)

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

tasks.py

任务基本上是一个提取队列，如下所示:

@app.task(queue='extraction')
def playlist_generator_with_audio_features(username):

    playlist_ids = get_user_playlists(username)

    api = get_api_client()

    (...) # some more code

    export_playlist_with_audio_features.delay(username=username, playlist_id=playlist_id,
                                                  artist=artist, track=track, popularity=popularity,                            
                                                  energy=energy, liveness=liveness, tempo=tempo, 
                                                  speechiness=speechiness, acousticness=acousticness, 
                                                  instrumentalness=instrumentalness, time_signature=time_signature, 
                                                  danceability=danceability, key=key, duration_ms=duration_ms,
                                                  loudness=loudness, valence=valence, mode=mode, 
                                                  uri=uri, preview=preview)
   logger.info('Got audio features from {} track(s) in playlist id:{} by username {}'.format(
                                                                len(tracks_and_features),
                                                                playlist_id,
                                                                username
                                                            ))

    return True

和一个编写器队列，如下所示:

@app.task(queue='writer')
def export_playlist_with_audio_features(username, playlist_id, artist, track, popularity, 
                                        energy, liveness,tempo, speechiness, acousticness,
                                        instrumentalness, time_signature, danceability,
                                        key, duration_ms, loudness, valence, mode,
                                        uri, preview):

    path = 'playlists/playlist.json'

    obj = json.dumps({
        'username':username,
        'playlist_id': playlist_id,
        'artist': artist,
        'track': track,
        'popularity': popularity,
        'energy': energy,
        'liveness': liveness,
        'tempo': tempo,
        'speechiness': speechiness,
        'acousticness': acousticness,
        'instrumentalness': instrumentalness,
        'time_signature': time_signature,
        'danceability': danceability,
        'key': key,
        'duration_ms': duration_ms,
        'loudness': loudness,
        'valence': valence,
        'mode': mode,
        'uri': uri,
        'preview': preview
    })

    codecs.open(path, 'a', encoding='utf-8').write(obj + '\n')

    logger.debug('Exported track {} by {} with audio features. Playlist owner: {}'.format(
                                                                    track, artist, username))
 
    return True

celeryconfig.py

from __future__ import absolute_import, unicode_literals
import os
from celery.schedules import crontab
    
INSTALLED_APPS = ['app'] # include app
CELERY_TASK_RESULT_EXPIRES = 30
CELERY_TIMEZONE = 'UTC'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_IMPORTS = ('extract.tasks')

DEFAULT_URL = 'redis://localhost:6379/0'

BROKER_URL = os.environ.get('EXTRACT_BROKER_URL', DEFAULT_URL)
CELERY_RESULT_BACKEND = os.environ.get('EXTRACT_RESULT_BACKEND',
                                       DEFAULT_URL)

CELERYBEAT_SCHEDULE = {
    'playlist_generator_with_audio_features': {
        'task': 'extract.tasks.playlist_generator_with_audio_features',
        # Every minute
        'schedule': crontab(minute="*"),
        'args' : [(1),]
    },
    'export_playlist_with_audio_features': {
        'task': 'extract.tasks.export_playlist_with_audio_features',
        # Every minute
        'schedule': crontab(minute="*"),
        'kwargs' : {"username":"username", 
                   "playlist_id": "playlist_id", 
                   "artist": "artist",
                   "track": "track",
                   "popularity": "popularity",
                   "energy":"energy",
                   "liveness": "liveness",
                   "tempo": "tempo",
                   "speechiness": "speechiness",
                   "acousticness":"acousticness",
                   "instrumentalness":"instrumentalness",
                   "time_signature":"time_signature",
                   "danceability":"danceability",
                   "key": "key",
                   "duration_ms": "duration_ms",
                   "loudness": "loudness",
                   "valance": "valance",
                   "mode":"mode",
                   "uri":"uri",
                   "preview": "preview"},
        }
       }

Celery 工作人员自行工作，即任务将 json 序列化数据转储到 playlists/ 中，使用以下命令:

$celery worker -A提取-l信息-Q提取

      [2018-12-16 23:31:58,075: WARNING/MainProcess] celery@Vitors-MacBook-Pro-2.local ready.
[2018-12-16 23:32:29,494: INFO/MainProcess] Received task: extract.tasks.playlist_generator_with_audio_features[0713b788-c004-4c0c-b1ee-85068287856a]
[2018-12-16 23:32:30,892: INFO/Worker-4] extract.tasks.get_user_playlists[None]: Playlist name: Discover Weekly, Number of songs: 30, Playlist ID: 37i9dQZEVXcRI2aS94C6hY 
[2018-12-16 23:32:30,893: INFO/Worker-4] extract.tasks.get_user_playlists[None]: Playlist name: Time, Number of songs: 10, Playlist ID: 1qxOGf3xD3fDQs03kwKteg 
[2018-12-16 23:32:30,893: INFO/Worker-4] extract.tasks.get_user_playlists[None]: Playlist name: Vocoder, Number of songs: 280, Playlist ID: 7nROSBznyIkVgjSdNaHDxm 

和

$ celeryworker -A extract -l info -Q writer

       [2018-12-16 23:32:14,198: WARNING/MainProcess] celery@Vitors-MacBook-Pro-2.local ready.
[2018-12-16 23:32:32,467: INFO/MainProcess] Received task: extract.tasks.export_playlist_with_audio_features[a73f776d-2412-4aed-8e84-50c6485ebac8]
[2018-12-16 23:32:32,470: INFO/MainProcess] Received task: extract.tasks.export_playlist_with_audio_features[930fb552-5687-48ce-b61d-b573a79aa13c]
[2018-12-16 23:32:32,479: INFO/MainProcess] Received task: extract.tasks.export_playlist_with_audio_features[162a77a1-20ba-48c3-86c3-bb9c51eeed75]
[2018-12-16 23:32:32,486: INFO/MainProcess] Received task: extract.tasks.export_playlist_with_audio_features[89eff015-e692-4af0-a7a8-dcad06dc2e3f]

问题

但是当我尝试使用celerybeat来执行任务时，工作人员仅选择第一轮任务，而不是定期选择。

celerybeat -A app.celery --schedule=/tmp/celerybeat-schedule --loglevel=INFO --pidfile=/tmp/celerybeat.pid

Configuration ->
    . broker -> redis://localhost:6379/0
    . loader -> celery.loaders.app.AppLoader
    . scheduler -> celery.beat.PersistentScheduler
    . db -> /tmp/celerybeat-schedule
    . logfile -> [stderr]@%INFO
    . maxinterval -> now (0s)
[2018-12-16 23:32:32,706: INFO/MainProcess] beat: Starting...
 [2018-12-16 23:32:32,736: INFO/MainProcess] Scheduler: Sending due task export_playlist_with_audio_features (tasks.export_playlist_with_audio_features)
[2018-12-16 23:32:32,743: INFO/MainProcess] Scheduler: Sending due task playlist_generator_with_audio_features (tasks.playlist_generator_with_audio_features)
[2018-12-16 23:33:00,000: INFO/MainProcess] Scheduler: Sending due task playlist_generator_with_audio_features (tasks.playlist_generator_with_audio_features)

我在这里缺少什么？

Answer 1

@app.task(queue='extraction')

此处的选项不会被beat使用。使用 @periodic_task 装饰器来进行节拍，并且常规 celery 客户端使用特定的队列。

@periodic_task(run_every=crontab(minute='5'),
           queue='celery_periodic',
           options={'queue': 'celery_periodic'})

• queue='celery_periodic' 选项在您从代码(.delay 或 .apply_async)调用任务时使用

• options={'queue': 'celery_periodic'} 选项在 celerybeat 调用它时使用。

您还可以在时间表配置中提供节拍选项，例如

CELERYBEAT_SCHEDULE = {
   'playlist_generator_with_audio_features': {
        'task': 'extract.tasks.playlist_generator_with_audio_features',
        'schedule': crontab(minute="*"),
        'options': {'queue' : 'celery_periodic'} # specify queue
    },
}