PIKA RABBIT INSTALL

The constructor will create an entirely new connection to Rabbit because pika is not thread safe.

In order to make use of the threading package, lets subclass the Thread class: class ThreadedConsumer(threading.Thread): def _init_(self): threading.Thread._init_(self) parameters = pika.URLParameters(RABBIT_URL) connection = pika.BlockingConnection(parameters) self.channel = connection.channel() _declare(queue=QUEUE_NAME, auto_delete=False) _bind(queue=QUEUE_NAME, exchange=EXCHANGE, routing_key=ROUTING_KEY) _qos(prefetch_count=THREADS*10) _consume(QUEUE_NAME, on_message_callback=self.callback) threading.Thread(target=_consume(QUEUE_NAME, on_message_callback=self.callback)) def callback(self, channel, method, properties, body): message = json.loads(body) time.sleep(5) print(message) channel.basic_ack(delivery_tag=livery_tag) def run(self): print ('starting thread to consume from rabbit.') _consuming() Lastly the threading package allows us to spawn threads. We will use time.sleep() to simulate i/o operations to ensure our concurrency is performing as expected. Pika is the package to interact with RabbitMQ. The messages we consume will be in JSON format so we need a parser. Next lets import all the packages we will be using: import json import pika import time import threading

THREADS is the number of threads we want to spawn to process messages. EXCHANGE is the name of the exchange we are using. QUEUE_NAME is the name of the queue we want to create and bind to. ROUTING_KEY is the name of the routing key we want our queue to receive messages from. RABBIT_URL is the connection string to the rabbit cluster. LetвЂ™s start by specifying some configurations we will need: # some configuration variables RABBIT_URL = ROUTING_KEY = 'routing_key' QUEUE_NAME = 'my_queue.' + ROUTING_KEY EXCHANGE = 'exchange_name' THREADS = 5

PIKA RABBIT INSTALL

You will also need to install pika via Pip. We will be using Python 3 for the following examples.

Pika rabbit