Кто отправит outbox? Валентин Удальцов, автор канала Пых

Доклад с митапа Пыхап от канала Пых и Lamoda Tech

Валентин Удальцов, автор канала Пых, рассказывает, как эффективно отправлять сообщения, сохранённые вместе со стейтом

Transcript

1. Кто отправит Outbox? Валентин Удальцов, Пых 1
3. 😖 At-most once: 0..1 😐 At-least once: 1..n 😍 Exactly once: 1 3
5. Какая семантика обработки? 5
6. ❌ Процесс ❌ База ❌ Очередь ❌ ❌ ❌ ❌ ❌ 6
7. ❌ 7
8. ❌ 8
9. ❌ 9
10. ❌ 😐 At-least once 10
11. 😍 Exactly once 11
12. message_id 01976697-7324-70bf-bc40-700a3ff1dc51 01976697-9b47-7143-8254-cf5aba81a0aa 01976697-abc7-7025-b038-e62782cfc795 … 12
14. message_id queue 01976697-7324-70bf-bc40-700a3ff1dc51 y.events 01976697-7324-70bf-bc40-700a3ff1dc51 z.events 01976697-abc7-7025-b038-e62782cfc795 z.events 0197669e-5ab4-70b2-ae48-d58a03681c3c y.events 14
15. function handled(string $queue, string $messageId): bool { $result = $connection->execute( <<<SQL insert into deduplicator (message_id, queue) values (?, ?) on conflict (message_id, queue) do nothing SQL, [$messageId, $queue], ); return $result->affectedRows === 0; } 15
16. 😍 Exactly once 16
17. Какая семантика публикации? 17
18. ❌ 18
19. ❌ 19
20. ❌ 20
21. 😖 At-most once 🧟 Зомби-записи 21
22. Какая семантика публикации? 22
23. ❌ 23
24. ❌ 24
25. ❌ 25
26. ❌ 26
27. 😐 At-least once 👻 Призрачные сообщения 27
28. ? 28
30. Кто отправит outbox? 30
33. А нужен ли нам фоновый скрипт? 33
35. incoming_message_id queue outgoing_messages 01976697-7324-70bf… y.events [{...}, {...}, …] 01976697-7324-70bf… z.events [{...}, {...}, …] 01976697-abc7-7025… z.events [{...}, {...}, …] … … … 35
37. ❌ 37
38. ❌ ❌ 38
39. ❌ ❌ 39
40. ❌ ❌ 40
41. 😐 At-least once 41
42. begin transaction; insert into deduplicator (message_id, queue) values (?, ?) on conflict (message_id, queue) do nothing; -- if (not handled) { -- handle insert into outbox (incoming_message_id, queue, outgoing_messages) values (?, ?, ?); -- else { select outgoing_messages from outbox where incoming_message_id = ?; -- } commit; -- publish events delete from outbox where incoming_message_id = ?; 42
43. incoming_message_id queue outgoing_messages 01976697-7324-70bf… y.events [] 01976697-7324-70bf… z.events [] 01976697-abc7-7025… z.events [{...}, {...}, …] … … … 43
44. 😐 At-least once 44
45. select outgoing_messages from outbox where incoming_message_id = ?; -- if (does not exist) { begin transaction; -- handle insert into outbox (incoming_message_id, queue, outgoing_messages) values (?, ?, ?); commit; -- } -- publish events update outbox set outgoing_messages = '[]' where incoming_message_id = ?; 45
46. 😐 At-least once 46
47. 😍 Exactly once 47
48. А если обработка в контроллере? 48
51. ❌ 😖 At-most once 🧟 Зомби-записи 51
53. ❌ 53
57. Как ещё можно отправить outbox? 57
61. Настраиваем базу данных create database app; \c app create table outbox ( message_id uuid primary key, message jsonb ); alter table outbox replica identity full; create publication sequin_pub for table outbox; select pg_create_logical_replication_slot('sequin_slot', 'pgoutput'); 61
62. Настраиваем RabbitMQ 62
63. Настраиваем RabbitMQ 63
64. Настраиваем RabbitMQ 64
65. Настраиваем Sequin databases: - name: app username: postgres password: postgres hostname: postgres database: app port: 5432 slot_name: sequin_slot publication_name: sequin_pub 65
66. Настраиваем Sequin sinks: - name: outbox database: app source: include_table_names: - public.outbox actions: - insert transform: record.message destination: type: rabbitmq host: rabbitmq port: 5672 exchange: events 66
68. insert into outbox (message_id, message) select gen_random_uuid(), json_build_object('payload', x) from generate_series(1, 20) as x 68
72. Материалы • Udi Dahan: Reliable Messaging Without Distributed Transactions • NServiceBus: Outbox • NServiceBus: Transactional session • Лекция про MessageBus • Debezium • Sequin • Демо-проект c outbox на Sequin 72