Keçən dəfə hadisəni tək consumer oxuyurdu. Amma mağaza böyüyür: hər sifariş OrderCreated, OrderPaid və OrderShipped hadisələrini yaradır, bildiriş servisi isə artıq məktubları çatdırmağa çatmır. Servisin bir neçə nüsxəsini işə salıb işi elə bölmək istəyirik ki, heç bir hadisə iki dəfə emal olunmasın. Üstəlik, eyni hadisələr analitikaya da lazımdır və o, məktublara mane olmamalıdır. Bütün bunları consumer qrupları həll edir. Həmçinin emal ilə əlfəcinin saxlanması arasında nasazlıq olanda nə baş verdiyini və hadisələri yenidən necə oxumağı öyrənəcəyik.

Consumer qrupu

Eyni groupId-li bütün consumer-lər qrup təşkil edir. Kafka topic-in partition-larını qrup üzvləri arasında elə bölür ki, hər partition-ın hər an qrupdan tam bir oxuyanı olsun. Bənzətmə: mağazada üç kassa (partition-lar) və kassirlər (consumer-lər). Hər kassanı bir kassir xidmət edir, yoxsa alıcıların növbəsi qarışar; bir kassir iki kassanı apara bilər, artıq kassir isə kassasız qalır.

Qrupdakı consumer-lərKim nəyi oxuyur (3 partition üçün nümunə)
1A: 0, 1, 2 partition-ları
2A: 0 və 2, B: 1
3A: 0, B: 1, C: 2
4üçü birər partition oxuyur, dördüncü boş durur
Üç partition bir qrupun consumer-ləri arasında bölünür; üçüncü qoşulanda rebalance baş verir və hər biri bir partition alır
Üç partition bir qrupun consumer-ləri arasında bölünür; üçüncü qoşulanda rebalance baş verir və hər biri bir partition alır

Buradan miqyaslama qaydası çıxır: paralel oxuma partition sayından çox ola bilməz. Servisin səkkiz nüsxəsini istəyirsinizsə, ən azı səkkiz partition lazımdır. Budur, hansı partition-ların ona düşdüyünü göstərən consumer:

js
import { Kafka } from 'kafkajs';

const name = process.argv[2] ?? 'A';
const kafka = new Kafka({
  clientId: `notifications-${name}`,
  brokers: ['localhost:9092'],
});
const consumer = kafka.consumer({ groupId: 'notifications' });

consumer.on(consumer.events.GROUP_JOIN, ({ payload }) => {
  const mine = payload.memberAssignment.orders ?? [];
  console.log(name, 'got partitions', mine);
});

await consumer.connect();
await consumer.subscribe({ topic: 'orders', fromBeginning: true });
await consumer.run({
  eachMessage: async ({ partition, message }) => {
    const event = JSON.parse(message.value.toString());
    console.log(name, `p${partition}`, event.type, event.orderId);
  },
});

Onu iki terminalda işə salın: node group.js A və node group.js B. Çıxış belə ola bilər (hansı consumer hansı partition-ları alır, alqoritmdən asılıdır; kafkajs-in xidməti JSON sətirləri buraxılıb):

text
A got partitions [ 0, 2 ]
B got partitions [ 1 ]
A p2 OrderCreated 3
B p1 OrderCreated 4
A p0 OrderCreated 5

Üçüncünü işə salın — qrup partition-ları yenidən bölüşdürəcək:

text
A got partitions [ 0 ]
B got partitions [ 1 ]
C got partitions [ 2 ]

Rebalance