---
# System prepended metadata

title: Однократные подписки
tags: [Habr]

---

###### tags: `Habr`

# Однократные подписки

![](https://i.imgur.com/mT9vUcS.png)



При реализации сервисов очередей различных видов часто встаёт вопрос: "А как лучше  реализовать систему уведомлений?".

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

Система распространения уведомлений встречается во многих программных комплексах. Как правило клиентов таких систем немного: десятки, реже - сотни.

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

---

Предположим, что нам надо построить сервис, способный уведомить множество подписчиков. Причём это множество достаточно велико (десятки, сотни тысяч на инстанс).

```graphviz
digraph "Pub-Sub example"{
    nodesep="1.5"

    source[label="source"];
    
    server[label="server"]
    
    source->server[label=" pub"]
    
    client1[label="Client 1"]
    client2[label="Client 2"]
    client3[label="Client 100.000"]
    
    server->client1[label="pub"]
    server->client2[label="pub"]
    server->client3[label="pub"]
    
    client1->server[label="sub"]
    client2->server[label="sub"]
    client3->server[label="sub"]
}

```

Какие проблемы нас ожидают при решении данной задачи?

### Клиентская сторона

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


### Серверная сторона

- Система уведомлений клиента X не должна зависеть от скорости канала на клиента Y.
- Система уведомлений вообще не должна зависеть от скоростей клиентов (обобщение предыдущего пункта).


Если порефлексировать над выписанными проблемами, то можно видеть какие обязательные свойства будет иметь система доставки событий от сервера клиенту, а так же вариативность по остальным свойствам даст нам множество решений.

### Требования к системе

Итак обязательные свойства:

- Система доставки событий должна записывать отправляемые события в буфер (а не сокет). Запись в память даст максимальный перфоманс. Запись в память предотвратит блокировку записывающего.

Вариативные свойства:

- Либо: Клиент должен получать **все** отправляемые ему сообщения вне зависимости от скорости канала.
- Либо: Клиент должен уметь диагностировать разрыв соединения и уметь восстанавливать консистентность своего состояния после сбоя.

### Поиск вариантов

#### Резиновая очередь

Рассмотрим первый из вариантов. В общем случае, с учётом обязательных свойств, это можно реализовать при помощи "резинового" буфера (или очереди неограниченных размеров).

Недостатки этого подхода очевидны:

- При наличии медленных клиентов потребление памяти системой передачи сообщений растёт
- При наличии хотя бы одного клиента, который всегда (на протяжении очень длительного времени) не справляется с передаваемым на него потоком сообщений система гарантированно придёт к тому, что к ней в гости постучится ООМ-киллер.

#### Очередь фиксированного размера

Первое что приходит в голову после того, как к серверу в гости зайдёт ООМ-киллер, это введение ограничения на размер очереди.

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

Пример алгоритма работы сервера может быть приблизительно таким:

```flow
start=>start: Start
wait=>operation: Waiting for push
check=>condition: Overflow?
unsubscribe=>end: Unsubscribe

start->wait(right)->check(left)
check(no, top)->wait
check(yes, right)->unsubscribe
```


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

Клиент реализует примерно такой алгоритм:


```flow
start=>start: Start
subscribe=>operation: Subscribe
gstate=>operation: Get server's state
gmess=>operation: Get next message
ocheck=>condition: Overflow?


start(left)->subscribe(bottom)->gstate(bottom)->gmess(right)->ocheck
ocheck(yes, right)->subscribe
ocheck(no, top)->gmess

```

Данный алгоритм подписки предусматривает устойчивость клиента к повторной обработке события (идемпотентность).

### Реальные приложения

Теперь если мы рассмотрим реальные приложения, то мы видим следующую ситуацию:

> Крайне редко через канал связи pub/sub передаются сами данные. В основном передаётся самый минимум данных.

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

Если это подписка-репликация, то по каналу уведомлений передаются номера последних изменённых сообщений итп.

В некоторых случаях **все** данные передаются по каналу уведомлений, но в этих случаях, как правило, это сделано потому, что самих данных очень немного.
Но даже в случаях, когда все данные передаются по каналу уведомлений, всё равно предусматривается режим восстановления состояния с нуля: Канал может быть разорван, клиент или сервер быть перезапущен итп.

#### Передача стейта

Множество приложений в архитектуре pub/sub по сути дела передают через систему уведомлений не сообщения, а некий стейт. Вернее сказать при помощи сообщений передаётся информация о стейте на сервере.

- Сервер присылает сообщение "на сервере 40 заказов"
- Клиент сверяет со своим "у меня 38 заказов, надо получить ещё два".

Или

- Сервер присылает сообщение "в чат XXX клиент отправил сообщение и в чате стало YYY сообщений"
- Клиент сверяет со своим "мой клиент смотрит в чат XXX?". И если смотрит "сколько у моего клиента сообщений в чате XXX?".


Редко когда архитектуру передачи уведомлений нельзя свести к передаче стейта. Обычно, если это не сделано, то это во первых можно сделать. Во вторых, если это сделать, скорее всего система станет лучше (перформить, отзываться, итп).


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

**Проблема 1**: Проблема батчинга сообщений.

Часто встречается особенно при интеграции различных разнородных систем. Сервер загружает с другого сервера 100500 данных кронскриптом. В систему уведомлений приходит большой пакет событий, который переполняет буфера (которые мы обсуждали выше).

**Проблема 2**: Клиент медленнее сервера

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

Примечание: "медленный клиент" может означать как недостаток вычислительной мощности на клиенте, так и низкую пропускную способность канала на него.

**Проблема 3**: Большой объём памяти, требуемый для буферов между клиентом и сервером (следствие проблем 1 и иногда 2).


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

То есть вариант "Отписка происходит всякий раз когда отправляется сообщение в канал".

Алгоритм сервера получается таким:

```graphviz
digraph "Server push"{
   push->unsubscribe
   
   push[label="Push event"]
   unsubscribe[label="Unsubscribe"]
}
```

А алгоритм клиента таким:

```graphviz
digraph "Client event processing"{
   subscribe->get->subscribe
   
   subscribe[label="Subscribe"]
   get[label="Get state from server"]
}
```

Если протокол между клиентом и сервером асинхронный, то, памятуя о паттерне "передача стейта", можем написать такой алгоритм обработки запроса `subscribe` на сервере:


```graphviz
digraph "Subscribe request" {
   add
   response
   
   add->response
   
   add[label="Add client to\nsubcribed list"]
   response[label="Send current (cached)\nstate to client"]
}
```

А алгоритм обслуживания серверного `push` таким образом:

```graphviz
digraph "Push processing"{
  store->send->unsubscribe

  store[label="Store event to cache"]
  send[label="Send event"]
  unsubscribe[label="Unsubscribe"]
}
```

То есть словами:

- Клиент отправляет запрос "подписаться"
- В ответ на запрос получает текущий стейт
- Когда происходит событие, получает информацию о нём
- Переподписывается

```sequence
participant Client
participant Server

Client-->Server: \nsubscribe
Server-->Client: \nstate

Server->Server: wait event
Server-->Client: \nstate (changed)

note right of Client: Restart

```

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

### Ещё оптимизируем

- Если сервер между переподписками клиента будет хранить статус того, что стейт поменялся пока клиент был отключён, то можно сэкономить дополнительный запрос за стейтом. Но эта оптимизация доступна только в условиях, когда мы можем контроллировать разницу между:

- клиент не переподписывается, потому что обрабатывает стейт
- клиент не переподписывается, потому что отключился

В случае, если между клиентом и сервером установлено постоянное соединение (например TCP), то эта разница легко диагностируется: TCP установлен, значит клиент в порядке.

```sequence
participant Client
participant Server

note left of Client: TCP
Client-->Server: \nfirst subscribe
Server-->Client: \nstate

note right of Server: Maintains pairs:\nclient-state
Client-->Server: \nnext subscribe
Server->Server: check if state changed
Server-->Client: \nstate

note right of Client: Restart
```


Достоинства получившейся схемы доставки уведомлений:

- Схема крайне хорошо переносит "батчинг".
- Схема максимально толерантна к медленным клиентам.

Недостатки:

- Схема оптимизирована именно под передачу стейта. Передача просто потока уведомлений ложится в схему плохо.


### Request-Response?

Схема получилась очень похожая на обычную запрос-ответ пару. Отличие в том, что сервер между запросами "помнит" о подписывавшихся клиентах и на основании этой памяти ответ может быть как выдан сразу, так и задержан.

### Tarantool

Традиционно в Tarantool pub-sub подписки делались следующим образом: клиент выполнял запрос subscribe. Этот запрос "задерживался" до появления события, либо до таймаута.

Классический long-polling. На сервере при этом "ожидала" события хранимая процедура (со всеми связанными с её запуском накладными расходами).

Какие недостатки такого подхода?

- На сервере существует поток, ожидающий событие.
- На клиенте существует поток, ожидающий ответ на запрос.

Подобные недостатки приводили к тому, что одним тарантулом было сложно обслужить более нескольких тысяч клиентов (ограничения по числу файберов, по размеру потребляемой памяти LuaJIT итп).

Начиная с версии 2.10 протокол и ядро тарантула будет поддерживать систему однократных подписок, описанную выше. Ни серверу ни клиенту больше не требуется "содержать" поток, обслуживающий систему доставки событий. С помощью этой технологии теперь можно писать приложения, позволяющие одному Тарантулу обслуживать десятки тысяч присоединённых клиентов.


Между тарантулами этот механизм будет выглядеть примерно так:

```lua=
-- Клиент
net_box.watch(
    key,
    function(key, state)
       -- Стейт, определяемый ключом `key`
       -- поменялся на новое значение 
        ...
        
       -- Переподписка произойдёт после завершения
       -- этой функции
    end
)

-- Сервер

-- уведомляем подписчиков о том что стейт, определяемый
-- ключом `key` теперь имеет новое значение - `new_state`
box.broadcast(key, new_state)
```

Старые решения pub/sub на Тарантуле продолжат работать без изменений. Новые могут использовать этот механизм для того, чтобы поддерживать огромные количества клиентов или потреблять меньшее количество ресурсов.

Помимо того, что пользователи смогут строить свои приложения, используя этот механизм, Tarantool будет распространять через него информацию о таких событиях как:

- смена мастера (лидера) в кластере
- предстоящее выключение узла
- некоторые статусы сервера

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


#### Применения

##### Пример 1

Если рассмотреть типовой кластер Тарантул с шардингом, то схема кластера будет выглядеть примерно так:


```graphviz
digraph "Шардированный кластер Tarantool" {

client -> router
router -> storage1
router -> storage2
router -> storageX
}
```

Каждый сторадж - это маленький кластер (репликасет) из Тарантулов. Роутер удерживает соединения с каждым стораджем (с каждым узлом стораджа). Посредством этого соединения выполняет следующую работу:

- осуществляет дискаверинг лидеров репликасетов
- мониторит работоспособность узлов
- перенаправляет пользовательские запросы к стораджам

Если стораджей в кластере скажем 50, а каждый сторадж - это репликасет из трёх нод. То только для дискаверинга лидеров роутер вынужден "держать на балансе" 150 потоков (файберов). Это значительные накладные ресурсы как на память, так и на CPU.

Начиная с версии 2.10 требования роутеров к ресурсам значительно уменьшатся, благодаря внедрению данного механизма.

##### Ещё пример

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

Система использовалась для того, чтобы уведомлять примерно 100 тыс исполнителей (преимущественно это водители такси) о появляющихся заказах.

Об этой системе [даже была написана статья](http://unera.net/all/2017/11/02/event-lp.html).

В то время для масштабирования мы выделяли одно ядро CPU на 3-5 тыс водителей. Система уведомления 100 тыс исполнителей утилизировала два 16-ядерных сервера.

Основным ограничителем была память LuaJIT.

С использованием описанного механизма можно редуцировать описанный сервис до всего одного ядра.

### Заключение

Tarantool - иногда называют базой данных. Иногда App-сервером. Но Tarantool - это нечто большее. Tarantool - это конструктор, используя который, Вы можете построить что-то мощное и серьёзное и при этом не требовательное к ресурсам.

В этой статье описан один из новых "кубиков" этого конструктора. Используя этот кубик можно сделать недорогую систему уведомлений огромного числа подписчиков.

Надеемся процесс применения кубиков доставит разработчикам удовольствие.
