Realtime և WebSockets

01Polling, long-polling, SSE և WebSockets

Polling-ը մի քանի վայրկյանը մեկ հարցնելն է՝ «նոր բան կա՞»։ Այն պարզ է և աշխատում է ամենուր, բայց իզուր հարցումներ է վատնում և ուշացում ավելացնում։ Long-polling-ը յուրաքանչյուր հարցումը բաց է պահում, մինչև սերվերը նորություն ունենա կամ timeout-ը լրանա, հետո կլիենտը նորից է միանում. ավելի քիչ վատնում, ավելի շատ բարդություն։ Server-Sent Events-ը (SSE) բաց է պահում մեկ HTTP պատասխան, և սերվերը դրա միջով հոսքով տեքստային իրադարձություններ է ուղարկում. այն միակողմանի է (սերվերից կլիենտ), ինքնաշխատ վերամիանում է և լավ է անցնում proxy-ների ու HTTP/2-ի միջով։ WebSockets-ը մեկ TCP միացման վրա բացում է լիարժեք երկկողմանի ալիք, որը իդեալական է, երբ կլիենտն էլ է հաճախ հաղորդագրություններ ուղարկում, ինչպես չատում, համատեղ խմբագրման կամ խաղերի դեպքում։ Գործնական կանոն՝ polling՝ հազվադեպ թարմացումների համար, SSE՝ սերվերից կլիենտ հոսքերի համար, օրինակ՝ ծանուցումներ կամ AI token-ների streaming, WebSockets՝ ինտերակտիվ երկկողմանի տրաֆիկի համար։

02Ինչպես է աշխատում WebSocket միացումը

WebSocket-ը սկսվում է որպես սովորական HTTP հարցում, որը խնդրում է արձանագրությունը upgrade անել։ Եթե սերվերը համաձայն է, պատասխանում է 101 Switching Protocols, և դրանից հետո նույն TCP միացումը HTTP-ի փոխարեն WebSocket frame-ներ է կրում։ Frame-ները փոքր փաթեթներ են՝ տեքստային կամ բինար տվյալների frame-ներ, գումարած կառավարման frame-ներ ping-ի, pong-ի և close-ի համար։ Կլիենտից սերվեր գնացող frame-ները mask են արվում։ Ping-ը և pong-ը միացումը կենդանի են պահում և հայտնաբերում «մեռած» կողմերին։ Քանի որ upgrade-ը HTTP է, միացումն անցնում է load balancer-ների և reverse proxy-ների միջով, բայց դրանք պետք է կարգավորված լինեն այնպես, որ թույլ տան upgrade-ը և երկարատև միացումները։ Production-ում միշտ օգտագործեք wss:// (TLS)։

1GET /ws HTTP/1.1
2Host: api.example.com
3Upgrade: websocket
4Connection: Upgrade
5Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
6Sec-WebSocket-Version: 13
7 
8HTTP/1.1 101 Switching Protocols
9Upgrade: websocket
10Connection: Upgrade
11Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=

03socket.io. namespace-ներ, room-ներ և auth

socket.io-ն WebSockets-ի վրա կառուցված գրադարան է, որն ավելացնում է ավտոմատ վերամիացում, acknowledgement-ներ, long-polling-ի fallback և միացումները խմբավորելու պարզ մոդել։ Namespace-ը (օրինակ՝ /staff կամ /customers) առանձին ալիք է՝ իր սեփական middleware-ով, որը հարմար է տարբեր լսարանների համար։ Room-ը խումբ է namespace-ի ներսում. սերվերը socket-ները ավելացնում է room-ների մեջ և emit է անում ամբողջ room-ին միանգամից։ Տիպիկ room-ներ՝ մեկը յուրաքանչյուր tenant-ի, մասնաճյուղի, տերմինալի և օգտատիրոջ համար։ Նույնականացումն արեք handshake middleware-ում՝ մինչև միացումն ընդունվելը, և room-ի անդամակցությունը որոշեք սերվերում՝ ստուգված ինքնությունից ելնելով։ Երբեք թույլ մի տվեք կլիենտին միանալ room-ի միայն այն պատճառով, որ ինքը խնդրել է։

1io.of('/staff').use(async (socket, next) => {
2  try {
3    socket.data.user = await verifyAccessToken(socket.handshake.auth.token);
4    next();
5  } catch {
6    next(new Error('unauthorized'));
7  }
8});
9 
10io.of('/staff').on('connection', (socket) => {
11  const { companyId, branchId, sub } = socket.data.user;
12  socket.join(['company:' + companyId, 'branch:' + branchId, 'user:' + sub]);
13});
14 
15// elsewhere, after an order changes:
16io.of('/staff').to('branch:' + order.branchId).emit('order.updated', { id: order.id, status: order.status, version: order.version });

04Մասշտաբավորում մի քանի սերվերի վրա Redis-ով

Մեկ պրոցեսը գիտի միայն իրեն միացած socket-ները։ Երբ աշխատեցնում եք API-ի չորս instance, A instance-ին միացած օգտատերը չի տեսնի B instance-ում emit արված իրադարձությունը։ Լուծումը ընդհանուր message bus-ն է։ socket.io-ի Redis adapter-ը room-ի յուրաքանչյուր emit հրապարակում է Redis pub/sub-ում. ամեն instance բաժանորդագրվում է և հասցնում իր լոկալ socket-ներին։ Եթե load balancer-ի հետևում օգտագործում եք socket.io-ի long-polling fallback-ը, միացրեք sticky session-ներ, որպեսզի կլիենտի հարցումները հասնեն նույն instance-ին. միայն WebSocket transport-ի դեպքում դա պետք չէ։ Շատ մեծ fan-out-ի կամ երաշխավորված առաքման համար տարբերակներ են Redis Streams-ը կամ sharded adapter-ը։ Background worker-ներից նույնպես emit արեք նույն adapter-ի միջով (socket.io emitter փաթեթով), որպեսզի queue-ի job-ը կարողանա ուղղակիորեն ծանուցել օգտատերերին։

1import { createClient } from 'redis';
2import { createAdapter } from '@socket.io/redis-adapter';
3 
4const pub = createClient({ url: process.env.REDIS_URL });
5const sub = pub.duplicate();
6await Promise.all([pub.connect(), sub.connect()]);
7 
8io.adapter(createAdapter(pub, sub));

05Laravel. broadcasting Pusher-ով, Reverb-ով և Echo-ով

Laravel-ում broadcasting-ը ներկառուցված է։ Իրադարձությունը նշում եք ShouldBroadcast-ով, սահմանում, թե որ ալիքով է այն գնում, և Laravel-ն այն ուղարկում է driver-ի միջոցով։ Pusher-ը hosted ծառայություն է. սերվերներ պահել պետք չէ, վճարում եք ըստ միացումների և հաղորդագրությունների։ Reverb-ը Laravel-ի սեփական first-party WebSocket սերվերն է, որը խոսում է Pusher-ի արձանագրությամբ, այնպես որ կարող եք self-host անել նույն կլիենտային կոդով։ php artisan install:broadcasting-ը կարգավորում է ամեն ինչ։ Frontend-ում Laravel Echo-ն բաժանորդագրվում է ալիքներին։ Private և presence ալիքների ավտորիզացիան արվում է routes/channels.php-ում. Echo-ն կանչում է /broadcasting/auth endpoint-ը, և ձեր callback-ը որոշում է, թե արդյոք այս օգտատերը կարող է լսել։ Broadcast-ները դրեք queue-ի մեջ, որպեսզի դանդաղ socket սերվերը երբեք չդանդաղեցնի HTTP պատասխանը։

1// routes/channels.php
2Broadcast::channel('orders.{orderId}', function (User $user, int $orderId) {
3    return Order::whereKey($orderId)->where('customer_id', $user->id)->exists();
4});
5 
6// Frontend (Echo)
7echo.private('orders.' + orderId)
8    .listen('OrderStatusChanged', (e) => updateStatus(e.status));

06Վերամիացում, հերթականություն և բաց թողնված հաղորդագրություններ

Մոբայլ ցանցերը անընդհատ կտրվում են, ուստի նախագծեք՝ հաշվի առնելով անջատումները։ Կլիենտները վերամիանում են exponential backoff-ով և jitter-ով, որպեսզի սերվերի վերագործարկումը չառաջացնի հազարավոր միաժամանակյա վերամիացումների «ջարդ»։ Սովորական pub/sub-ում այն իրադարձությունները, որոնք ուղարկվել են, մինչ կլիենտը offline էր, կորչում են։ Socket-ին վերաբերվեք որպես «ինչ-որ բան փոխվեց» ազդանշանի, իսկ REST API-ին՝ որպես ճշմարտության աղբյուրի. վերամիանալիս նորից բեռնեք ընթացիկ վիճակը։ Յուրաքանչյուր իրադարձության մեջ ներառեք version համար կամ timestamp, որպեսզի կլիենտը կարողանա անտեսել հնացած կամ խառը հերթականությամբ եկած թարմացումները։ Եթե մեկ entity-ի համար հերթականությունը կարևոր է, նրա իրադարձություններն ուղղեք մեկ ալիքով և մշակեք հաջորդաբար։ Կրիտիկական տվյալների համար օգտագործեք acknowledgement-ներ կամ պահպանված event log՝ «since» cursor-ով։

• Payload-ները փոքր պահեք. id-ներ և կարգավիճակ, ոչ թե ամբողջական օբյեկտներ։
• Ուղարկեք heartbeat-ներ. փակեք և մաքրեք «մեռած» միացումները։
• Proxy-ի timeout-ները (nginx proxy_read_timeout) կարգավորեք ձեր ping interval-ից մեծ։

07Թողունակություն՝ file descriptor-ներ, proxy-ներ և heartbeat-ներ

Յուրաքանչյուր բաց WebSocket բաց file descriptor է socket սերվերի և դրա առջև կանգնած յուրաքանչյուր proxy-ի վրա։ Linux-ը սահմանափակում է դրանք ըստ պրոցեսի, հաճախ լռելյայն 1,024-ով, ուստի scaling-ի առաջին պատը սովորաբար «too many open files»-ն է, ոչ թե CPU-ն։ Բարձրացրեք սահմանը ծառայության համար (ulimit -n, systemd-ում LimitNOFILE, supervisor-ում minfds), իսկ nginx-ում բարձրացրեք worker_connections-ը և worker_rlimit_nofile-ը՝ հիշելով, որ proxy արված socket-ը nginx-ում օգտագործում է երկու կապ՝ client-ի և upstream-ի կողմերում։ Հաջորդ սահմանը մեկ կապի հիշողությունն է, ուստի չափեք այն բեռի տակ։ Proxy-ները փակում են անգործուն կապերը. nginx-ի proxy_read_timeout-ը լռելյայն 60 վայրկյան է, իսկ socket.io-ն լռելյայն ping է անում 25 վայրկյանը մեկ, ուստի timeout-ը պետք է մնա ping-ի ինտերվալից բարձր։ Laravel Reverb-ն ունի հատուկ թակարդ. դրա լռելյայն stream_select loop-ը սահմանափակված է մոտ 1,024 կապով, և դրանից ավելիի համար ձեզ պետք է ext-uv ընդլայնումը։

1map $http_upgrade $connection_upgrade {
2    default upgrade;
3    ''      close;
4}
5 
6server {
7    location /socket.io/ {
8        proxy_pass http://127.0.0.1:3001;
9        proxy_http_version 1.1;
10        proxy_set_header Upgrade $http_upgrade;
11        proxy_set_header Connection $connection_upgrade;
12        proxy_set_header Host $host;
13        proxy_read_timeout 75s;   # above the 25 s ping interval plus timeout
14        proxy_send_timeout 75s;
15    }
16}

08Առաքման երաշխիքներ՝ acknowledgement-ներ և վիճակի վերականգնում

Պարզ emit-ը «ուղարկիր և մոռացիր» է. եթե socket-ը մի պահ առաջ կտրվել է, հաղորդագրությունը կորած է։ socket.io-ն առաջարկում է երկու գործիք։ Acknowledgement-ները թույլ են տալիս ստացողին հաստատել մշակումը, իսկ emitWithAck-ը timeout-ով դա վերածում է Promise-ի, որը կարող եք await անել, այնպես որ սերվերը գիտի՝ դրամարկղն իրոք տպե՞լ է կտրոնը, և եթե ոչ, կարող է կրկին փորձել queue-ի միջոցով։ Connection state recovery-ն թույլ է տալիս client-ին, որը վերամիանում է որոշակի պատուհանի ընթացքում՝ լռելյայն երկու րոպե, հետ ստանալ իր id-ն, room-երը և բաց թողած փաթեթները։ Այն ունի սահմանափակումներ. ծածկում է միայն կարճ անջատումները, չի աջակցվում ամեն adapter-ի կողմից (դասական Redis pub/sub adapter-ը չի աջակցում, Redis Streams adapter-ը՝ աջակցում է), և լռելյայն վերականգնման ժամանակ բաց է թողնում ձեր auth middleware-ը, ինչը գուցե ցանկանաք անջատել։ Այն ամենի համար, ինչը երբեք չպետք է կորչի, հուսալի պատասխանը մնում է cursor-ով պահպանվող event log-ը։

1const io = new Server(httpServer, {
2  connectionStateRecovery: {
3    maxDisconnectionDuration: 2 * 60 * 1000,
4    skipMiddlewares: false, // re-run auth on recovered connections
5  },
6});
7 
8try {
9  const replies = await io.to('terminal:' + terminalId)
10    .timeout(5_000)
11    .emitWithAck('receipt.print', { orderId });
12  logger.info({ replies }, 'terminal confirmed');
13} catch {
14  await printQueue.add('retry', { orderId, terminalId }); // no ack in time
15}

09Fan-out, «տաք» room-եր և սահմանափակված թարմացումներ

Fan-out-ը մեկ event է, որը հասցվում է բազմաթիվ socket-ների։ Կենդանի dashboard-ի 50,000 դիտողով room-ը մեկ emit-ը վերածում է 50,000 գրառման, որը adapter-ի միջոցով բազմապատկվում է նաև instance-ների միջև։ Երեք տեխնիկա այն ողջամիտ են պահում։ Միավորեք (coalesce). յուրաքանչյուր փոփոխություն ուղարկելու փոխարեն պահեք յուրաքանչյուր entity-ի վերջին վիճակը և ուղարկեք առավելագույնը մի քանի հարյուր միլիվայրկյանը մեկ, այնպես որ վայրկյանում 40 պահեստային թարմացումը դառնում է երկու հաղորդագրություն։ Օգտագործեք volatile emit-ներ այն տվյալների համար, որտեղ կարևոր է միայն նորագույն արժեքը, օրինակ՝ cursor-ների դիրքեր կամ կենդանի մետրիկաներ. socket.volatile.emit-ը բաց է թողնում փաթեթը, եթե client-ը պատրաստ չէ, այն buffer անելու փոխարեն։ Payload-ները պահեք փոքր՝ միայն id-ներ և փոփոխված դաշտեր, և մեծ ծավալի դեպքում դիտարկեք msgpack-ի նման binary parser։ Բազմաթիվ instance-ների համար socket.io-ի sharded Redis adapter-ը օգտագործում է Redis 7-ի sharded pub/sub-ը, որպեսզի տրաֆիկը բաշխվի Redis Cluster-ով, փոխարենը որ յուրաքանչյուր հանգույց ստանա յուրաքանչյուր հաղորդագրություն։

• Չափեք հաղորդագրությունները վայրկյանում ըստ room-ի և ըստ instance-ի, ոչ միայն կապերի քանակը։
• perMessageDeflate սեղմումը socket.io-ում լռելյայն անջատված է. այն հիշողություն է արժենում յուրաքանչյուր socket-ի համար, ուստի միացրեք միայն մեծ payload-ների դեպքում։
• Շատ «տաք» room-երը կարելի է բաժանել ըստ տարածաշրջանի կամ shard-ի, որտեղ client-ները միանում են մեկ shard-ի։

10Երկարակյաց կապերի անվտանգությունը

WebSocket-ները շրջանցում են որոշ պաշտպանություններ, որոնց վրա հենվում եք HTTP-ի դեպքում։ Բրաուզերները WebSocket handshake-ի հետ ուղարկում են cookie-ներ, բայց CORS-ը դրա վրա չի տարածվում, ուստի չարամիտ էջը կարող է socket բացել ձեր սերվերի հետ օգտատիրոջ cookie-ներով. սա cross-site WebSocket hijacking է։ Cookie auth օգտագործելիս միշտ ստուգեք Origin header-ը allow-list-ով։ Միացման պահին ստուգված token-ը կարող է ժամկետանց լինել կապի փակվելուց ժամեր առաջ, ուստի պլանավորեք անջատում token-ի ժամկետի ավարտին կամ պահանջեք, որ client-ը թարմացնի և socket-ի միջոցով կրկին նույնականացվի։ Յուրաքանչյուր մուտքային հաղորդագրություն դիտարկեք HTTP հարցման body-ի պես. վավերացրեք սխեմայով, ստուգեք թույլտվությունը կոնկրետ գործողության համար և rate limit դրեք հաղորդագրությունների վրա ըստ socket-ի։ Սահմանափակեք հաղորդագրության չափը, քանի որ socket.io-ն լռելյայն ընդունում է մինչև 1 MB մեկ հաղորդագրության համար, ինչը շատ ավելին է, քան պետք է հավելվածների մեծ մասին։

1const io = new Server(httpServer, {
2  maxHttpBufferSize: 64 * 1024, // default is 1 MB
3  allowRequest: (req, cb) => cb(null, ALLOWED_ORIGINS.has(req.headers.origin ?? '')),
4});
5 
6io.on('connection', (socket) => {
7  const expiresInMs = socket.data.user.exp * 1000 - Date.now();
8  const timer = setTimeout(() => socket.disconnect(true), expiresInMs);
9  socket.on('disconnect', () => clearTimeout(timer));
10 
11  socket.on('cart.update', (raw, ack) => {
12    const msg = CartUpdateSchema.safeParse(raw);
13    if (!msg.success) return ack?.({ error: 'invalid_payload' });
14    // permission check, then handle msg.data
15  });
16});