Hadisələr və növbələr: Valkey, NATS və RabbitMQ.

Sayt işi fon prosesinə ötürəndə — məktub göndərmək, şəkli kiçiltmək, sifarişin ödənişini keçirmək — və ya bir servis digərlərinə nəyinsə baş verdiyini bildirəndə mesaj növbədən və ya mesaj brokerindən keçir. Layihədə onların üç növü ola bilər, hər biri servis kimi əlavə olunur: İcmal → Servislər → Servis əlavə et.

Əmin deyilsinizsə, fon tapşırıqları üçün Valkey ilə başlayın: əksər freymvorklar onu hazır şəkildə dəstəkləyir. Mesajları çoxlu növbə arasında paylamaq lazımdırsa və heç biri itməməlidirsə, RabbitMQ, servislər çoxlu kiçik mesaj mübadiləsi edirsə və ya bir-birindən cavab istəyirsə, NATS seçin.

Valkey: tapşırıq növbələri və Streams

Valkey Redis ilə uyğun yaddaşdır — Verilənlər bazaları məqaləsindəki eyni idarə olunan servis. Tətbiqin ilk Valkey servisi kodla işləyən hər servisə REDIS_URL, REDIS_HOST, REDIS_PORT və REDIS_PASSWORD dəyişənlərini verir; hər Valkey həmçinin öz RIVERVO_<NAME>_URL dəyişənini verir.

Onun vasitəsilə mesaj ötürməyin üç yolu:

• Tapşırıq növbələri — redis drayverli Laravel növbələri və Horizon, BullMQ, Celery, RQ. Kitabxana tapşırıqları Valkey-də saxlayır; fon prosesi onları bir-bir götürür və xəta ilə bitənləri təkrarlayır.
• İstehlakçı qrupları ilə Streams — XADD hadisəni axına əlavə edir, qrupun istehlakçıları onu XREADGROUP ilə oxuyur və XACK ilə təsdiqləyir. Hadisələr saxlanılır: dayanmış istehlakçı qaldığı yerdən davam edir, təsdiqlənməmiş hadisəni yenidən götürmək olar, yeni qrup isə axını əvvəldən oxuya bilər.
• Pub/Sub — PUBLISH və SUBSCRIBE mesajı yalnız həmin anda qoşulu olan abunəçilərə çatdırır. Mesaj saxlanılmır: yenidən başladılan və ya hələ qoşulmamış abunəçi onu qaçırır. Pub/Sub-dan yalnız mesajın itməsinin zərər vermədiyi hallarda (məsələn, «yenilə» siqnalı), digər bütün hallarda isə Streams və ya növbədən istifadə edin.

Valkey-in ehtiyat nüsxəsi digər verilənlər bazaları kimi hər gün çıxarılır; girişi açdıqda ona xaricdən də qoşulmaq olar. Ətraflı: Verilənlər bazaları.

NATS

NATS yüngül və sürətli mesaj mübadiləsi sistemidir, JetStream aktivdir:

• Dərc və abunəlik — servis mesajı orders.created kimi mövzuya (subject) dərc edir, həmin mövzunun hər abunəçisi onu alır. JetStream olmadan mesaj çatdırılma zəmanəti olmadan göndərilir, yuxarıdakı Pub/Sub kimi.
• Sorğu və cavab — bir servis soruşur, digəri cavab verir. Bir neçə nüsxə bir qrupda (queue group) cavab verdikdə NATS sorğuları onlar arasında paylayır.
• JetStream — etibarlı axınlar: mesajlar servisin diskində saxlanılır, istehlakçı (consumer) nəyi oxuduğunu yadda saxlayır və hər mesaj təsdiqlənənədək ən azı bir dəfə çatdırılır.
• JetStream əsasında «açar — dəyər» və obyekt yaddaşı — servislər arasında kiçik ümumi parametrlər, kilidlər və ya fayllar üçün.

Dəyişənlər. Tətbiqin ilk NATS servisi kodla işləyən hər servisə bunu verir:

NATS_URL=nats://<user>:<password>@<service name>:4222

Hər NATS servisi həmçinin eyni formada ünvanla RIVERVO_<NAME>_URL dəyişənini verir — NAME servisin böyük hərflərlə yazılmış adıdır, defislər _ ilə əvəz olunur: events adlı servis RIVERVO_EVENTS_URL verir.

JetStream axınları servisin diskində saxlayır; yaddaşda saxlanılan axınlar servisin yaddaşının təxminən dörddə birini tuta bilər.

RabbitMQ

RabbitMQ 4.1 klassik mesaj brokeridir: göndərənlər mesajları exchange-ə dərc edir, exchange onları marşrut açarlarına görə növbələrə yönləndirir, istehlakçılar hər mesajı təsdiqləyir, rədd edilmiş və ya müddəti bitmiş mesaj isə dead-letter exchange-ə keçə bilər. Mesajları yenidən başlatmadan sonra da qalmalı olan növbələr üçün quorum növbələrindən (x-queue-type: quorum arqumenti) istifadə edin və mesajları daimi (persistent) kimi dərc edin.

Dəyişənlər. Tətbiqin ilk RabbitMQ servisi kodla işləyən hər servisə bunları verir:

AMQP_URL sonundakı %2f — qoşulma ünvanlarının tələb etdiyi şəkildə yazılmış standart / virtual hostudur. Hər RabbitMQ servisi həmçinin eyni formada ünvanla öz RIVERVO_<NAME>_URL dəyişənini verir.

İdarəetmə API-si. İdarəetmə HTTP API-si tətbiqin daxili şəbəkəsində 15672 portunu dinləyir: tətbiqin servisləri ona eyni istifadəçi və şifrə ilə http://<service name>:15672 ünvanında müraciət edir, məsələn:

curl -u "$RABBITMQ_USER:$RABBITMQ_PASSWORD" "http://$RABBITMQ_HOST:15672/api/queues"

İnternetdə idarəetmə səhifəsi yoxdur. Paneldə servisin səhifəsində onun növbələrinin yalnız baxış üçün qısa siyahısı göstərilir: çatdırılmağa hazır mesajlar, təsdiqlənməmiş mesajlar və istehlakçılar.

Yalnız daxili

NATS və RabbitMQ-nun xarici ünvanı yoxdur və onlara tətbiqdən kənardan — kompüterinizdən, başqa serverdən və ya başqa tətbiqdən — qoşulmaq mümkün deyil. PostgreSQL, MySQL və Valkey-dən fərqli olaraq, onlar üçün xaricdən giriş təklif olunmur. Mesajları dərc edən və qəbul edən servislər həmin tətbiqin servisləri kimi işləyir: veb servislər, fon prosesləri və planlaşdırılmış tapşırıqlar.

Eyni növdən ikinci servis

Adi adları (REDIS_*, NATS_URL, AMQP_URL və RABBITMQ_*) yalnız tətbiqin ilk Valkey, NATS və RabbitMQ servisləri alır. Eyni növdən növbəti servis yalnız öz RIVERVO_<NAME>_URL dəyişənini əlavə edir; bu dəyişəni kodunuzda oxuyun və ya Dəyişənlər bölməsində adi adlara onun dəyərlərini təyin edin. Dəyişənlər servisə növbəti yerləşdirmədə çatır — ətraflı: Mühit dəyişənləri.

Laravel

Valkey. Dəyişənlər bölməsində təyin edin:

QUEUE_CONNECTION=redis

php artisan queue:work əmrini — Horizon ilə isə php artisan horizon — fon prosesi kimi işə salın. Ətraflı: Laravel.

RabbitMQ. Laravel-də daxili RabbitMQ drayveri yoxdur və Rivervo QUEUE_CONNECTION dəyərini özü dəyişmir. Laravel növbələri üçün RabbitMQ-dan istifadə etmək üçün:

1. Paketi quraşdırın: composer require vladimir-yuldashev/laravel-queue-rabbitmq.
2. rabbitmq bağlantısını config/queue.php faylına paketin README sənədində versiyanız üçün göstərildiyi kimi əlavə edin. O, tətbiqdə artıq olan RABBITMQ_HOST, RABBITMQ_PORT, RABBITMQ_USER, RABBITMQ_PASSWORD və RABBITMQ_VHOST dəyişənlərini oxuyur:

'rabbitmq' => [
    'driver' => 'rabbitmq',
    'hosts' => [
        [
            'host' => env('RABBITMQ_HOST', '127.0.0.1'),
            'port' => env('RABBITMQ_PORT', 5672),
            'user' => env('RABBITMQ_USER', 'guest'),
            'password' => env('RABBITMQ_PASSWORD', 'guest'),
            'vhost' => env('RABBITMQ_VHOST', '/'),
        ],
    ],
],

3. Dəyişənlər bölməsində QUEUE_CONNECTION=rabbitmq təyin edin və tətbiqi yerləşdirin. Fon prosesi əvvəlki kimi qalır — php artisan queue:work.

NATS. NATS üçün rəsmi Laravel növbə drayveri yoxdur. Öz kodunuzda PHP üçün NATS müştərisindən — məsələn, basis-company/nats — istifadə edin və ona NATS_URL dəyişənindəki host, port, istifadəçi və şifrəni ötürün. Ona əsaslanmazdan əvvəl müştərinin Sizə lazım olan imkanları (JetStream, «açar — dəyər») dəstəklədiyini yoxlayın.

Node.js

BullMQ (Valkey)

import { Queue, Worker } from 'bullmq'
import IORedis from 'ioredis'
 
const connection = new IORedis(process.env.REDIS_URL, { maxRetriesPerRequest: null })
 
// veb servisdə
const emails = new Queue('emails', { connection })
await emails.add('welcome', { userId: 42 }, { attempts: 3 })
 
// fon prosesində
new Worker('emails', async (job) => {
  await sendWelcome(job.data.userId)
}, { connection })

BullMQ prosesləri istifadə etdiyi bağlantıda maxRetriesPerRequest: null tələb edir.

Valkey Streams

import IORedis from 'ioredis'
 
const redis = new IORedis(process.env.REDIS_URL)
 
// göndərən
await redis.xadd('orders', '*', 'id', '42')
 
// «billing» qrupunun istehlakçısı
await redis.xgroup('CREATE', 'orders', 'billing', '0', 'MKSTREAM').catch((err) => {
  if (!String(err).includes('BUSYGROUP')) throw err // qrup artıq var
})
const me = process.env.HOSTNAME ?? 'worker'
for (;;) {
  const res = await redis.xreadgroup('GROUP', 'billing', me, 'COUNT', 10, 'BLOCK', 5000, 'STREAMS', 'orders', '>')
  for (const [, entries] of res ?? []) {
    for (const [id, fields] of entries) {
      await handleOrder(fields)
      await redis.xack('orders', 'billing', id)
    }
  }
}

Oxunmuş, lakin təsdiqlənməmiş qeyd qrupda gözləmədə qalır; XAUTOCLAIM onu başqa istehlakçıya ötürür.

NATS (nats.js)

import { connect, JSONCodec, AckPolicy } from 'nats'
 
const nc = await connect({ servers: process.env.NATS_URL })
const jc = JSONCodec()
 
// bir dəfə: orders.* mövzuları üçün axın və daimi istehlakçı
const jsm = await nc.jetstreamManager()
await jsm.streams.add({ name: 'ORDERS', subjects: ['orders.>'] })
await jsm.consumers.add('ORDERS', { durable_name: 'billing', ack_policy: AckPolicy.Explicit })
 
// göndərən
const js = nc.jetstream()
await js.publish('orders.created', jc.encode({ id: 42 }))
 
// istehlakçı
const consumer = await js.consumers.get('ORDERS', 'billing')
for await (const m of await consumer.consume()) {
  await handleOrder(jc.decode(m.data))
  m.ack()
}

Sorğu və cavab: bir servis nc.subscribe('prices.get', { queue: 'prices', callback: (err, msg) => msg.respond(jc.encode(price)) }) ilə cavab verir, digəri await nc.request('prices.get', jc.encode({ sku: 'A1' }), { timeout: 2000 }) ilə soruşur.

RabbitMQ (amqplib)

import amqp from 'amqplib'
 
const conn = await amqp.connect(process.env.AMQP_URL)
const ch = await conn.createChannel()
await ch.assertQueue('emails', { durable: true, arguments: { 'x-queue-type': 'quorum' } })
 
// göndərən
ch.sendToQueue('emails', Buffer.from(JSON.stringify({ userId: 42 })), { persistent: true })
 
// istehlakçı: eyni anda ən çox 10 təsdiqlənməmiş mesaj
await ch.prefetch(10)
await ch.consume('emails', async (msg) => {
  try {
    await sendWelcome(JSON.parse(msg.content.toString()).userId)
    ch.ack(msg)
  } catch {
    ch.nack(msg, false, false) // növbənin dead-letter exchange-i varsa, ora
  }
})

Növbəni elan edən hər servis onu eyni arqumentlərlə elan etməlidir.

Python

Celery (Valkey və ya RabbitMQ)

import os
from celery import Celery
 
app = Celery("tasks", broker=os.environ["REDIS_URL"])  # və ya os.environ["AMQP_URL"]
 
@app.task(acks_late=True)
def send_welcome(user_id):
    ...

celery -A tasks worker --loglevel=info əmrini fon prosesi kimi işə salın. Celery amqp:// ünvanlarını olduğu kimi qəbul edir. RabbitMQ ilə Celery-nin son versiyaları quorum növbələrini elan edə bilir (task_default_queue_type = "quorum") — versiyanız üçün Celery sənədlərinə baxın.

RQ (Valkey)

import os
from redis import Redis
from rq import Queue
 
queue = Queue("emails", connection=Redis.from_url(os.environ["REDIS_URL"]))
queue.enqueue(send_welcome, 42)

Fon prosesi: rq worker emails --url "$REDIS_URL".

NATS (nats-py)

import asyncio, json, os
import nats
from nats.errors import TimeoutError
 
async def main():
    nc = await nats.connect(os.environ["NATS_URL"])
    js = nc.jetstream()
    await js.add_stream(name="ORDERS", subjects=["orders.>"])
 
    await js.publish("orders.created", json.dumps({"id": 42}).encode())
 
    sub = await js.pull_subscribe("orders.>", durable="billing")
    while True:
        try:
            msgs = await sub.fetch(10, timeout=5)
        except TimeoutError:
            continue
        for msg in msgs:
            await handle_order(json.loads(msg.data))
            await msg.ack()
 
asyncio.run(main())

RabbitMQ (pika)

import json, os
import pika
 
conn = pika.BlockingConnection(pika.URLParameters(os.environ["AMQP_URL"]))
ch = conn.channel()
ch.queue_declare(queue="emails", durable=True, arguments={"x-queue-type": "quorum"})
 
# göndərən
ch.basic_publish(exchange="", routing_key="emails",
                 body=json.dumps({"user_id": 42}),
                 properties=pika.BasicProperties(delivery_mode=2))
 
# istehlakçı
def on_message(ch, method, properties, body):
    send_welcome(json.loads(body)["user_id"])
    ch.basic_ack(delivery_tag=method.delivery_tag)
 
ch.basic_qos(prefetch_count=10)
ch.basic_consume(queue="emails", on_message_callback=on_message)
ch.start_consuming()

Təkrar çatdırılma

Proses mesajı təsdiqləməzdən əvvəl dayanarsa — yerləşdirmə, yenidən başlatma və ya bərpa zamanı — növbələr və brokerlər mesajı yenidən çatdırır. İstehlakçıları elə yazın ki, eyni mesajın iki dəfə emalı zərər verməsin: emal olunmuş hər mesajın identifikatorunu saxlayın və ya əməliyyatın özünü təkrarlana bilən edin (məsələn, «ödəniş əlavə et» əvəzinə «sifarişi ödənilmiş kimi qeyd et»).

Resurslar və qiymət

Hər birinin qiyməti istənilən servis kimi onun yaddaşı, CPU və diskinə görə hesablanır — ətraflı: Layihənin servisləri. Resursları sonradan servisin Parametrlər bölməsində dəyişmək olar.

Ehtiyat nüsxələr və bərpa

NATS və RabbitMQ-nun ehtiyat nüsxəsi hər gün çıxarılır; digər servislərdə olduğu kimi son 7 günün və 4 həftənin nüsxələri saxlanılır.

• Ehtiyat nüsxə servis işlədiyi müddətdə onun məlumat fayllarını köçürür. RabbitMQ üçün eyni nüsxədə onun təriflərinin ixracı da saxlanılır: istifadəçilər, virtual hostlar, növbələr, exchange-lər, bağlamalar (bindings) və siyasətlər.
• Dəqiq ana bərpa yoxdur — bu imkan yalnız PostgreSQL və MySQL üçündür. Gündəlik nüsxələrdən biri bərpa olunur.
• Bərpa zamanı servis qısa müddətə dayanır, fayllar yerinə qaytarılır və servis yenidən işə düşür. Nüsxənin çıxarıldığı vaxt ətrafında gələn və ya təsdiqlənən mesajlar olmaya və ya yenidən çatdırıla bilər — istehlakçıları təkrara davamlı etmək üçün daha bir səbəb.
• RabbitMQ üçün: köçürülmüş faylları işə salmaq mümkün olmazsa, servis yalnız təriflərlə işə düşür — növbələr, exchange-lər və istifadəçilər yerindədir, lakin nüsxədəki mesajlar itir.

Texniki xidmət və köçürmə

NATS və RabbitMQ üçün texniki xidmət vaxtı təyin olunmur. Onlar başqa serverə avtomatik köçürülmür; bu nə vaxtsa lazım olarsa, dəstək xidməti köçürməni Sizinlə razılaşdırır.