rabbitmq.py 文件源码

def run(self, function, args=None, kwargs=None, retry_policy=None, callback=None):
        with self._pool.acquire() as cxn:
            cxn.channel.basic_publish(
                body=self.serialize(
                    {
                        "function": function,
                        "parameters": {
                            "args": args or tuple(),
                            "kwargs": kwargs or {},
                            "retry_policy": retry_policy,
                            "callback": callback
                        }
                    }
                ),
                exchange='',
                routing_key=self.queue_name,
                properties=pika.BasicProperties(
                    delivery_mode=2,
                )
            )
            self.log(logging.DEBUG, "Task received : {}".format(function))