Към съдържанието
wannacode

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

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

Проблем

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

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

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




def pay(order, total):
    start = Service.clock
    card.call(f"таксува {total} €")
    warehouse.call(f"запази стоката за {order}")
    mailer.call(f"потвърждение за {order}")
    ledger.call(f"{total} € за {order}")
    waited = Service.clock - start
    print(f"✓ {order}: клиентът чака {waited} тика")

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

class Service:
    clock = 0  # тиковете, които минават


    def __init__(self, name, cost, failures=0):
        self.name = name
        self.cost = cost
        self.failures = failures  # записан сценарий: колко пъти подред пада


    def call(self, what):
        Service.clock += self.cost
        if self.failures > 0:
            self.failures -= 1
            raise RuntimeError(f"{self.name} не отговаря")
        print(f"{self.name}: {what}")

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

pay(1001, 120)
mailer.failures = 1  # имейлът пада веднъж
try:
    pay(1002, 80)
except RuntimeError as e:
    print(f"✗ 1002: {e}")
Изход
карта: таксува 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). Тук е клас в паметта с речник от опашки, за да се вижда механизмът. Събитието:

@dataclass(frozen=True)
class OrderPaid:
    id: str
    order: int
    total: int
    qty: int
В Python събитието е @dataclass(frozen=True): опит да промениш поле хвърля грешка.

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

class Broker:
    def __init__(self, network):
        self.queues = {}  # абонат → неговата опашка
        self.handlers = {}
        self.network = network


    def subscribe(self, name, handler):
        self.queues[name] = deque()
        self.handlers[name] = handler


    def publish(self, event):
        for queue in self.queues.values():
            queue.append(event)  # копие за всеки абонат


    # един кръг: всеки абонат получава първото от опашката си
    def deliver(self):
        any_left = False
        for name, queue in self.queues.items():
            if not queue:
                continue
            any_left = True
            try:
                self.handlers[name](queue[0])
            except RuntimeError as e:
                print(f"  ✗ {e}: остава в опашката")
                continue
            # ack се губи: събитието остава и ще дойде пак
            if self.network.ack_lost(name):
                continue
            queue.popleft()  # ack: махни го от опашката
        return any_left

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

class Network:
    def __init__(self, *lost):
        self.lost = list(lost)  # записан сценарий: чий ack се губи


    def ack_lost(self, name):
        if name not in self.lost:
            return False
        self.lost.remove(name)
        print(f"  ✗ ack от {name} се губи")
        return True

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

card = Service("карта", 2)




def pay(order, total, qty):
    start = Service.clock
    card.call(f"таксува {total} €")
    broker.publish(OrderPaid(f"evt-{order}", order, total, qty))
    waited = Service.clock - start
    print(f"✓ {order}: клиентът чака {waited} тика")

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

mailer = Service("имейл", 25, 1)
ledger = Service("счетоводство", 4)
broker = Broker(Network())
warehouse = Warehouse()
broker.subscribe("склад", warehouse.on)
broker.subscribe("имейл", lambda e: mailer.call(f"потвърждение за {e.order}"))
broker.subscribe("счетоводство", lambda e: ledger.call(f"{e.total} € за {e.order}"))
pay(1001, 120, 2)
pay(1002, 80, 1)
while broker.deliver():
    pass
Изход
карта: таксува 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:
    def __init__(self):
        self.stock = 10


    def on(self, e):
        self.stock -= e.qty
        print(f"склад: {e.qty} бр. за {e.order}, остават {self.stock}")
mailer = Service("имейл", 25, 0)
ledger = Service("счетоводство", 4)
broker = Broker(Network("склад"))
warehouse = Warehouse()
broker.subscribe("склад", warehouse.on)
broker.subscribe("имейл", lambda e: mailer.call(f"потвърждение за {e.order}"))
broker.subscribe("счетоводство", lambda e: ledger.call(f"{e.total} € за {e.order}"))
pay(1001, 120, 2)
while broker.deliver():
    pass
Изход
карта: таксува 120 €
✓ 1001: клиентът чака 2 тика
склад: 2 бр. за 1001, остават 8
  ✗ ack от склад се губи
имейл: потвърждение за 1001
счетоводство: 120 € за 1001
склад: 2 бр. за 1001, остават 6

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

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

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

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

Виж го

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

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

class Warehouse:
    def __init__(self):
        self.stock = 10
        # id-тата на обработените събития
        self.seen = set()


    def on(self, e):
        if e.id in self.seen:
            print(f"склад: {e.id} е дубликат")
            return
        self.stock -= e.qty
        self.seen.add(e.id)
        print(f"склад: {e.qty} бр. за {e.order}, остават {self.stock}")

def publish(self, event):
    for queue in self.queues.values():
        queue.append(event)  # копие за всеки абонат


# един кръг: всеки абонат получава първото от опашката си
def deliver(self):
    any_left = False
    for name, queue in self.queues.items():
        if not queue:
            continue
        any_left = True
        try:
            self.handlers[name](queue[0])
        except RuntimeError as e:
            print(f"  ✗ {e}: остава в опашката")
            continue
        # ack се губи: събитието остава и ще дойде пак
        if self.network.ack_lost(name):
            continue
        queue.popleft()  # ack: махни го от опашката
    return any_left
Стъпка по стъпка, с думи
  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 стига и опашката се изпразва. Доставено поне веднъж, обработено точно веднъж.
mailer = Service("имейл", 25, 0)
ledger = Service("счетоводство", 4)
broker = Broker(Network("склад"))
warehouse = Warehouse()
broker.subscribe("склад", warehouse.on)
broker.subscribe("имейл", lambda e: mailer.call(f"потвърждение за {e.order}"))
broker.subscribe("счетоводство", lambda e: ledger.call(f"{e.total} € за {e.order}"))
pay(1001, 120, 2)
while broker.deliver():
    pass
Изход
карта: таксува 120 €
✓ 1001: клиентът чака 2 тика
склад: 2 бр. за 1001, остават 8
  ✗ ack от склад се губи
имейл: потвърждение за 1001
счетоводство: 120 € за 1001
склад: evt-1001 е дубликат

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

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

Пробвай

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

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

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

Пробвай

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

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

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

Идеята

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

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

  • отговорът ти трябва веднага. „Има ли стока?“ преди плащането е въпрос, не събитие. Ако без отговора не можеш да продължиш, извикай другия модул;
  • системата е малка и в един процес. Извикване или 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 е команда: един получател, който трябва да я изпълни. И двете могат да минат през опашка, но имената казват различно намерение.

Как е в езиците. В Python абонатът е всичко, което може да се извика: метод на обект (warehouse.on) или lambda. Интерфейс няма: брокерът вика каквото му дадеш. deque маха отпред бързо, а list.pop(0) мести всички елементи.

Упражнение

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

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

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

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

Още задачи

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

Тест

Провери се

1. С кое име е събитие в домейна, а не команда?
2. Ack от склада се губи по пътя. Какво прави брокерът?
3. Защо складът записва id-то в seen СЛЕД като извади стоката, а не преди това?
4. Кога е по-добре да извикаш другия модул направо, а не да пуснеш събитие?

Създаден от Иво Гергов като част от кампанията на Evo Tech за достъпно обучение по програмиране ·