Клиент

Клиент pulsar ищет каждый топик через адрес подключения (service URL), держит одно соединение на брокер и снабжает каждый консьюмер сообщениями в пределах разрешений управления потоком.

Схема текстом

1. Пул соединений: одно соединение на брокер: Producer по одному на топик; Consumer по одному на подписку; Шаблон заново получает список топиков каждые 30 с; Разрешения потока: 1000 выдано заранее, пополнение ниже 500
2. PulsarBroker::new(url) один адрес подключения → Поиск через адрес подключения: какой брокер владеет топиком: спрашивает
3. Поиск через адрес подключения: какой брокер владеет топиком → Пул соединений: одно соединение на брокер: адрес владельца
4. Пул соединений: одно соединение на брокер → Брокеры Pulsar каждым топиком владеет один: TCP
5. Операцию, которую сервер пока отвергает, клиент повторяет: по умолчанию без ограничения, раз в пять секунд

Кластер Pulsar распределяет свои топики по брокерам, и каждым топиком владеет один брокер. Клиент спрашивает у адреса подключения, какой это брокер, а затем подключается к владельцу. Для партиционированного топика клиент сначала узнаёт число партиций, а затем ищет каждую партицию отдельно. Адрес подключения - один адрес: кластер обслуживает DNS-имя, балансировщик нагрузки или прокси Pulsar перед брокерами. Адрес pulsar+ssl:// подключается к TLS-слушателю брокеров, как показывает страница Безопасность.

Клиент держит одно соединение на адрес брокера и делит его. Этот крейт открывает один продюсер на топик, когда в топик впервые что-то публикуют, и этим продюсером пользуются все издатели соединения. Каждая подписка открывает один консьюмер.

Консьюмер получает сообщения, которые присылает брокер. Он заранее выдаёт брокеру разрешения на 1000 сообщений и добавляет новые каждый раз, когда остаётся меньше половины, поэтому брокер никогда не присылает больше, чем консьюмер может вместить. Этот крейт оставляет это значение по умолчанию.

Операцию, на которую сервер отвечает «не сейчас», клиент повторяет. Таков, например, поиск во время перезапуска брокера или второй консьюмер подписки Exclusive. По умолчанию клиент повторяет без ограничения, раз в пять секунд, и даёт каждой попытке 30 секунд. OperationRetries задаёт число попыток, задержку и тайм-аут для каждой операции соединения, включая переподключение консьюмера.

Клиент выполняет свои соединения и консьюмеры как задачи рантайма tokio сервиса. Этот крейт открывает каждый консьюмер и продюсер на том рантайме, где подключился брокер, поэтому обработчик на собственном потоке не уносит соединение вместе со своим рантаймом.

Протокол взят у самого Apache Pulsar (бинарный протокол: обнаружение сервисов и управление потоком) в том виде, в каком его реализует клиент pulsar 6.9.