Как реализовать очередь в Redis

Инструкция про то, как можно использовать базу данных Redis в качестве брокера сообщений.

Введение

В статье расскажем, как можно использовать базу данных Redis в качестве брокера сообщений, рассмотрим различные способы сделать это и покажем как реализовать один из них на практике с помощью Python.

Redis как брокер сообщений

Redis чаще всего используется как кэширующая БД для ускорения работы с данными. Но также в ней есть функциональность брокера сообщений, с помощью которого можно реализовать очереди, например, как в Kafka или RabbitMQ.

Если вы хотите освежить память – у нас есть отдельная статья о брокерах сообщений. Там мы рассказываем об очередях на примере Kafka, но для общего понимания это не так важно.

У Redis есть несколько возможностей, с помощью которых можно реализовать очередь сообщений. У всех есть свои особенности, преимущества и недостатки.

Pub/Sub — механизм синхронной доставки сообщений. Publisher отправляет сообщение в очередь, и все подписанные на нее subscriber-ы получают это сообщение. Самый большой недостаток этого способа: если в момент публикации сообщения нет ни одного подписчика, то сообщение никто не получит и оно потеряется.
List — Очередь по типу FIFO. Publisher отправляет сообщение в очередь, но его получит лишь один из subscriber-ов. При этом если подписчик во время обработки сообщения аварийно завершит работу — то сообщение потеряется, т.к. оно уже вышло из очереди.
Stream — возможность, появившаяся в Redis 5.0. Похож на механизм Pub/Sub, но с гарантией доставки сообщений. Есть возможность удаления сообщений из очереди, если subscriber принял его и обработал.

В этой инструкции мы рассмотрим механизм Pub/Sub и реализуем его на Python.

Создание кластера Redis

Создадим кластер Redis в облачной платформе Selectel. Это позволит нам не заниматься настройкой и установкой БД Redis, а сразу приступить к реализации очереди сообщений.

Заходим в раздел Облачная платформа — Базы данных и нажимаем кнопку Создать кластер.

Выбираем Redis 6. Для наших целей будет достаточно небольшого кластера с 2 vCPU и 4ГБ оперативной памяти. Если вам необходима отказоустойчивость — добавьте одну или две реплики. В этом случае если мастер-нода выйдет из строя, кластер автоматически переключится на одну из реплик.

Политику вытеснения выберем noeviction, но вы можете использовать другой необходимый вам вариант. Не забудьте сохранить пароль пользователя, он пригодится для подключения к базе данных. Также обратите внимание, чтобы кластер был расположен в публичной сети, чтобы к нему можно было подключиться из интернета.

После создания кластера зайдем в него настройки, в раздел Подключение.

Скопируйте и сохраните DNS-имя кластера для подключения. Мы будем использовать его в следующем разделе.
На этой же вкладке написано, что для подключения по SSL понадобится CA-сертификат. Его нужно скачать и сохранить на той машине, с которой будем подключаться к серверу.

Реализация очереди сообщений Pub/Sub на Python

Наш кластер готов к приему сообщений. Мы создадим два простых сервиса: первый будет публиковать сообщения в канал, а второй – получать их.

Для начала установим Python-модуль, который необходим для работы с Redis:


    pip install redis

Начнем с подписчика. Создайте файл subscriber.py со следующим содержимым:


    import redis
import time

r = redis.Redis(
  host="<REDIS_URL>",
  password="<REDIS_PASSWORD>",
  port=6380,
  db=0,
  ssl=True,
  ssl_ca_certs="<CA_PATH>",
)

p = r.pubsub()
p.subscribe("channel_1", "channel_2")

while True:
  message = p.get_message()
  if message:
   print(message)
  time.sleep(0.01)

Вместо <REDIS_URL> подставьте путь до вашего кластера.
Вместо <REDIS_PASSWORD> укажите пароль.
Вместо <CA_PATH> укажите путь до CA-сертификата, который должен быть скачан.

Итак, сначала мы создаем подключение к серверу Redis, указывая все необходимые данные. Затем создаем subscriber-а, который подписывается на каналы channel_1 и channel_2. Затем в бесконечном цикле subscriber следит за наличием новых сообщений в каналах, и если там появляется новое сообщение – выводит его в консоль. При этом задаем небольшой таймаут ожидания.

Запустим этот скрипт командой python subscriber.py. Мы увидим, что subscriber подключился к Redis и готов принимать сообщения:


    {'type': 'subscribe', 'pattern': None, 'channel': b'channel_1', 'data': 1}
{'type': 'subscribe', 'pattern': None, 'channel': b'channel_2', 'data': 2}

Перейдем к publisher-у. Создадим файл publisher.py и запишем в него следующий код:


    import redis

r = redis.Redis(
  host="<REDIS_URL>",
  password="<REDIS_PASSWORD>",
  port=6380,
  db=0,
  ssl=True,
  ssl_ca_certs="<CA_PATH>",
)

r.publish('channel_1', 'This is message from channel 1')
r.publish('channel_2', 'This is message from channel 2')

Сначала мы создаем подключение к Redis аналогично subscriber-у. Затем в каждый канал публикуем по одному сообщению. Откроем второй терминал и запустим скрипт: python publisher.py. В окне с запущенным подписчиком мы должны увидеть следующее:


    {'type': 'message', 'pattern': None, 'channel': b'channel_1', 'data': b'This is message from channel 1'}
{'type': 'message', 'pattern': None, 'channel': b'channel_2', 'data': b'This is message from channel 2'}

Теперь давайте немного изменим файл subscriber.py, чтобы показать возможность использования шаблонов. Вместо строки


    p.subscribe("channel_1", "channel_2")

напишем


    p.subscribe("channel_*")

Тем самым мы подписываемся на все каналы, попадающие под маску channel_*. Снова запустим subscriber и publisher. В консоли подписчика увидим следующее:


    {'type': 'pmessage', 'pattern': 'channel_*', 'channel': b'channel_1', 'data': b'This is message from channel 1'}
{'type': 'pmessage', 'pattern': 'channel_*', 'channel': b'channel_2', 'data': b'This is message from channel 2'}

Мы видим, что снова получили два сообщения из обоих каналов, хотя явно не указывали из каких именно. Так сработала подписка по шаблону.

Заключение

Мы рассмотрели использование Redis в качестве брокера сообщений различными способами. Рассмотрели механизм Pub/Sub и опробовали его на практике, создав publisher-а и subscriber-а на Python.

Базы данных

Как реализовать очередь в Redis

Введение

Redis как брокер сообщений

Создание кластера Redis

Redis в облаке

Реализация очереди сообщений Pub/Sub на Python

Заключение

Зарегистрируйтесь в панели управления

Читайте также: