RabbitMQ

01Գաղափարը՝ փոստատուն հաղորդագրությունների համար

Պատկերացրեք RabbitMQ-ն որպես փոստատուն։ Producer-ը նամակը (message) թողնում է պատուհանում (exchange)՝ հասցեի պիտակով (routing key)։ Փոստատունը այն ըստ կանոնների (bindings) տեսակավորում է փոստարկղերի (queues) մեջ։ Consumer-ը դատարկում է իր փոստարկղը և ստորագրում յուրաքանչյուր նամակի համար (acknowledgement)։ Եթե consumer-ը երբեք չի ստորագրում, նամակը վերադառնում է՝ նորից առաքվելու։ Այս բաժանումը նշանակում է, որ producer-ը չգիտի և չի մտածում, թե ով է մշակում հաղորդագրությունը, քանի worker կա կամ արդյոք նրանք հիմա առցանց են․ հաղորդագրությունները ապահով սպասում են հերթում։

• Message՝ payload (բայթեր, սովորաբար JSON) և հատկություններ, ինչպիսիք են content type-ը, message ID-ն և header-ները։
• Queue՝ կարգավորված բուֆեր, որը պահում է հաղորդագրությունները, մինչև consumer-ը դրանք վերցնի։
• Connection և channel՝ մեկ TCP connection մեկ պրոցեսի համար, իսկ դրա ներսում թեթև channel-ներ յուրաքանչյուր thread-ի կամ առաջադրանքի համար։

02Exchange-ներ և ուղղորդում

Producer-ները երբեք ուղղակիորեն հերթում չեն հրապարակում․ նրանք հրապարակում են exchange-ում։ Exchange-ի տեսակն է որոշում ուղղորդումը։ Direct exchange-ը առաքում է այն հերթերին, որոնց binding key-ը ճշգրիտ համընկնում է routing key-ի հետ։ Topic exchange-ը համեմատում է կետերով pattern-ներ․ «order.*»-ը համընկնում է «order.created»-ի հետ, իսկ «order.#»-ը՝ ցանկացած խորության։ Fanout exchange-ը յուրաքանչյուր հաղորդագրություն պատճենում է բոլոր կապված հերթերին՝ անտեսելով key-երը։ Headers exchange-ը ուղղորդում է ըստ header-ների արժեքների։ Իրադարձությունների համար տարածված դիզայնն է՝ մեկ topic exchange յուրաքանչյուր դոմենի համար (orders, users) և մեկ հերթ consumer-ի յուրաքանչյուր նպատակի համար (email.order-created, analytics.orders), որպեսզի յուրաքանչյուր consumer ստանա իր պատճենը և աշխատի իր տեմպով։

• Default exchange՝ անանուն direct exchange, որը ուղղորդում է այն հերթին, որի անունը համընկնում է routing key-ի հետ։
• Չուղղորդվող հաղորդագրությունները կորչում են, եթե չեք հրապարակում mandatory դրոշով կամ չեք կարգավորել alternate exchange։

03Հերթերի տեսակները՝ quorum, classic և streams

Quorum queue-ները տվյալները ռեպլիկացնում են կենտ թվով node-երի վրա՝ օգտագործելով Raft կոնսենսուսի ալգորիթմը․ հաղորդագրությունը հաստատվում է միայն այն բանից հետո, երբ մեծամասնությունն այն ունի, ուստի այն վերապրում է node-ի խափանումը։ Դրանք առաջարկվող ընտրությունն են ամեն կարևոր բանի համար։ Classic queue-ները մեկ node-ի վրա են և չեն ռեպլիկացվում, քանի որ RabbitMQ 4.0-ը հեռացրեց classic mirrored queue-ները․ դրանք լավ են ժամանակավոր կամ վերարտադրելի աշխատանքի համար։ Streams-ը append-only log է․ consumer-ները կարդում են offset-ից, և հաղորդագրությունները սպառելիս չեն ջնջվում, ինչը հնարավոր է դարձնում replay և շատ մեծ fan-out՝ Kafka-ի topic-ների նման։ Quorum queue-ները աջակցում են նաև delivery limit, որը պաշտպանում է poison message-ներից, որոնք կրկին ու կրկին կործանում են consumer-ներին։

04Հուսալի հրապարակում

Durable հերթերը և persistent հաղորդագրությունները վերապրում են broker-ի վերագործարկումը, բայց producer-ին չեն ասում, որ broker-ը ստացել է հաղորդագրությունը։ Դրա համար օգտագործեք publisher confirms․ broker-ը հաստատում է յուրաքանչյուր հաղորդագրություն, երբ այն ապահով պահված է (quorum queue-ների դեպքում՝ ռեպլիկացված մեծամասնությանը)։ Եթե confirm-ը չի գալիս կամ վերադառնում է որպես nack, նորից հրապարակեք։ Յուրաքանչյուր հաղորդագրության տվեք եզակի messageId, որպեսզի consumer-ները կարողանան հայտնաբերել կրկնօրինակները, քանի որ կրկնումները առաքումը դարձնում են at-least-once։ Exchange-ներն ու հերթերը հայտարարեք կոդում կամ definitions ֆայլում, որպեսզի միջավայրերը վերարտադրելի լինեն։

1import amqp from "amqplib";
2 
3const conn = await amqp.connect("amqp://your_user:your_password@localhost:5672");
4const ch = await conn.createConfirmChannel();
5 
6await ch.assertExchange("orders", "topic", { durable: true });
7await ch.assertQueue("email.order-created", {
8  durable: true, arguments: { "x-queue-type": "quorum" },
9});
10await ch.bindQueue("email.order-created", "orders", "order.created");
11 
12ch.publish("orders", "order.created", Buffer.from(JSON.stringify({ orderId: 42 })), {
13  persistent: true, contentType: "application/json", messageId: "order-42-created",
14});
15await ch.waitForConfirms(); // throws if the broker rejected the message

05Սպառում՝ ack-եր, prefetch և կրկնումներ

Consumer-ները պետք է օգտագործեն ձեռքով acknowledgement․ ack արեք միայն այն բանից հետո, երբ աշխատանքը հաջողվել է։ Եթե consumer-ը կործանվում է մինչև ack-ը, RabbitMQ-ն հաղորդագրությունը նորից առաքում է մեկ այլ consumer-ի։ Prefetch-ը (basic.qos) սահմանափակում է, թե քանի չհաստատված հաղորդագրություն կարող է պահել մեկ consumer-ը․ առանց դրա broker-ը ամեն ինչ մղում է առաջին consumer-ին, իսկ մյուսները պարապ են մնում։ Սկսեք 10–50 prefetch-ից և կարգավորեք։ Ձախողման դեպքում որոշեք․ requeue արեք անցողիկ սխալի դեպքում, կամ reject արեք առանց requeue-ի, որպեսզի հաղորդագրությունը գնա dead letter exchange (DLX)՝ ուսումնասիրության կամ ուշացված կրկնման համար։ Երբեք ack մի արեք մինչև մշակումը և handler-ները դարձրեք idempotent։

1await ch.prefetch(20);
2 
3await ch.consume("email.order-created", async (msg) => {
4  if (!msg) return;
5  try {
6    const event = JSON.parse(msg.content.toString());
7    await sendOrderEmail(event.orderId); // must be idempotent
8    ch.ack(msg);
9  } catch (err) {
10    // false, false = do not requeue: route to the dead letter exchange
11    ch.nack(msg, false, false);
12  }
13}, { noAck: false });

06Dead letter-ներ, policy-ներ և շահագործում

Dead letter exchange-ը ստանում է այն հաղորդագրությունները, որոնք մերժվել են, TTL-ով ժամկետանց են դարձել կամ գերազանցել են delivery limit-ը։ Կարգավորեք այն policy-ով, ոչ թե կոդում կոշտ գրված queue arguments-ով, քանի որ policy-ները կարելի է փոխել առանց հերթերը վերահայտարարելու։ Կրկնման դասական pattern-ը ձախողված հաղորդագրություններն ուղղորդում է TTL-ով retry հերթ, որի սեփական DLX-ը ցույց է տալիս հետ դեպի աշխատանքային exchange-ը՝ տալով ուշացված կրկնումներ։ Quorum queue-ների համար աշխատեցրեք RabbitMQ-ն որպես երեք node-անոց cluster, միացրեք management plugin-ը UI-ի և HTTP API-ի համար, մետրիկաները բացեք Prometheus plugin-ով և օգտագործեք TLS ու յուրաքանչյուր հավելվածի համար առանձին user՝ vhost-ի սահմանափակ թույլտվություններով։ Ջնջեք լռելյայն guest user-ը, որը, ամեն դեպքում, աշխատում է միայն localhost-ից։

1# Dead-letter failed messages and cap redeliveries for all email.* queues
2rabbitmqctl set_policy email-dlx "^email\." \
3  '{"dead-letter-exchange":"dlx","delivery-limit":5}' --apply-to quorum_queues
4 
5rabbitmqctl add_vhost shop
6rabbitmqctl add_user orders_app 'use-a-strong-password'
7rabbitmqctl set_permissions -p shop orders_app '^orders.*' '^orders.*' '^(orders|email).*'

07Quorum queue-ները ներսից

Quorum queue-ն Raft խումբ է․ մեկ leader և follower-ներ այլ node-երի վրա, սովորաբար երեք կամ հինգ անդամ։ Publish-ները գնում են leader-ին, որն ավելացնում է դրանք իր log-ին և ռեպլիկացնում․ երբ մեծամասնությունը գրառումը գրել է սկավառակին, leader-ը հաստատում է այն publisher-ին։ Եթե leader-ի node-ը խափանվում է, մնացած մեծամասնությունը մի քանի վայրկյանում ընտրում է նոր leader, և consumer-ները շարունակում են։ Քանի որ յուրաքանչյուր հաղորդագրություն ռեպլիկացվում և fsync է արվում, quorum queue-ները ավելի շատ disk I/O են օգտագործում և լավագույնն են այն հերթերի համար, որոնք պարբերաբար դատարկվում են, ոչ թե օրերով կայանված միլիոնավոր հաղորդագրությունների համար։ Դրանք հետևում են վերաառաքումների քանակին, կիրառում են delivery limit (4.0-ից ի վեր կա լռելյայն արժեք) և 4.0-ից աջակցում են հաղորդագրությունների առաջնահերթություններ։ Շատ երկար հերթերը պետք է դառնան streams։

• Հերթերի մեծ մասի համար օգտագործեք երեք անդամ․ հինգը հանդուրժում են երկու խափանում՝ ավելի մեծ գնով։
• Անդամները տեղաբաշխվում են cluster-ի node-երի վրա․ node-երը պահեք առանձին availability zone-երում։

08Flow control, alarm-ներ և back-pressure

RabbitMQ-ն պաշտպանում է ինքն իրեն։ Երբ node-ի memory-ի օգտագործումը անցնում է high watermark-ը (լռելյայն՝ RAM-ի մի մասը) կամ ազատ սկավառակը իջնում է disk limit-ից ցածր, այն alarm է բարձրացնում և ամբողջ cluster-ում արգելափակում է հրապարակող բոլոր connection-ները, մինչև ռեսուրսները վերականգնվեն։ Consumer-ները շարունակում են աշխատել, և այդպես համակարգը դատարկվում է։ Credit-based flow control-ը նաև դանդաղեցնում է արագ publisher-ներին, որոնք առաջ են անցնում հերթի պրոցեսներից։ Գործնականում alarm-ները ախտանիշ են այն բանի, որ հերթը աճում է ավելի արագ, քան սպառվում է։ Անսահմանափակ աճող հերթերի վրա սահմանեք max-length կամ TTL policy-ներ, alert դրեք հերթի խորության աճի տեմպի վրա, մասշտաբավորեք consumer-ները և broker-ին տվեք արագ սկավառակներ՝ պաշարով։ Client-ները պետք է մշակեն connection.blocked ծանուցումը և դադարեցնեն կամ բուֆերացնեն հրապարակումը։

09Կրկնումներ backoff-ով և poison message-ներ

Անմիջական requeue-ը կրկնում է չափազանց արագ և կարող է poison message-ը ցիկլի մեջ գցել։ Հուսալի pattern-ը ուշացումների համար օգտագործում է dead lettering․ ձախողման դեպքում consumer-ը մերժում է հաղորդագրությունը, որը DLX-ի միջոցով գնում է retry հերթ՝ message TTL-ով (օրինակ 30 վայրկյան) և առանց consumer-ների․ երբ TTL-ը լրանում է, այդ հերթը այն dead-letter է անում հետ դեպի հիմնական exchange-ը։ Էքսպոնենցիալ քայլերի համար (30 վ, 5 րոպե, 30 րոպե) օգտագործեք մի քանի retry հերթ և փորձերը հաշվեք header-ում կամ հենվեք quorum queue-ների x-delivery-count header-ի վրա։ Վերջին փորձից հետո ուղղորդեք parking հերթ, որը մարդն է ուսումնասիրում։ Կա նաև delayed message exchange plugin, բայց այն չի ռեպլիկացվում, ուստի կրիտիկական աշխատանքի համար TTL հերթերն ավելի անվտանգ են։

1# Retry queue: holds failed messages for 30 s, then sends them back
2rabbitmqctl set_policy retry-30s "^retry\.30s$" \
3  '{"message-ttl":30000,"dead-letter-exchange":"orders"}' --apply-to queues
4 
5# Main work queues dead-letter into the retry exchange
6rabbitmqctl set_policy work-dlx "^email\." \
7  '{"dead-letter-exchange":"retry","delivery-limit":10}' --apply-to quorum_queues

10Exactly-once-ը առասպել է՝ idempotency և outbox

Ցանցերը ձախողվում են «մշակված»-ի և «ack արված»-ի միջև, ինչպես նաև «հրապարակված»-ի և «հաստատված»-ի միջև, ուստի ցանկացած հուսալի համակարգ at-least-once է։ Երկու pattern դա դարձնում է անվտանգ։ Transactional outbox-ը producer-ի կողմում․ բիզնես փոփոխությունը և outbox տողը գրեք նույն database transaction-ում, և թող relay-ը outbox տողերը հրապարակի confirms-ով ու նշի որպես ուղարկված։ Սա կանխում է «պատվերը պահվեց, բայց իրադարձությունը կորավ» և «իրադարձությունն ուղարկվեց, բայց պատվերը rollback եղավ» իրավիճակները։ Idempotent consumer-ը ստացող կողմում․ մշակված message ID-ները պահեք (կամ օգտագործեք բնական key-եր և upsert-ներ) նույն transaction-ում, ինչ side effect-ը, և բաց թողեք կրկնօրինակները։ Երկուսի դեպքում համակարգը վարվում է այնպես, կարծես առաքումը exactly once լիներ։