Глава 144 Проект 5: сервис WebSocket-пуша реального времени

Ориентация главы

Пятый проект тома 13: сервис пуша реального времени (pushd) — шлюз долгоживущих соединений WebSocket: клиент подключается и подписывается на канал ({"op":"sub","ch":"metrics"}), сервер доставляет сообщения, порождаемые бэкендом, всем членам канала (широковещание); поддержаны выселение (kick), запрос состава канала и graceful-остановка (полный цикл «сперва Seal, потом расчистка»). Это комплексное применение четырёх глав xws (главы 110-113): конечная точка маршрутизации (глава 113, xrtWsServerRoute — безопасный вход с политикой Origin и обратным вызовом авторизации), группа соединений (глава 112, xwsgroup — управление членами / широковещание по снимку / Seal-полузакрытие), путь отправки (глава 111, writer/ссылки/сжатие — примитивы многоадресной рассылки), жизненный цикл соединения (глава 110, pause/resume/протокол закрытия). Новое системное ограничение — долгоживая конкурентность: жизненный цикл сотен одновременно живых соединений, веерное усиление рассылки, влияние медленного клиента на целое — опыт Проектов 1, 2 и 4 плюс это ограничение сходится в полный стек в статическом сервере (глава 145).

144.1引入

Сцена требования: панель мониторинга должна показывать метрики сервиса в реальном времени — опрос (раз в секунду) даёт высокую задержку и тратит впустую; SSE (глава 107) одностороннего хватает, но нужны семантика подписки на каналы со стороны клиента и двустороннее управление (sub/unsub/kick) — двустороннее долгоживущее соединение WebSocket — естественная форма. Дизайн интерфейса pushd: протокол соединения (апгрейд на /ws, подпротокол push.v1); сообщения подписки (JSON: sub/unsub имя канала); сообщения пуша (JSON: канал + нагрузка + номер — клиент по номеру ловит пропуски); управляющие действия (kick указанного соединения, список членов канала — интерфейс эксплуатации). Критерии приёмки: (1) 100 конкурентных соединений, три канала, доставка всех широковещаний (без пропусков — сверка номеров); (2) медленный клиент (принимает, но не читает) не тормозит быстрых (проверка изоляции); (3) kick действует немедленно (последующие сообщения не достигают выселенного); (4) graceful-остановка: после Seal новые соединения отвергаются, сообщения в полёте доставляются, выход только после закрытия всех соединений (сверка конечных счётчиков); (5) память рассылки отвязана от числа соединений × размера сообщения (отправка по ссылке — проверка stats); (6) под противодавлением сервис стабилен (рассылка не блокируется из-за одного переполненного соединения).

144.2概念

Архитектура: сборка трёх слоёв

Решение о хранении множества подписок: группа соединений ведает только «множеством соединений» (граница главы 112 — не про кодирование сообщений и не про бизнес); отображение канал → множество соединений — бизнес-состояние: независимая структура многих групп (по xwsgroup на канал) или Map<канал, группа>. Решение: по группе на канал — конкурентная семантика группы (уникальный член / ссылки / снимки) — ровно то, что нужно управлению членами канала, достаётся даром; альтернатива Map<канал, массив соединений> заставляет самому реализовать всю конкурентную семантику — отвергнуто.

Выбор модулей и обоснования

Конвейер рассылки: от сообщения бэкенда к N клиентам

Роль номера: на канал монотонно растёт — клиент детектирует пропуски (получил 3 и 5 — 4 потеряна: активный re-sync). Групповой апгрейд отправки по ссылке (глава 111): нагрузка одна, очереди N соединений держат по ссылке — обратный вызов освобождения считает (настоящее освобождение после последнего дописавшего); внутри GroupTextAsync ровно эта форма. Три уровня политики медленного соединения: пропуск + учёт (по умолчанию — потеря кадра приемлема, клиент делает re-sync); выселение (настраиваемый порог — чистка хронически деградирующих соединений); безграничное ожидание (не предоставляется — одно соединение, убивающее всех, нарушает критерии (2)(6)).

Протокол подписки и обработка сообщений

Получено сообщение (событие главы 110) → JSON-разбор (глава 32) → диспетчеризация по полю op: sub (проверить законность имени канала → вступить в соответствующую группу GroupAdd — первому подписчику создать группу); unsub (GroupRemove + возврат ссылки); сердцебиение (ping клиента — авто-Pong слоя соединения уже обработал (глава 110), прикладной слой может опционально ответить статусом присутствия). Незаконные сообщения: неудача разбора — кадр ошибки (код JSON) без разрыва соединения (протокольная снисходительность — испорченный кадр не значит испорченное соединение); незаконное имя канала — кадр ошибки — каждая ошибка локализуема (прикладная версия модели ошибок главы 4).

Полный цикл graceful-остановки

Полное развёртывание шаблона остановки главы 112 в pushd: шаг первый прекратить приём (Drain слоя прослушивания — новые HTTP-апгрейды не принимаются); шаг второй Sealing всех групп каналов (GroupSeal — идемпотентно; новые подписки не входят, наличные члены продолжают получать); шаг третий финальное широковещание (кадр "server shutting down" — клиенту явный сигнал, а не молчаливый разрыв); шаг четвёртый закрытие по соединению (ConnClose с кодом NORMAL — протокольное прощание); шаг пятый ожидание обнуления счёта Close-обратных вызовов (все соединения в конечном состоянии); шаг шестой уничтожение групп и маршрута (жизненный цикл Data у Release закрывается здесь — глава 113 «Router и все соединения вышли — ровно один раз»); шаг седьмой остановка Engine. Точки сверки критерия (4): кадр закрытия доставлен (клиент получил shutting down) + ноль утечек (счёт членов групп обнулился) + код выхода 0.

Сопоставление с опытом предыдущих проектов

Двенадцать уроков (инвентаризация главы 143) в форме долгоживущих соединений — повтор и пополнение: приёмка прежде (шесть пунктов — доля конкурентно-временных записей растёт); форма решает контейнер (канал = группа, множество подписок = Map, сообщение = JSON — три формы, три контейнера); представления живут одним кадром (представление события сообщения — разобрал и диспетчеризовал — без буферизации); слоистая заменяемость (сменить транспорт на TLS / сериализацию на XLON — один слой); факты платформы входят в проект (политика Origin и согласование подпротоколов — «факты платформы» WS); бюджет прежде (веер рассылки = скорость сообщений × число соединений — основание бюджета противодавления и политики медленных); тестирование в четырёх квадрантах (конкурентное тестирование становится главным полем: 100 соединений × три канала × сверка номеров). Две новые: веер — усилитель (1 сообщение бэкенда × 1000 соединений = 1000 отправок — каждая наносекунда пути рассылки усиливается, отправка по ссылке не оптимизация, а необходимость); очистка состояния долгоживущих соединений — финальный экзамен (очистка коротких соединений происходит естественно после каждого запроса; очистка долгоживущих — только при остановке и выселении — утечки коварнее, сверка конечных счётчиков — единственная надёжная приёмка).

144.3示例

Первая полная программа: управление членами группы каналов

Программа ниже из examples/websocket/group — шаблон подключения управления членами канала:

#include <stdio.h>

#include <xws.h>



/* Upgrade 成功后把借用连接加入应用连接组。 */
static void connectionOpen(xwsgroup* pGroup, xwsconn* pConnection)
{
	(void)xrtWsGroupAdd(pGroup, pConnection);
}



/* 唯一 Close 回调中移除连接，归还连接组持有的引用。 */
static void connectionClose(xwsgroup* pGroup, xwsconn* pConnection)
{
	(void)xrtWsGroupRemove(pGroup, pConnection);
}



/* 示例保留回调形状，实际服务器可直接把组放入路由上下文。 */
int main(void)
{
	xwsgroup* pGroup = xrtWsGroupCreate(0);

	if ( pGroup == NULL ) {
		return 1;
	}
	(void)connectionOpen;
	(void)connectionClose;
	printf("WebSocket connection group is ready\n");
	xrtWsGroupDestroy(pGroup);
	return 0;
}

Что только что произошло. (1) Две функции connectionOpen/Close — ядро подписки/отписки pushd: апгрейд успешен — GroupAdd (в канал), единственный Close-обратный вызов — GroupRemove (возврат ссылки) — групповой жизненный цикл соединения строго синхронен жизненному циклу соединения (утечка отписки невозможна — Close есть единственная гарантия конечного состояния «ровно один раз»). (2) pushd добавляет сверху маршрутизацию каналов (много групп) и протокол сообщений — конкурентная семантика управления членами (идемпотентный повторный Add / отказ по ёмкости / стабильный снимок) целиком наследуется от группы. (3) Настоящий сервер кладёт группы в Data маршрута (глава 113) — образец фиксирует «форму подключения».

Вторая полная программа: асинхронный объект операции рассылки

Вторая программа из examples/websocket/group_future — ожидаемое широковещание:

#include <stdio.h>

#include <xws.h>



/* 展示空组广播也会返回可等待、可检查的完整操作对象。 */
int main(void)
{
	xwsgroup* pGroup = xrtWsGroupCreate(0);
	xwsgroupop* pOperation;

	if ( pGroup == NULL ) {
		return 1;
	}
	pOperation = xrtWsGroupTextAsync(
		pGroup,
		XRT_STR_LITERAL("service-ready")
	);
	if ( (pOperation == NULL) ||
		(xrtWsGroupOpWait(pOperation) != XWAIT_OK) ) {
		xrtWsGroupOpDestroy(pOperation);
		xrtWsGroupDestroy(pGroup);
		return 1;
	}
	printf(
		"members=%zu accepted=%zu rejected=%zu\n",
		xrtWsGroupOpCount(pOperation),
		xrtWsGroupOpAccepted(pOperation),
		xrtWsGroupOpRejected(pOperation)
	);
	xrtWsGroupOpDestroy(pOperation);
	xrtWsGroupDestroy(pGroup);
	return 0;
}

Что только что произошло. (1) xrtWsGroupTextAsync возвращает объект операции — три шага: запустить / подождать / освободить: форма вызова рассылки pushd (производитель отправил и не ждёт — темп задаёт таймер; управляющий интерфейс ждёт — после kick подтвердить доставку). (2) Пустая группа тоже полная операция (ноль соединений = ноль успехов — агрегатная семантика самосогласованна). (3) Точка политики медленного соединения здесь: объект операции не ждёт отдельное медленное соединение (завершение = приём всеми — AGAIN медленного обрабатывается противодавлением его собственного соединения) — «каждое соединение принимает независимо» и есть механизм критериев (2)(6).

Разбор реализации: исходник ядра рассылки и подписки

Два ядровых цикла pushd — обработка подписки и рассылка по каналу — форма исходника (разбор ключевых сегментов над учебными образцами):

/* событие сообщения: диспетчеризация JSON-протокола (sub/unsub) */
static void on_message(xwsconn* Conn, const uint8_t* Data,
	size_t Size, ptr Ctx)
{
	pushd* S = Ctx;
	xvalue Msg;
	xstrview Op, Ch;

	/* неудача разбора: кадр ошибки в ответ, соединение живёт (снисходительный контракт) */
	xvalue* V;
	if ( !xrtJsonParse(Data, Size, &Msg) ||
		((V = xrtValueObjectGet(Msg,
			XRT_STR_LITERAL("op"))) == NULL) ||
		!xrtValueGetString(V, &Op) ) {
		send_error(Conn, "bad message");
		return;
	}
	if ( ((V = xrtValueObjectGet(Msg,
			XRT_STR_LITERAL("ch"))) == NULL) ||
		!xrtValueGetString(V, &Ch) ||
		!channel_name_valid(Ch) ) {
		send_error(Conn, "bad channel");
		return;
	}

	if ( xrtStrEqual(Op, XRT_STR_LITERAL("sub")) ) {
		xwsgroup* Gr = pushd_channel(S, Ch);  /* взять или создать */
		xrtWsGroupAdd(Gr, Conn);               /* идемпотентно */
		send_ok(Conn, "subscribed");
	} else if ( xrtStrEqual(Op, XRT_STR_LITERAL("unsub")) ) {
		xwsgroup* Gr = pushd_find(S, Ch);
		if ( Gr != NULL ) { xrtWsGroupRemove(Gr, Conn); }
		send_ok(Conn, "unsubscribed");
	}
}

/* 频道广播：快照 + 引用发送 + 慢连接跳过 */
static void pushd_broadcast(pushd* S, xstrview Ch, str Payload)
{
	xwsgroupsnapshot* Snap;
	size_t i;

	xwsgroup* Gr = pushd_find(S, Ch);
	if ( (Gr == NULL) || xrtWsGroupSealed(Gr) ) { return; }
	Snap = xrtWsGroupSnapshotCreate(Gr);   /* 稳定快照 */
	for ( i = 0; i < xrtWsGroupSnapshotCount(Snap); i++ ) {
		xwsconn* C = xrtWsGroupSnapshotGet(Snap, i);
		xwsgroupop* Op = xrtWsGroupTextAsync(Gr, view(Payload));
		if ( Op == NULL ) { S->SlowCount++; }  /* 跳过+记慢 */
		else { xrtWsGroupOpDestroy(Op); }
	}
	xrtWsGroupSnapshotDestroy(Snap);       /* 归还成员引用 */
}

Пункты разбора. (1) Порядок диспетчеризации on_message: сперва разбор целого кадра (испорченный кадр — ошибка без разрыва), затем проверка полей, потом исполнение — каждый уровень ошибок независимо локализуем (bad message / bad channel / квитанция успеха). (2) pushd_channel берёт или создаёт группу — первая подписка создаёт группу, GroupAdd идемпотентен (повторный sub на тот же канал безопасен — исполнение семантики уникального члена главы 112). (3) Цикл снимка pushd_broadcast — позитивная форма ловушек 1/2: стабильный обход по снимку, групповая асинхронная операция, отказ — пропуск со счётом — три контракта за один раз. (4) Проверка GroupSealed на пути остановки: после Seal рассылка естественно молчит (финальное широковещание отправляется до Seal — порядок и есть семантика).

Матрица приёмки и дизайн конкурентных тестов

Тестовая матрица шести критериев (версия четырёх квадрантов главы 133 для долгоживущих соединений): позитив — 100 соединений, три канала × 1000 кадров, сверка номеров (критерий (1)); гонки — рассылка конкурентно с подпиской/отпиской (практическая проверка семантики снимков), рассылка конкурентно с kick (после Remove не доставляется — критерий (3)); медленный клиент — 1 медленный (принимает не читая) + 99 быстрых (критерий (2): номера быстрых не затронуты); противодавление — инъекция производительности больше потребления (критерий (6): сервис стабилен, счёт медленных растёт); остановка — 50 активных соединений в момент остановки (критерий (4): кадры shutting down доставлены всем, счётчики обнулены, код выхода 0); ресурсы — OOM по точкам (отказ выделения на пути рассылки — кадр пропускается без падения). Сторона клиента тестов тоже конкурентная — исполнитель главы 59 ведёт 100 виртуальных клиентов (подключение / подписка / приём / сверка) — тестовый код сам конкурентен, его корректность держится на сверке номеров (не на предположениях о времени).

Сопоставление с картой знаний книги

Кирпичи, на которые наступил pushd: глава 4 (модель ошибок — кадр ошибки на уровне сообщений), глава 18 (Map — маршрутизация имени канала к группе), глава 25 (строки — проверка имён каналов и сравнение op), глава 32 (JSON — кодек протокола подписки), главы 55-59 (конкурентность / исполнитель — каркас тестовых клиентов), глава 64 (порт событий — низ Engine), глава 67 (TCP — транспортная база соединений), глава 94 (HTTP-апгрейд — слой рукопожатия WS), главы 95-97 (ядро протокола WebSocket — кадры / поток / закрытие), главы 104-105 (HTTP-сервер — вход Router и промежуточного ПО), главы 110-113 (четыре главы xws — прямой инструментарий проекта). Шестнадцать глав каждая на своём месте в pushd — Проект 5 на сегодня самый широкий по охвату (финальная глава статического сервера будет шире). Смысл проверки карты знаний: ни одна глава не изолирована — каждая была скомбинирована в какой-то реальной форме.

144.4契约

• Безопасность точки входа: фиксированная точка xrtWsServerRoute — Origin в три уровня по умолчанию (SAME_HOST_OR_ABSENT) + Authorize с опциональной проверкой токена; Open занимает соединение на время, долгое хранение — только после Ref (контракт главы 113).
• Канал есть группа: по xwsgroup на канал — семантика членов (идемпотентный Add / отказ по ёмкости / стабильный снимок) даётся группой; множество подписок отделено от управления членами (бизнес — прикладному слою).
• Конвейер рассылки: снимок (стабильный обход + добавление ссылок) → отправка по ссылке каждому → уничтожение снимка с возвратом ссылок; одна память нагрузки на N соединений (память отвязана от веера).
• Три уровня медленного клиента: пропуск + учёт (по умолчанию) / выселение (порог) / безграничное ожидание не предоставляется — одно соединение не тащит всех.
• Протокол номеров: на канал монотонно растёт — клиент по номеру ловит пропуски и сам делает re-sync (эксплуатационный дивиденд семантики «перезапуск без потерь»).
• Снисходительность сообщений: неудача разбора одного кадра — кадр ошибки без разрыва; каждая ошибка локализуема.
• Семь шагов остановки: прекратить приём → Seal (идемпотентно) → финальное широковещание → закрытие по соединению → обнуление счётчиков → уничтожение групп и маршрута → остановка Engine — сверка конечных состояний.
• Жизненный цикл Data: Release маршрута ровно один раз после выхода Router и всех соединений (контракт главы 113).
• Семантика выселения: сперва GroupRemove из группы, затем ConnClose — рассылка после выхода из группы не достигает выселенного (механизм критерия (3)).

Форма поставки и интерфейс эксплуатации

Поставка и эксплуатационный дизайн pushd: форма развёртывания — отдельный процесс (ссылка на один заголовок или статическую библиотеку — выбор главы 132); адрес прослушивания / ёмкость каналов / порог медленности читаются из командной строки или окружения (глава 43) — плоскость конфигурации инструмента минимальна (три параметра достаточно для развёртывания; сложная конфигурация — территория configd). Интерфейс эксплуатации — управляющий эндпойнт (HTTP: список каналов / число членов / счёт медленных / выселение) на другом пути того же Router (маршрутизация HTTP и WS-точка сосуществуют — глава 105; форма «один Router, два вида точек» главы 113); вывод JSON (глава 32) для скриптов. Наблюдаемость — счётчики соединений / каналов / скорость рассылки / счёт медленных периодически печатаются (logger главы 38 или прямо stdout JSON Lines — форма file_json главы 139); stats (глава 6) следит за уровнем памяти. Graceful-перезапуск — семь шагов остановки + надзор в духе systemd: в окно перезапуска клиент переподключается (глава 110: переподключение — ответственность клиента) + re-sync по номерам доставляет потерянные кадры — перезапуск без потери семантики — эксплуатационный дивиденд протокола номеров.

Ревизия границы с SSE-пушем

После завершения pushd стоит перепроверить выбор главы 107: когда SSE достаточно, а когда необходим WebSocket? Из трёх требований pushd подписка на каналы (клиент говорит серверу, что ему нужно) и управление kick (двустороннее управление) SSE не может — SSE только сервер толкает, у клиента нет восходящего канала (подписка через параметры URL — много каналов нереально); пуш по номерам сам по себе SSE тоже умеет (у SSE есть поле id и возобновление по Last-Event-ID — механизм главы 107). Значит граница уточняется: чисто нисходящий пуш — SSE (проще — порт 80, автопереподключение встроено в браузер); нужен восходящий контроль клиента (подписка/квитанции/своё сердцебиение) — WebSocket. Семантика подписки ставит pushd на сторону WS — но если требование вырождается в «всем один и тот же поток» (как объявление на весь сайт), SSE — более лёгкий ответ. Ревизия выбора (после реализации решения оглянуться на альтернативы) — часть способности проектирования — она делает следующий выбор быстрее и точнее.

Методика нагрузочного тестирования и планирование ёмкости

Метод главы 136 на службе с долгоживущими соединениями: вопрос нагрузки («при 1000 соединений × 3 канала × 100 кадров/с рассылки — каковы P99 задержки кадров и уровень памяти?» — сперва вопрос, потом стенд); модель нагрузки — виртуальные клиенты (исполнитель ведёт: подключение / подписка / приём / сверка — четыре фазы) + производитель (таймер на заданной скорости) — оба конца детерминированные скрипты (без настоящей сети, локальная петля — нагрузочная версия идеи mock главы 104); метрики — распределение задержки кадров (гистограмма задержки от отправки до приёма — метки времени главы 42), уровень памяти (периодическая выборка stats), траектория счёта медленных; скан — градиент числа соединений (100/500/1000) × градиент скорости (10/100/1000 кадров в секунду) — точка перегиба (какое измерение насыщается первым: память? веерный CPU? Worker Engine?) указывает направление масштабирования. Продукт планирования ёмкости — рекомендуемая полоса нагрузки одного инстанса (например «≤800 соединений и ≤200 кадров/с дают P99 < 50 мс») + действия при выходе (добавить инстансы / снизить кадровую скорость / шардировать каналы).

Ретроспектива: новые уроки Проекта 5

Опыт, добавленный pushd сверх двенадцати: дизайн веера прежде оптимизации (отправка по ссылке / снимки / пропуск на пути рассылки — проектные решения, не патчи производительности — задаются на этапе архитектуры); сверка конечных состояний долгоживущих (проверка очистки трёх классов состояния — соединения / группы / множества подписок — возможна только обнулением счётчиков — «ровно один раз» глав 110-112 на уровне проекта сходится в одну ведомость); номера — дешёвый самовосстанавливающийся протокол (один монотонный целый покупает детекцию пропусков + re-sync — на два порядка дешевле механизма квитанций и повторов; достаточный протокол — хороший протокол); тестовый клиент — тоже испытуемая система (корректность конкурентного тестового каркаса не может держаться на предположениях о времени — сверка номеров применима и к самому тестовому коду); интерфейс эксплуатации — часть продукта (список каналов / выселение / счёт медленных — не «добавим потом», а часть определения поставки — как пятисегментные тайм-ауты главы 104: эксплуатируемость решается на этапе проектирования). Пять уроков, суммарно семнадцать — последний проект (финальный статический сервер) проверит их устойчивость под комбинацией полного стека.

Типичные ошибки

n = xrtWsGroupCount(pGroup);
for ( i = 0; i < n; i++ ) { 发(第 i 个成员); }   /* добавления/удаления в процессе: хаос */

xwsgroupsnapshot* Snap = xrtWsGroupSnapshotCreate(pGroup);
for ( i = 0; i < xrtWsGroupSnapshotCount(Snap); i++ ) {
	发送(xrtWsGroupSnapshotGet(Snap, i));
}
xrtWsGroupSnapshotDestroy(Snap);

for ( each member ) {
	while ( send(conn, msg) == AGAIN ) { wait(); }  /* медленный тащит всех */
}

/* объект операции группы: возврат после приёма всеми; AGAIN медленного принадлежит противодавлению его соединения */
xwsgroupop* Op = xrtWsGroupTextAsync(pGroup, msg);
xrtWsGroupOpDestroy(Op);   /* запустил и пошёл (или ждём агрегат по надобности) */

stop_listener();
xrtWsGroupDestroy(pGroup);   /* в полёте апгрейд/соединения не закрыты: висячие ссылки и зависание */

stop_listener();
xrtWsGroupSeal(pGroup);          /* 1. держать новых */
broadcast("shutting down");       /* 2. сигнал */
close_all_members(pGroup);       /* 3. протокольное закрытие */
wait_close_count_zero();         /* 4. счёт конечных состояний */
xrtWsGroupDestroy(pGroup);       /* 5. уничтожение */
xrtNetEngineDestroy(pEngine);    /* 6. Engine останавливается последним */

Практические задания

Итоги