rabbitmqからセロリメッセージをフォーマットする方法がわからない

Question

ガイドとして使用する：http://docs.celeryproject.org/en/latest/internals/protocol.html#definition セロリプログラムにメッセージを送信しようとしていますが、メッセージのフォーマット方法はわかりません。私が何をしようと、私はKeyError: 'chord'またはValueError('not enough values to unpack (expected 3, got 0)'のいずれかである間違ったエラーをセロリから得る。これらのメッセージのフォーマット方法に関する情報はどこにありますか。私はちょうどrabbitmq Webインターフェイスからこれを達成しようとしています。それは実際にタスクを実行するために

：私はここに入力する必要が最低限で何

from celery import Celery 
import time 

app = Celery('tasks', backend='rpc://', broker='pyamqp://ycunrxpq:nU31gS9yTlbFNC0vXJFpU7P9GxmzHkk8@orangutan.rmq.cloudamqp.com/ycunrxpq') 

@app.task 
def random_task(): 
    print('this task is random') 

：

これは私のセロリのアプリがどのように見えるかです。

Answer 1

私はあなたのために役立つかもしれませんが、私は同じような問題があったことを覚えています。

プロパティに 'content_type'、 'content_encoding'、ヘッダーに 'η'と 'タスク名'（タスクなし）を渡すだけです。

また、kwargsとargsは必須であることを覚えています。

これが私の仕事の例

app = Celery('tasks', broker='pyamqp://logstash:logstash@rabbitmq.local:5672') 

app.conf.task_queues = (
    Queue('logstash_events', Exchange('logstash_events'), routing_key='logstash_events'), 

) 

app.conf.task_default_queue = 'logstash_events' 
app.conf.task_default_exchange_type = 'direct' 
app.conf.task_default_routing_key = 'logstash_events' 

@app.task(serializer='json', name='process_next_task', bind=True) 
def process_next_task(self,*args,**kwargs): 
    self.message = kwargs.get('message',"null") 
    self.geodata = kwargs.get('geodata',"null") 
    print(self.message) 
    print(self.geodata) 

これらは私がLogstashから渡しているが、私は同じだと思うのparams（Logstashでルビー）あるあります。

event.set('task', 'process_next_task') 
event.set('lang', 'py') 
event.set('origin', '@logstash') 
event.set('id', event.get('fingerprint')) 
event.set('args', {:empty => 'null'}) 
event.set('kwargs', { 
    :message => { 
    :data => event.get('data'), 
    :event_type => event.get('event_type'), 
    }, 
    :geodata => event.get('geoip') 
})