0

Kombuでrabbitmqキューを処理し、django関数/管理コマンドなどを呼び出しています。私の問題は、正しい実行順序が絶対に必要なことです。メッセージ 3 のハンドラは、メッセージ 1 と 2 のハンドラが終了するまで実行できません。前のメッセージの処理を完了する前に、Kombu が別のメッセージを処理しないようにする必要があります。

この基本クラスを検討してください

class UpdaterMixin(object):
    #  binding management commands to event names
    #  override in subclass
    event_handlers = {}
    app_name = ''   #override in subclass

    def __init__(self):
        if not self.app_name or len(self.event_handlers) == 0:
            print('app_name or event_handlers arent implemented')
            raise NotImplementedError()
        else:
            self.connection_url = settings.BROKER_URL
            self.exchange_name = settings.BUS_SETTINGS['exchange_name']
            self.exchange_type = settings.BUS_SETTINGS['exchange_type']
            self.routing_key = settings.ROUTING_KEYS[self.app_name]

    def start_listener(self):
        logger.info('started %s updater listener' % self.app_name)\\
        with Connection(self.connection_url) as connection:
            exchange = Exchange(self.exchange_name, self.exchange_type, durable=True)
            queue = Queue('%s_updater' % self.app_name, exchange=exchange, routing_key=self.routing_key)
            with connection.Consumer(queue, callbacks=[self.process_message]) as consumer:
                while True:
                    logger.info('Consuming events')
                    connection.drain_events()

    def process_message(self, body, message):


        logger.info('data received: %s' % body)
        handler = self.event_handlers[body['event']]
        logger.info('Executing management command: %s' % str(handler))
        data = json.dumps(body)
        call_command(handler, data, verbosity=3, interactive=False)
        message.ack()

この種の行動のために昆布を強制する方法はありますか? ロックが処理が完了するまで別のイベントを排出しないかprocess_message、前のイベントが完了するまで別のイベントを実行しないか、またはこれを達成するための他の方法であるかどうかは気にしません。実行順序が厳密に維持されていることを確認する必要があるだけです。

これについて何かお役に立てば幸いです。

4

1 に答える 1

0

Pythonはデフォルトでシングルスレッドであるため、明示的に非同期に書き直さない限り、このコードはデフォルトでブロック/同期であることがわかりました。誰かがこれにぶつかったら

于 2014-03-02T08:35:39.407 に答える