Събития и опашки

Код на:JavaC#PythonJavaScript

Накратко: Събитието казва „това стана“ и отива в опашка, от която всеки абонат го взима сам; доставката е „поне веднъж“, затова потребителят трябва да е идемпотентен.

Проблем

Бавният имейл бави плащането

Когато клиентът плати, магазинът прави четири неща: таксува картата, запазва стоката в склада, праща имейл с потвърждение и записва плащането в счетоводството. Плащането ги вика едно след друго:

static final Service card = new Service("карта", 2, 0);
static final Service warehouse = new Service("склад", 3, 0);
static final Service mailer = new Service("имейл", 25, 0); // бавен SMTP: 25 тика
static final Service ledger = new Service("счетоводство", 4, 0);

static void pay(int order, int total) {
    int start = Service.clock;
    card.call("таксува " + total + " €");
    warehouse.call("запази стоката за " + order);
    mailer.call("потвърждение за " + order);
    ledger.call(total + " € за " + order);
    int waited = Service.clock - start;
    System.out.println("✓ " + order + ": клиентът чака " + waited + " тика");
}

Всяка „услуга“ тук е обект с брояч: всяко извикване добавя своите тикове. Тиковете са времето в програмата, вместо секунди. Имейлът минава през бавен пощенски сървър и струва 25 тика:

class Service {
    static int clock = 0; // тиковете, които минават
    final String name;
    final int cost;
    int failures; // записан сценарий: колко пъти подред пада

    Service(String name, int cost, int failures) {
        this.name = name;
        this.cost = cost;
        this.failures = failures;
    }

    void call(String what) {
        clock += cost;
        if (failures > 0) {
            failures--;
            throw new IllegalStateException(name + " не отговаря");
        }
        System.out.println(name + ": " + what);
    }
}

Първото плащане минава. После имейлът пада веднъж:

pay(1001, 120);
mailer.failures = 1; // имейлът пада веднъж
try {
    pay(1002, 80);
} catch (IllegalStateException e) {
    System.out.println("✗ 1002: " + e.getMessage());
}

Изход

карта: таксува 120 €
склад: запази стоката за 1001
имейл: потвърждение за 1001
счетоводство: 120 € за 1001
✓ 1001: клиентът чака 34 тика
карта: таксува 80 €
склад: запази стоката за 1002
✗ 1002: имейл не отговаря

Клиентът чака 34 тика, от които 25 са за имейл, който може да дойде и след минута. А при поръчка 1002 картата е таксувана и стоката е запазена, но клиентът вижда грешка, защото пощенският сървър не отговаря. Той натиска „Плати“ пак.

Има и трета болка. Плащането познава всички, които реагират на него. Утре маркетингът иска точки за лоялност, а мобилното приложение иска известие. Всеки нов ред е промяна в кода на плащането.

Идеята

Събитие: това вече стана

Плащането не трябва да казва на склада „запази“, на имейла „прати“ и на счетоводството „запиши“. Достатъчно е да съобщи какво е станало: „поръчка 1001 е платена“. Това е събитие в домейна (domain event): факт от езика на магазина, в минало време, с име като OrderPaid. Носи своето id и данните, които трябват на другите: номера на поръчката, сумата, броя. След като е станало, не се променя.

Събитието отива в опашка от съобщения (message queue). Програмата, която пази опашките, е брокер. Всеки, който иска да знае за плащанията, се абонира. Брокерът пази отделна опашка за всеки абонат и слага копие на събитието във всяка. Това е publish/subscribe: плащането публикува, абонатите получават, и никой от двете страни не знае за другата.

Идеята я видя в Observer: който иска да знае, се абонира. Разликата е опашката по средата. Observer вика абонатите веднага, в същото извикване, и чака всеки от тях. Тук плащането оставя събитието в опашката и отговаря на клиента. Абонатите го взимат, когато могат, както опаковчиците в производител и потребител.

Плащането и складът вече могат да са отделни модули, дори отделни услуги, както в предишния урок: говорят си само през събития.

Виж го

Брокерът в паметта

Истинският брокер е отделна програма (Kafka, RabbitMQ). Тук е клас в паметта с речник от опашки, за да се вижда механизмът. Събитието:

record OrderPaid(String id, int order, int total, int qty) {}

В Java събитието е record: полетата се дават в конструктора и не се променят.

Брокерът. Абонатът дава име и функция, която обработва събитието:

class Broker {
    private final Map<String, Deque<OrderPaid>> queues = new LinkedHashMap<>();
    private final Map<String, Consumer<OrderPaid>> handlers = new HashMap<>();
    private final Network network;

    Broker(Network network) {
        this.network = network;
    }

    void subscribe(String name, Consumer<OrderPaid> handler) {
        queues.put(name, new ArrayDeque<>());
        handlers.put(name, handler);
    }

    void publish(OrderPaid event) {
        for (var queue : queues.values()) {
            queue.add(event); // копие за всеки абонат
        }
    }

    // един кръг: всеки абонат получава първото от опашката си
    boolean deliver() {
        boolean any = false;
        for (var name : queues.keySet()) {
            var queue = queues.get(name);
            if (queue.isEmpty()) continue;
            any = true;
            try {
                handlers.get(name).accept(queue.peek());
            } catch (IllegalStateException e) {
                System.out.println("  ✗ " + e.getMessage()
                        + ": остава в опашката");
                continue;
            }
            // ack се губи: събитието остава и ще дойде пак
            if (network.ackLost(name)) continue;
            queue.poll(); // ack: махни го от опашката
        }
        return any;
    }
}

Важното е в deliver. Абонатът получава първото събитие от опашката си, но то излиза оттам чак когато абонатът свърши без грешка. Потвърждението „свърших“ се казва ack (acknowledgement). Ако абонатът хвърли грешка, ack няма и събитието остава за следващия кръг. Загубеният ack идва от класа Network: записан сценарий чий ack да се загуби. Тук той не губи нищо, а на следващата карта губи.

class Network {
    private final List<String> lost; // записан сценарий: чий ack се губи

    Network(String... lost) {
        this.lost = new ArrayList<>(List.of(lost));
    }

    boolean ackLost(String name) {
        if (!lost.remove(name)) return false;
        System.out.println("  ✗ ack от " + name + " се губи");
        return true;
    }
}

Плащането вече само таксува картата и публикува:

static final Service card = new Service("карта", 2, 0);
static Broker broker;

static void pay(int order, int total, int qty) {
    int start = Service.clock;
    card.call("таксува " + total + " €");
    broker.publish(new OrderPaid("evt-" + order, order, total, qty));
    int waited = Service.clock - start;
    System.out.println("✓ " + order + ": клиентът чака " + waited + " тика");
}

Две поръчки, а имейлът пада веднъж, както в проблема:

Service mailer = new Service("имейл", 25, 1);
Service ledger = new Service("счетоводство", 4, 0);
broker = new Broker(new Network());
Warehouse warehouse = new Warehouse();
broker.subscribe("склад", warehouse::on);
broker.subscribe("имейл", e -> mailer.call("потвърждение за " + e.order()));
broker.subscribe("счетоводство", e -> ledger.call(e.total() + " € за " + e.order()));
pay(1001, 120, 2);
pay(1002, 80, 1);
while (broker.deliver()) { }

Изход

карта: таксува 120 €
✓ 1001: клиентът чака 2 тика
карта: таксува 80 €
✓ 1002: клиентът чака 2 тика
склад: 2 бр. за 1001, остават 8
  ✗ имейл не отговаря: остава в опашката
счетоводство: 120 € за 1001
склад: 1 бр. за 1002, остават 7
имейл: потвърждение за 1001
счетоводство: 80 € за 1002
имейл: потвърждение за 1002

Клиентът чака 2 тика и при двете поръчки. Имейлът пак е бавен и пак пада, но това вече е негова грешка: събитието остава в неговата опашка и идва пак в следващия кръг. Складът и счетоводството не разбират.

Виж и реда. Потвърждението за 1001 тръгва след като складът е обработил 1002. Всяка опашка върви със своето темпо. Затова абонатите не бива да разчитат, че друг абонат вече е свършил своето.

Идеята

Поне веднъж значи понякога два пъти

Брокерът маха събитието от опашката само след ack. Какво става, ако складът извади стоката, но ack-ът се изгуби по мрежата? Брокерът не може да различи „складът падна, преди да започне“ от „складът свърши, но не успя да каже“. За да не изгуби поръчка, той доставя събитието пак.

Сценарият губи ack от склада веднъж. Складът е наивен: всяко събитие вади стоката:

class Warehouse {
    int stock = 10;

    void on(OrderPaid e) {
        stock -= e.qty();
        System.out.printf("склад: %d бр. за %d, остават %d%n",
                e.qty(), e.order(), stock);
    }
}

Service mailer = new Service("имейл", 25, 0);
Service ledger = new Service("счетоводство", 4, 0);
broker = new Broker(new Network("склад"));
Warehouse warehouse = new Warehouse();
broker.subscribe("склад", warehouse::on);
broker.subscribe("имейл", e -> mailer.call("потвърждение за " + e.order()));
broker.subscribe("счетоводство", e -> ledger.call(e.total() + " € за " + e.order()));
pay(1001, 120, 2);
while (broker.deliver()) { }

Изход

карта: таксува 120 €
✓ 1001: клиентът чака 2 тика
склад: 2 бр. за 1001, остават 8
  ✗ ack от склад се губи
имейл: потвърждение за 1001
счетоводство: 120 € за 1001
склад: 2 бр. за 1001, остават 6

Една поръчка за 2 броя, а в склада липсват 4. Наличността е грешна и магазинът ще откаже следваща поръчка, макар стоката да е на рафта.

Брокерът може да обещае едно от двете:

• най-много веднъж: маха събитието, преди абонатът да работи. Дубликати няма, но падне ли абонатът, събитието е изгубено;
• поне веднъж: маха го след ack. Нищо не се губи, но понякога идва два пъти.

„Точно веднъж“ през мрежа брокерът не може да обещае сам. Затова почти всички системи избират „поне веднъж“, а дубликата го хваща получателят.

Виж го

Идемпотентен потребител

Потребител, при когото второто получаване на същото събитие не променя нищо, е идемпотентен (idempotent consumer). Складът помни id-тата на събитията, които вече е обработил, и хвърля дубликата:

class Warehouse {
    // id-тата на обработените събития
    private final Set<String> seen = new HashSet<>();
    int stock = 10;

    void on(OrderPaid e) {
        if (seen.contains(e.id())) {
            System.out.println("склад: " + e.id() + " е дубликат");
            return;
        }
        stock -= e.qty();
        seen.add(e.id());
        System.out.printf("склад: %d бр. за %d, остават %d%n",
                e.qty(), e.order(), stock);
    }
}

void publish(OrderPaid event) {
    for (var queue : queues.values()) {
        queue.add(event); // копие за всеки абонат
    }
}

// един кръг: всеки абонат получава първото от опашката си
boolean deliver() {
    boolean any = false;
    for (var name : queues.keySet()) {
        var queue = queues.get(name);
        if (queue.isEmpty()) continue;
        any = true;
        try {
            handlers.get(name).accept(queue.peek());
        } catch (IllegalStateException e) {
            System.out.println("  ✗ " + e.getMessage()
                    + ": остава в опашката");
            continue;
        }
        // ack се губи: събитието остава и ще дойде пак
        if (network.ackLost(name)) continue;
        queue.poll(); // ack: махни го от опашката
    }
    return any;
}

Стъпка по стъпка, с думи

1. Плащането таксува картата и пуска събитие OrderPaid evt-1001: „поръчка 1001 е платена“. Клиентът чака само 2 тика.
2. Брокерът слага копие във всяка опашка: склад, имейл и счетоводство. Плащането вече е отговорило.
3. Във всеки кръг всеки абонат получава първото събитие от своята опашка. То остава в опашката до ack.
4. Складът вади 2 броя (остават 8) и записва evt-1001 в seen. Имейлът и счетоводството вършат своето.
5. Ack от имейла и счетоводството стига и събитието излиза от опашките им. Ack от склада се губи по пътя.
6. Брокерът не знае, че складът е свършил работата, и доставя evt-1001 пак. Това е „поне веднъж“.
7. evt-1001 вече е в seen и складът го хвърля. Наличността остава 8, не 6.
8. Този път ack стига и опашката се изпразва. Доставено поне веднъж, обработено точно веднъж.

Service mailer = new Service("имейл", 25, 0);
Service ledger = new Service("счетоводство", 4, 0);
broker = new Broker(new Network("склад"));
Warehouse warehouse = new Warehouse();
broker.subscribe("склад", warehouse::on);
broker.subscribe("имейл", e -> mailer.call("потвърждение за " + e.order()));
broker.subscribe("счетоводство", e -> ledger.call(e.total() + " € за " + e.order()));
pay(1001, 120, 2);
while (broker.deliver()) { }

Изход

карта: таксува 120 €
✓ 1001: клиентът чака 2 тика
склад: 2 бр. за 1001, остават 8
  ✗ ack от склад се губи
имейл: потвърждение за 1001
счетоводство: 120 € за 1001
склад: evt-1001 е дубликат

Брокерът пак доставя два пъти, но складът вади стоката веднъж. id-то се записва след работата: ако работата падне по средата, събитието ще дойде пак и ще бъде обработено.

Някои операции са идемпотентни сами: „статусът на 1001 е платена“ е същото и на втория път. „Наличност минус 2“ не е. За тях трябва id. В истинска система seen е таблица в същата база като наличността, а изваждането и записът на id-то са в една транзакция. Иначе срив между двете пак дава грешка. Какво става, когато записът в базата и пращането на събитието са в различни системи, е в следващия урок.

Пробвай

Предвиди изхода

Складът е идемпотентен. Този път сценарият губи ack от имейла. Какво печата програмата?

Service mailer = new Service("имейл", 25, 0);
Service ledger = new Service("счетоводство", 4, 0);
broker = new Broker(new Network("имейл"));
Warehouse warehouse = new Warehouse();
broker.subscribe("склад", warehouse::on);
broker.subscribe("имейл", e -> mailer.call("потвърждение за " + e.order()));
broker.subscribe("счетоводство", e -> ledger.call(e.total() + " € за " + e.order()));
pay(1001, 120, 2);
while (broker.deliver()) { }

Отговор

карта: таксува 120 €
✓ 1001: клиентът чака 2 тика
склад: 2 бр. за 1001, остават 8
имейл: потвърждение за 1001
  ✗ ack от имейл се губи
счетоводство: 120 € за 1001
склад: 2 бр. за 1001, остават 6
имейл: потвърждение за 1001

карта: таксува 120 €
✓ 1001: клиентът чака 2 тика
склад: 2 бр. за 1001, остават 8
имейл: потвърждение за 1001
  ✗ ack от имейл се губи
счетоводство: 120 € за 1001
имейл: evt-1001 е дубликат

карта: таксува 120 €
✓ 1001: клиентът чака 2 тика
склад: 2 бр. за 1001, остават 8
имейл: потвърждение за 1001
  ✗ ack от имейл се губи
счетоводство: 120 € за 1001
имейл: потвърждение за 1001

Провери

Пробвай

Намери грешката

Колега преписва склада и тестът с изгубения ack пак показва, че стоката се вади два пъти. На кой ред е грешката?

Избери реда с грешката.

1class Warehouse {2 int stock = 10;34 void on(OrderPaid e) {5 Set<String> seen = new HashSet<>(); // id-тата на обработените събития6 if (seen.contains(e.id())) {7 System.out.println("склад: " + e.id() + " е дубликат");8 return;9 }10 stock -= e.qty();11 seen.add(e.id());12 System.out.printf("склад: %d бр. за %d, остават %d%n",13 e.qty(), e.order(), stock);14 }15}

Провери

Идеята

Кога събитията вредят

Събитията не са безплатни. Не ги ползвай, когато:

• отговорът ти трябва веднага. „Има ли стока?“ преди плащането е въпрос, не събитие. Ако без отговора не можеш да продължиш, извикай другия модул;
• системата е малка и в един процес. Извикване или Observer стигат. Брокерът е още една програма, която трябва да работи, да се следи и да се обновява.

И знай цената, когато ги ползваш:

• потокът вече не е на едно място. Какво става след плащането, не се чете в една функция, а се събира от абонатите. Затова всяко събитие носи id, по което после се търси в логовете;
• другите научават по-късно. Клиентът плаща и веднага отваря „Моите поръчки“, а фактурата още я няма. Това се казва евентуална съгласуваност. Екранът трябва да го казва честно: „фактурата идва до минута“;
• редът не е обещан. Видя го в програмата: опашките вървят със свое темпо. Ако ред има значение (платена, после отказана), събитията на една поръчка трябва да минат през една опашка и един потребител;
• дубликатите са твоя грижа. Всеки абонат трябва да е идемпотентен.

Как се пуска и следи истински брокер в работа (клъстер, дискове, наблюдение) е тема на DevOps курса ZeroToOps.

По-точно, за любопитните

Истинските брокери. Kafka пази събитията като дневник, който абонатите четат всеки от своето място. Там ack е запомненото място (offset). RabbitMQ има опашки и ack за всяко съобщение, като в урока. Има още Redis Streams, Amazon SQS и SNS, Azure Service Bus и Google Pub/Sub. В Spring абонатът е метод с @KafkaListener или @RabbitListener. В ASP.NET Core често се ползва MassTransit. В Python има клиенти като confluent-kafka и aio-pika, а Celery е опашка от задачи. В Node.js са kafkajs и amqplib, а NestJS има модул за съобщения.

Опашка за мъртви съобщения (dead letter queue). Събитие, което пада всеки път (повредени данни), не бива да върти опашката вечно. След няколко опита брокерът го мести в отделна опашка, където човек го гледа.

Точно веднъж. Kafka има режим „exactly once“, но той важи вътре в Kafka: прочети, обработи, запиши обратно в Kafka. Щом ефектът е навън (имейл, плащане, база), дубликатът пак е твоя грижа.

Събитие или команда. OrderPaid е събитие: много абонати, никой не е задължен. ReserveStock е команда: един получател, който трябва да я изпълни. И двете могат да минат през опашка, но имената казват различно намерение.

Как е в езиците. В Java абонатът е Consumer<OrderPaid>, затова се подава и препратка към метод (warehouse::on). ArrayDeque е опашката, а LinkedHashMap пази реда на абонатите.

Упражнение

Известие в приложението

Реши задачата в редактора си.

Мобилното приложение иска известие „поръчката е платена“. Добави четвърти абонат, „приложение“, без да пипаш кода на плащането. Той трябва да е идемпотентен, като склада. Сценарият губи неговия ack веднъж:

Изход

карта: таксува 120 €
✓ 1001: клиентът чака 2 тика
склад: 2 бр. за 1001, остават 8
имейл: потвърждение за 1001
счетоводство: 120 € за 1001
приложение: известие за 1001
  ✗ ack от приложение се губи
приложение: evt-1001 е дубликат

Още задачи

• Отказана поръчка. Добави събитие OrderCancelled. Складът връща стоката, но само ако е обработил OrderPaid за същата поръчка. Какво става, ако отказът дойде преди плащането?
• Опашка за мъртви съобщения. Ако абонат падне три пъти подред на едно и също събитие, брокерът го мести в списък dead и продължава със следващото.
• Точки за лоялност. Добави абонат, който дава по 1 точка на всеки 10 € от сумата. Какво трябва да помни, за да не даде точките два пъти?

Тест

Провери се

1. С кое име е събитие в домейна, а не команда?

SendConfirmationEmail: „прати имейл“OrderPaid: „поръчката е платена“ReserveStock: „запази стоката“

2. Ack от склада се губи по пътя. Какво прави брокерът?

Маха събитието: складът сигурно го е получилСпира всички опашки, докато складът не потвърдиДоставя събитието пак, защото не знае дали складът го е обработил

3. Защо складът записва id-то в seen СЛЕД като извади стоката, а не преди това?

Ако работата падне по средата, събитието ще дойде пак и ще се обработи. Записано преди, щеше да бъде хвърлено като дубликат, без стоката да е изваденаНяма разлика, редът не влияеЗащото множеството е по-бързо след изваждането

4. Кога е по-добре да извикаш другия модул направо, а не да пуснеш събитие?

Когато отговорът ти трябва веднага, например „има ли стока?“ преди плащанетоВинаги, събитията са само за големи фирмиНикога, събитията винаги са по-добри

Провери

Започни урока отначало

Свързани уроци

• Нужно отпреди: Observer · Singleton и Dependency Injection · Производител и потребител · Монолит, модулен монолит, услуги
• Продължава в: Transactional outbox · Дизайн на API: договор, версии, повторени заявки · ADR: записани архитектурни решения · Нефункционални изисквания, които се мерят
• Задачи по темата: Софтуерни архитектури