Varolan RabbitMQ iletilerini kullanarak Kereviz kullanma

Question

Var olan bir RabbitMQ uygulamasında, birkaç Java uygulamasının, gönderim günlüğü iletilerini çeşitli kanallarda dizgi JSON nesneleri olarak kullandığını biliyorum. Bu mesajları kullanmak ve bunları çeşitli yerlere (örneğin DB, Hadoop, vb.) Yazmak için Kereveri kullanmak istiyorum.

Kereverin, bu mesajların teslim edildiği mekanizmayı gizlemeye çalıştığı için RabbitMQ mesajlarının üreticisi ve tüketicisi olarak tasarlandığını görebiliyorum. Yine de, Celery'in başka bir uygulama tarafından yaratılan mesajları kullanması ve geldiklerinde işleri yürütmesi için var mı?

Answer 1

Şu anda özel tüketicileri kereviz işçilerine eklemek zordur, ancak bu, Tüketici önyükleme adımları için destek eklediğim geliştirme sürümünde (3.1 haline gelmek) değişmektedir.

Orada ben sadece bunu uygulamak bitirdikten olarak hiçbir belge henüz, ama burada bir örnek: API, son sürümde henüz emin değilim bir şey değişebileceğini

from celery import Celery 
from celery.bin import Option 
from celery.bootsteps import ConsumerStep 
from kombu import Consumer, Exchange, Queue 

class CustomConsumer(ConsumerStep): 
    queue = Queue('custom', Exchange('custom'), routing_key='custom') 

    def __init__(self, c, enable_custom_consumer=False, **kwargs): 
     self.enable = self.enable_custom_consumer 

    def get_consumers(self, connection): 
     return [ 
      Consumer(connection.channel(), 
       queues=[self.queue], 
       callbacks=[self.on_message]), 
     ] 

    def on_message(self, body, message): 
     print('GOT MESSAGE: %r' % (body,)) 
     message.ack() 


celery = Celery(broker='amqp://localhost//') 
celery.steps['consumer'].add(CustomConsumer) 
celery.user_options['worker'].add(
    Option('--enable-custom-consumer', action='store_true', 
      help='Enable our custom consumer.'), 
) 

Not Kanallar get_consumer(connection)'dan sonra nasıl ele alınır. Şu anda tüketicinin kanalı bağlantı koptuğunda kapanıyor ve kapatma sırasında, ancak kullanıcılar kanalları manuel olarak işlemek isteyebilir. Bu durumda, ConsumerStep'i özelleştirmenin veya yeni bir StartStopStep yazmanın her zaman olasılığı vardır.