I am currently looking to develop a synchronous workflow in a RabbitMQ producer-consumer setup. The producer will publish a message to the queue and it will take about 6 - 8 seconds for the consumer to finish processing and return the result to consumer. The producer will then get the result and continue to execute the rest of the code, which requires the result from the consumer.
I understand Message Queues are designed to be synchronous, but is there any method to implement the process producer call -> consumer process -> consumer return result to producer synchronous? I searched around and found the tutorial from RabbitMQ docs, where it used a busy loop to do the waiting on the producer side:
rpc_client.py (producer)
#!/usr/bin/env python
import pika
import uuid
class FibonacciRpcClient(object):
def __init__(self):
self.connection = pika.BlockingConnection(
pika.ConnectionParameters(host='localhost'))
self.channel = self.connection.channel()
result = self.channel.queue_declare(queue='', exclusive=True)
self.callback_queue = result.method.queue
self.channel.basic_consume(
queue=self.callback_queue,
on_message_callback=self.on_response,
auto_ack=True)
def on_response(self, ch, method, props, body):
if self.corr_id == props.correlation_id:
self.response = body
def call(self, n):
self.response = None
self.corr_id = str(uuid.uuid4())
self.channel.basic_publish(
exchange='',
routing_key='rpc_queue',
properties=pika.BasicProperties(
reply_to=self.callback_queue,
correlation_id=self.corr_id,
),
body=str(n))
while self.response is None:
self.connection.process_data_events()
return int(self.response)
fibonacci_rpc = FibonacciRpcClient()
print(" [x] Requesting fib(30)")
response = fibonacci_rpc.call(30)
print(" [.] Got %r" % response)
rpc_server.py (consumer)
#!/usr/bin/env python
import pika
connection = pika.BlockingConnection(
pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='rpc_queue')
def fib(n):
if n == 0:
return 0
elif n == 1:
return 1
else:
return fib(n - 1) + fib(n - 2)
def on_request(ch, method, props, body):
n = int(body)
print(" [.] fib(%s)" % n)
response = fib(n)
ch.basic_publish(exchange='',
routing_key=props.reply_to,
properties=pika.BasicProperties(correlation_id = \
props.correlation_id),
body=str(response))
ch.basic_ack(delivery_tag=method.delivery_tag)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='rpc_queue', on_message_callback=on_request)
print(" [x] Awaiting RPC requests")
channel.start_consuming()
But I do not think the busy loop while self.response is None: is a very good idea, since it adds a very heavy load to the server running this code. I tried to use a lock to acquire the lock on the producer side and send the lock object to the consumer side, so the consumer can release the lock when the processing is done and the lock is released. However, the RabbitMQ client for Python, Pika, does not allow passing lock objects across the queue.
What will be an elegant way to avoid busy-wait to achieve the synchronous workflow here?