- 00000000000000000000

    - 00000000001073741824

-- queue id

    -- 00000000000000000000

    -- 00000000000000001040

    -- 00000000000000002080

{
    "topicConfigTable": {
        "topicConfigTable": {
            "%RETRY%consumerGroupId-example-200": {
                "SEPARATOR": "",
                "topicName": "%RETRY%consumerGroupId-example-200",
                "readQueueNums": 1,
                "writeQueueNums": 1,
                "perm": 6,
                "topicFilterType": 0,
                "topicSysFlag": 0,
                "order": false
            }
        }
    },
    "dataVersion": {
        "timestamp": 1511333414604049700,
        "counter": 2023
    }
}

В тексте перевода были исправлены некоторые грамматические ошибки и неточности. Потребитель поддерживает SubscriptionGroup через сервис сердцебиения.

* **Персистентность**

Каждый раз, когда Broker изменяет отношения SubscriptionGroup, происходит процесс персистентности. Smargo сохраняет путь к файлу subscriptionGroup в каталоге текущего пользователя /store/config/subscriptionGroup.json. Структура хранения выглядит следующим образом:

{
    topicName: {
        "groupName": "xx", // имя группы подписки
        "consumeEnable": true, // возможность потребления
        "consumeFromMinEnable": true, // разрешение на потребление с минимального положения очереди, по умолчанию в рабочей среде установлено значение false
        "consumeBroadcastEnable": true, // разрешение на широковещательное потребление
        "retryQueueNums": 1, // количество очередей повтора для сообщений при неудачном потреблении
        "retryMaxTimes": 16, // максимальное количество попыток повторного потребления, превышение которого приводит к отправке сообщения в очередь недоставленных сообщений, прекращению повторных попыток и генерации оповещения
        "brokerId": 0,  // брокер, с которого начинается потребление
        "whichBrokerWhenConsumeSlowly": 0  // при обнаружении накопления сообщений перенаправление запросов на потребление от Consumer на другую Slave-машину
    },
    "dataVersion": {
        "timestamp": 1511342161274071800,
        "counter": 3
    }
} Весь процесс отправки сообщения делится на два этапа:

* **Consumer** отправляет сообщение обратно.
  * В случае, если сообщение не было успешно обработано на стороне Consumer, оно отправляется в очередь повторных попыток. Broker получает сообщение и проверяет следующее:
    1. Существует ли группа подписки для текущего сообщения.
    2. Есть ли у текущего Broker разрешение на запись.
    3. Получает Topic очереди повторных попыток (обычно это %RETRY% + groupName) и вычисляет QueueID.
    4. Если количество попыток обработки сообщения больше установленного значения, сообщение отправляется в очередь недоставленных сообщений (Topic обычно имеет вид %DLQ% + GroupName) и больше не обрабатывается.
    5. Если количество попыток меньше установленного значения, текущая попытка увеличивается на 3, и обработка сообщения откладывается.
    6. Сообщение повторно упаковывается, после чего вызывается store-сервис.
    7. Проводится статистика.

* **Producer** отправляет обычное сообщение.
  * Отправка обычных сообщений гораздо проще, чем отправка сообщений для повторной попытки. Процесс отправки обычного сообщения включает следующие шаги:
    1. Проверяется законность отправки сообщения (например, проверяется соответствие Topic и наличие прав доступа у Broker).
    2. Сообщение заново упаковывается и вызывается store-сервис.
    3. Проводится статистика.

1. Обработка сообщения потребителем.
  * На стороне потребителя существует два способа обработки сообщений: push (активная отправка) и pull (пассивный запрос). Независимо от того, используется ли push или pull, взаимодействие с Broker осуществляется через операцию pull, инициируемую потребителем. Основной процесс выглядит следующим образом: 4.7Hold

- **Назначение**

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

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

- **Процесс Hold**

Клиентский конец запрашивает у брокера сообщение с определённым смещением. Если смещение больше максимального смещения, это означает, что сообщение ещё не сохранено в брокере и его нельзя получить. Запрос на получение помещается в очередь ожидания, ожидая пробуждения при поступлении нового сообщения. Последовательность запросов Hold показана ниже. 4.8 Сообщение статистики

- Назначение

Broker собирает статистику по количеству и объёму сообщений, помещённых в Topic и Group, а также полученных из них, с учётом таких параметров, как время, источник сообщения и потребитель. В режиме реального времени производится расчёт и запись соответствующего Tps. На основе статистических данных пользователи могут получить представление об объёме своего бизнеса и его колебаниях, а администраторы — оценить текущую пропускную способность кластера Broker и принять решение о необходимости расширения ресурсов.

- Процесс обработки

При запуске Broker инициализирует службу статистики. При успешном получении или сохранении сообщения Broker увеличивает соответствующие значения количества и объёма на основе исходных статистических данных, а также рассчитывает и записывает Tps для разных временных интервалов. График временной шкалы статистической информации о сообщениях выглядит следующим образом: 4.9Producer, Consumer соединение

- Назначение

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

- Процесс соединения

Клиенты через реестр запрашивают список маршрутизации брокера, перебирают список маршрутизации и находят адрес брокера для соответствующей бизнес-темы. На основе адреса брокера клиенты соединяются с брокером и периодически отправляют сердцебиение для поддержания соединения канала. После получения сердцебиения брокер в реальном времени поддерживает канал, тему, группу и список подписчиков. Последовательность управления соединением клиента показана на схеме ниже: 1.
  1. Структура пакета связи:
    - length домен: длина сообщения.
    - field домен: содержание сообщения. **Длина**: длина сообщения.

**Длина заголовка**: длина заголовка сообщения.

**Заголовок**: заголовок сообщения.

**Тело**: содержание сообщения. 

Заголовок и тело сообщения после анализа данных представляют собой структуру RemotingCommand, которая при коммуникации сериализуется в массив байтов byte[] для передачи на уровне связи.

1.  
   1.  
     1. **4.5.1 Структура RemotingCommand**

| **Поле** | **Запрос** | **Ответ** |
| --- | --- | --- |
| code | Код операции запроса, на основе которого получатель выполняет различные действия | Код результата ответа, 0 означает успех, не-0 — различные ошибки |
| Language | Язык реализации отправителя | Язык реализации получателя |
| Version | Версия программы отправителя | Версия программы получателя |
| Opaque | Идентификационный код запроса, используется для многопоточности и повторного использования соединения | Получатель не изменяет, а возвращает напрямую |
| Flag | Флаг уровня связи | Флаг уровня связи |
| Remark | Передаваемая произвольная текстовая информация | Подробное описание ошибки |
| ExtFields | Пользовательские поля запроса | Пользовательские поля ответа |
| CustomHeader | Пользовательская структура, преобразуется в данные типа extFields при передаче | Пользовательская структура, преобразуется в данные типа extFields при передаче |
| Body | Тело запроса | Тело ответа |

1.
  1. **4.6 Сжатие сообщений**

После достижения определённой длины сообщения предоставляется функция сжатия сообщений. Алгоритм сжатия использует zip. Второй бит флага SysFlag в RemotingCommand указывает, сжато ли сообщение.

1.
  1. **4.7 Обработка пульса**

Компонент связи сам по себе не обрабатывает пульс, обработка пульса выполняется верхним уровнем.

#### Техническое описание SmartGo-Registry

###

###

Для версии V1.0.0

### ©

#### Архитектура базовой платформы Чэнду

#### 2017/11/21

###

#### Содержание

1 Введение 4

2 Взаимодействие модуля Registry 4

2.1 Брокер 4

2.2 Консоль 4

2.3 Сеть 4

2.4 Реестр 4

3 Профессиональные термины 5

4 Принцип реализации реестра 5

4.1 Регистрация брокера 5

4.2 Сканирование активных брокеров 6

4.3 Реестр и ZooKeeper 7

4.4 Основные данные структуры реестра 8

4.5 Изменения данных в памяти реестра 8

4.6 Обычная тема и упорядоченная тема 9

4.7 Тема и сопоставление брокера с темой 10

1. Введение

Реестр поддерживает множество информации о брокерах и темах, устанавливая длительное соединение через сеть и брокера, чтобы поддерживать связь с брокером, одновременно предоставляя обнаружение пульса, обновление данных и другие обычные услуги. Он сохраняет список активных брокеров, включая мастер и подчиненные; также сохраняет все темы и списки очередей всех тем.

1. Взаимодействие модуля реестра
  1. 2.1 Брокер

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

1.
  1. 2.2 Консоль

Модуль консоли косвенно устанавливает соединение и обменивается данными со всеми реестрами через клиентский интерфейс нижнего уровня.

1.
  1. 2.3 Сеть

Реестр создаёт службу через Net (текущий порт — 9876), регистрирует и публикует службу, которую могут вызывать модули Broker, Client, Web и т. д.

1.
  1. 2.4 Реестр

Между узлами реестра нет взаимодействия. Большая часть данных, поддерживаемых реестром, хранится в памяти, а данные упорядоченной темы хранятся в файле.

1. Профессиональные термины

 Тема

Тема — это сообщение, один онлайн-экземпляр Producer может соответствовать только одной теме, один онлайн-экземпляр Consumer может соответствовать нескольким темам, одно сообщение должно принадлежать одной теме.

 QueueData

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

 BrokerData

Данные, описывающие подробную информацию о брокере, включая имя брокера, адрес брокера, идентификатор брокера и т.д.

 TopicQueueTable

Описывает сопоставление между данными структуры темы, очереди и брокера.

 BrokerAddrTable

Описывает сопоставление между именами брокеров, идентификаторами брокеров и адресами брокеров.

 ClusterAddrTable

Описывает соответствие между брокерами и кластерами, можно запросить подробную информацию о брокерах через имена брокеров.

 BrokerLiveTable

Описывает структуру информации о сердцебиении брокера, после того как каждый брокер отправляет информацию о сердцебиении, реестр будет поддерживать время последнего обновления информации о сердцебиении брокера. Брокер каждые 30 секунд отправляет информацию о сердцебиении. После получения информации о сердцебиении, Registry поддерживает в памяти структуру данных BrokerLiveTable. Эта структура данных хранит информацию о последнем обновлении времени сердцебиения каждого брокера, адресе брокера и сетевом соединении и т. д.

После запуска Registry сразу начинает выполнять периодическую задачу: каждые 10 секунд выполняется сканирование брокеров, которые не отправляли информацию о сердцебиении в течение 2 минут.

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

1. 
  1.1 Registry и ZooKeeper

(1) Для Smartgo данные темы на каждом Master равноправны, ни один из Master не имеет всех данных темы, поэтому функция выбора лидера ZooKeeper не подходит для Smartgo.

(2) В кластере Smartgo необходимо иметь компонент для обработки некоторых общих данных, таких как список брокеров, время обновления брокеров. Обработка некоторых логических отношений между данными с помощью клиента ZooKeeper может быть довольно сложной, и ZooKeeper также должен обеспечивать согласованность между несколькими мастерами, что ещё больше усложняет код. Если есть несколько ролей, код ZooKeeper становится ещё более сложным.

(3) Поскольку в кластере Smartgo не используются некоторые функции ZooKeeper высокого уровня, а только его функции обеспечения согласованности данных и публикации-подписки, лучше написать облегчённую версию Registry, чем полагаться на сложный ZooKeeper.

(4) Registry также можно развернуть в кластерной среде. Между Registry нет никакой синхронизации данных, и Registry с менее чем тысячей строк кода определённо более стабилен, чем ZooKeeper.

1.
  1.2 Основные структуры данных Registry

| Структура данных | Тип | Формат данных | Хранимые данные |
| --- | --- | --- | --- |
| TopicQueueTable | HashMap | topic[list<QueueData>] | Сохраняет информацию о теме-очереди |
| BrokerAddrTable | HashMap | brokerName[BrokerData] | Сохраняет адрес брокера |
| ClusterAddrTable | HashMap | clusterName[set<brokerName>] | Сохраняет связь между брокером и кластером |
| BrokerLiveTable | HashMap | brokerAddr[brokerLiveTable] | Сохраняет данные сердцебиения брокера | По умолчанию алгоритм распределения нагрузки для потребительских пушей заключается в равномерном распределении очередей по ConsumerId.

Например, если есть Topic с 24 очередями и один клиент, который подписывается на этот Topic, то он будет потреблять все 24 очереди. Если же клиентов два, каждый из них будет потреблять по 12 очередей из 24. Таким образом достигается распределение нагрузки между клиентами. Клиенты могут переписать эту стратегию распределения нагрузки.

**4.6 Push-широковещательное потребление**

Процесс похож на push-потребление из пуша, но с сохранением локального offset. Распределение нагрузки не влияет на широковещательное потребление.

**4.7 Pull-потребление**

Временная диаграмма сообщений при pull-потреблении выглядит следующим образом: