предисловие

Очереди сообщений обеспечивают асинхронную связь между приложениями путем отправки сообщений. Короче говоря, производительность отправителей и потребителей обычно непоследовательна, поэтому нам нужна абстрактная модель для разделения, поэтому здесь можно ввести очереди сообщений. , временно запишите задачу в промежуточное программное обеспечение сообщений. и подождите, пока потребитель медленно его обработает. Уже существует множество вариантов промежуточного программного обеспечения для сообщений, таких как RocketMQ, Kafka, Pulsar и т. д. Хотя очередь сообщений приносит много удобства, она также представляет некоторую техническую сложность, как черный ящик, если вы не можете понять его принцип, если вы столкнулись с проблемой, проверять будет очень больно.Сегодня мы рассмотрим, как реализовать простую очередь сообщений.

текст

Во-первых, давайте посмотрим на структуру пакетов Kafka и RocketMQ, чтобы увидеть, какие компоненты необходимы для распределенной очереди сообщений.

уровень хранения

Одним из основных компонентов очереди сообщений является уровень хранения.Важным моментом здесь является то, как получить и прочитать сообщение.Например, RocketMQ и Kafka выбирают хранение на локальном компьютере, то есть в локальной файловой системе.Pulsar выбирает для хранения в распределенной файловой системе bookKeeper.Конечно, некоторые выбирают распределенные системы KV или даже базы данных.Например, сам Redis также поддерживает модель публикации/потребления.Конкретный метод реализации зависит от вас.Бизнес-сценарии, такие как сценарии которые требуют высокой надежности, но не столь чувствительны к производительности, могут выбрать в качестве носителя данных базу данных.
Выбор локальной файловой системы для реализации распределенной очереди сообщений является относительно наиболее сложным из них.Состояние системы восстановления журналов также необходимо реализовывать самостоятельно, и каждая из этих частей требует значительного количества энергии для изучения. время, мы только начинаем с относительно простого прототипа, и мы будем работать над этим решением, когда у нас будет время.
Решение на основе распределенного KV также является относительно хорошим решением, производительность очень хорошая, а интерфейс более удобный, но надежность немного хуже, оно не подходит для таких сценариев, как транзакции и синхронизация кеша, которые требуют высокая надежность.Тогда используйте.

Подход, основанный на базе данных, понесет большие потери в производительности, а структура данных БД по своей сути не подходит для реализации очередей сообщений. На этот раз мы решили использовать в качестве носителя данных распределенную файловую систему, такую как HDFS, Apache BookKeeper и т. д. Давайте проанализируем сценарий очереди сообщений, однопоточной записи-многопоточного чтения, концепция разделения темы должна быть представлен здесь. Как правило, если какая-то тема относительно активна и пропускная способность относительно высока, то мы можем разделить сообщение. Идея реализации, как правило, состоит в том, чтобы разделить тему на подтемы с высокой степенью детализации и распределить каждую подтему по различных брокеров, чтобы добиться линейного улучшения производительности. То есть однопоточная запись здесь относится конкретно к одному разделу. Многопоточное чтение относительно легко понять, и HDFS как раз подходит для этого сценария, и мы не не нужно беспокоиться о репликах, написании осколков, стратегиях очистки и т. д. Снижает сложность реализации, BookKeeper — хороший выбор в этом отношении.

Реализация клиентского API

Для пользователей они больше подвержены API, предоставляемому клиентом, а брокеру клиента и сервера также нужен способ связи.Для RocketMQ и Kafka они выбирают реализацию пользовательских протоколов, очередей сообщений.Если вы хотите достичь чрезвычайно высокой пропускной способности , очень важно реализовать высокопроизводительную структуру сетевого взаимодействия. RocketMQ построен поверх Netty, в то время как Kafka напрямую основан на относительно сложном NIO. Один момент, если вы прочитали исходный код, вы поймете , После отправки клиента Kafka он помещается в локальную очередь, а затем объединяется и отправляется на сервер в соответствии с брокером, темой, информацией о разделе и т. д., в то время как впечатление Pulsar основано на протоколе. Он реализован с помощью буфера, который имеет много преимуществ по сравнению с пользовательскими протоколами. Прежде всего, если процесс реализации протокола изменится позже, детали того, как быть совместимым со старым протоколом, были решены с помощью буфера протокола. Еще один важный момент: этот буфер протокола может вам помочь. Вы создаете API на разных языках. Если это пользовательский протокол, для его реализации потребуются значительные усилия.

последовательность

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

Высокая доступность

Во-первых, на уровне хранения наш выбор технологии уже определился как высокодоступный, потому что сам BookKeeper поддерживает указанный механизм, скопированный на несколько подчиненных устройств, и подтверждение, например, необходимость записи всех разделов только для того, чтобы клиенты возвращали успех, и для конца брокера, как мы рассчитываем, очередь сообщений хранится и изолирована, т. е. сам брокер не имеет состояния, отключен или когда тайм-аут соединения поставщика / потребителя сети брокера отключен, может затем напрямую обслуживаться другим брокером, конечно, есть много деталей, но относительная сложность RocketMQ уже сильно уменьшилась.

потребительское хранилище прогресса

Мы знаем, что сообщение имеет три семантики: не более одного раза, не менее одного раза, ровно один раз, тогда хранение смещения потребителя в механизме синхронизации в определенной степени определяет нашу конкретную семантику, например отправителя, если отправка не удалась, сделать не повторять Не более одного раза. Если отправка не удалась и выбрать определенное количество повторных попыток, то это хотя бы один раз. Это может привести к повторному размещению сообщения на диске и, таким образом, к повторному потреблению. Например, сообщение на самом деле был помещен на диск, но происходит процесс отправки ответа на отправку.Это происходит, когда сеть ненормальна, и сценарий точного времени немного сложнее. Вернемся к сценарию смещения: по умолчанию RocketMQ и Kafka регулярно синхронизируют текущий прогресс потребления, так что где хранить прогресс потребления — это другой вопрос. Метод RocketMQ заключается в том, чтобы хранить его в локальной файловой системе. Кафка предпочел сохранить его в Zookeeper до версии 0.8, а затем изменил его на другую тему. Итак, каковы преимущества и недостатки этих двух методов:

Производительность/горизонтальное масштабирование: Zookeeper — это согласованная система, и поддерживаемые ею API также основаны на форматах ключ/значение.ZK по существу не поддерживает большое количество операций записи, а ZK не поддерживает горизонтальное масштабирование, поскольку каждый узел будет синхронизировать все транзакцию и сохранить весь набор данных, фактически ZK — это распределенная система, основанная на записи одного журнала и синхронно реплицируемой на другие узлы. Согласно моему тесту, пропускная способность ZK составляет около 1 Вт/с, но если у нас есть десятки миллионов топиков, а прогресс потребления синхронизируется каждую секунду, ZK уже не может удовлетворить потребности в это время, и он не может масштабироваться по горизонтали. , Это можно решить с помощью шардинга, который вводит прокси-уровень.
Сложность реализации: хотя производительность, основанная на локальной файловой системе, значительна, она такая же, как у хранилища сообщений, и необходимо учитывать многие детали реализации.

Поэтому мы ссылаемся на последнюю реализацию Kafka здесь. Мы решили хранить прогресс потребления в BookKeeper, который может поддерживать большое количество записей и поддерживать линейное расширение. BK также будет объединять и хранить небольшие журналы в одном файле, избегая производительности. Затронуты некоторые неактивные темы.

Суммировать

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