[Часть 2.2] RabbitMQ примеры кода Python

Будем следовать сценарию, использованному в предыдущей статье  “Что такое RabbitMQ?”. Веб-сайт будет обрабатывать информацию и генерировать PDF-файл и отправлять его обратно пользователю. Создание PDF-файла и отправка электронной почты в этом сценарии займет несколько секунд. Если вы не знакомы с RabbitMQ и очередью сообщений, я бы рекомендовал вам прочитать RabbitMQ для новичков - что такое RabbitMQ? перед началом работы с этим руководством.

Начало работы с RabbitMQ и Python

Начните с загрузки клиентской библиотеки для Python. Рекомендуемой библиотекой для Python является Pika. Поместите pika == version в файл require.txt.

Для начала вам понадобится установить RabbitMQ.

• Docker image
• Установка RabbitMQ на Linux
• Установка RabbitMQ на Windows

При запуске кода будет установлено соединение между RabbiMQ и вашим приложением. Будут объявлены и созданы очереди, если они еще не существуют, и, наконец, сообщение будет опубликовано. Метод публикации будет помещать сообщения в очередь, и если соединение отключено, повторно отправит сообщение позже. Пользователь подписывается на очередь. Сообщения обрабатываются один за другим и отправляются методу обработки PDF.

Новое сообщение будет публиковаться каждую секунду. Будет использоваться обмен по умолчанию, идентифицирующий пустую строку (""). Обмен по умолчанию означает, что сообщения направляются в очередь с именем, указанным routing_key, если оно существует. (Обмен по умолчанию - это прямой обмен без имени)

1
import pika, os, logging
2
logging.basicConfig()
3

4
# Parse CLODUAMQP_URL (fallback to localhost)
5
url = os.environ.get('CLOUDAMQP_URL', 'amqp://guest:guest@localhost/%2f')
6
params = pika.URLParameters(url)
7
params.socket_timeout = 5
8

9
connection = pika.BlockingConnection(params) # Connect to CloudAMQP
10
channel = connection.channel() # start a channel
11
channel.queue_declare(queue='pdfprocess') # Declare a queue
12
# send a message
13

14
channel.basic_publish(exchange='', routing_key='pdfprocess', body='User information')
15
print ("[x] Message sent to consumer")
16
connection.close()

1
import pika, os, time
2

3
def pdf_process_function(msg):
4
  print(" PDF processing")
5
  print(" Received %r" % msg)
6

7
  time.sleep(5) # delays for 5 seconds
8
  print(" PDF processing finished");
9
  return;
10

11
# Access the CLODUAMQP_URL environment variable and parse it (fallback to localhost)
12
url = os.environ.get('CLOUDAMQP_URL', 'amqp://guest:guest@localhost:5672/%2f')
13
params = pika.URLParameters(url)
14
connection = pika.BlockingConnection(params)
15
channel = connection.channel() # start a channel
16
channel.queue_declare(queue='pdfprocess') # Declare a queue
17

18
# create a function which is called on incoming messages
19
def callback(ch, method, properties, body):
20
  pdf_process_function(body)
21

22
# set up subscription on the queue
23
channel.basic_consume(callback,
24
  queue='pdfprocess',
25
  no_ack=True)
26

27
# start consuming (blocks)
28
channel.start_consuming()
29
connection.close()

1
import pika, os, logging
2

3
# Parse CLODUAMQP_URL (fallback to localhost)
4
url = os.environ.get('CLOUDAMQP_URL', 'amqp://guest:guest@localhost/%2f')
5
params = pika.URLParameters(url)
6
params.socket_timeout = 5

Загрузите клиентскую библиотеку и настройте параметры конфигурации. Значение DEFAULT_SOCKET_TIMEOUT равно 0,25 с, мы рекомендуем повысить этот параметр примерно до 5 с, чтобы избежать таймаута соединения, params.socket_timeout = 5 Другие параметры подключения для Pika можно найти здесь: http://pika.readthedocs.io/en/latest/modules/parameters.html

Устанавливаем соединение 

1
connection = pika.BlockingConnection(params) # Connect to CloudAMQP

pika.BlockingConnection устанавливает соединение с RabbitMQ сервером.

Подключаем канал

1
channel = connection.channel()

connection.channel создает канал в TCP соединении.

Назначаем очередь

1
channel.queue_declare(queue='pdfprocess') # Declare a queue

channel.queue_declare  создает очередь, на которую будет доставлено сообщение. В очереди будет указано имя pdfprocess.

Публикация сообщения

1
channel.basic_publish(exchange='', routing_key='pdfprocess', body='User information')
2
print("[x] Message sent to consumer")

channel.basic_publish опубликует сообщение в канал с ключом маршрутизации и телом. Будет использоваться обмен по умолчанию, идентифицирующий пустую строку (""). Обмен по умолчанию означает, что сообщения направляются в очередь с именем, указанным routing_key, если оно существует. (Обмен по умолчанию - это прямой обмен без имени).

Закрываем соединение  

1
connection.close()

Consumer

1
def pdf_process_function(msg):
2
  print(" PDF processing")
3
  print(" Received %r" % msg)
4

5
  time.sleep(5) # delays for 5 seconds
6
  print(" PDF processing finished");
7
  return;

pdf_process_function такая же todo-функция как и в примере по Node.js. Функция будет спать в течение 5 секунд, чтобы имитировать создание PDF.

Функция которая будет вызываться для входящих сообщений 

1
# create a function which is called on incoming messages
2
def callback(ch, method, properties, body):
3
  pdf_process_function(body)

Функция обратного вызова будет вызываться для каждого сообщения, полученного из очереди. Функция вызовет функцию, которая имитирует обработку PDF.

1
#set up subscription on the queue
2
channel.basic_consume(callback,
3
  queue='pdfprocess',
4
  no_ack=True)

basic_consume связывает сообщения с функцией обратного вызова консьюмера.

1
channel.start_consuming() # start consuming (blocks)
2
connection.close()

start_consuming начинает читать сообщения из очереди