Message Broker на Go

Message broker — одна из тех вещей, которые обычно воспринимаются как инфраструктура, существующая где-то за пределами приложения.

Мы отправляем сообщение в Kafka или RabbitMQ:

producer.Send(ctx, message)

а затем другой сервис получает его:

consumer.Consume(ctx)

Внутри же происходит довольно много интересного: хранение сообщений, подтверждения, повторная доставка, конкуренция, порядок сообщений и т.д.

В данной статье я хочу разобрать работу брокеров сообщений на примере Go кода, написав минимальную демонстрацию.

Цель — разобраться, из каких примитивов состоит брокер сообщений.

В данной статье будет описана работа с:

очередями;
producer и consumer;
конкурентной обработкой;
acknowledgements;
повторной доставкой;
persistence на диске;
consumer groups;
graceful shutdown.

Что мы хотим получить

Начнём с простой модели.

Есть producer и consumer:

Producer │ ▼ ┌─────────────┐ │ Broker │ │ │ │ Queue │ └─────────────┘ │ ▼ Consumer

Producer публикует сообщение в очередь. Consumer подписывается на очередь и получает сообщения.

Например:

type Message struct {
	ID      uint64
	Payload []byte
}

А API брокера может выглядеть примерно так:

type Broker interface {
	Publish(context.Context, string, []byte) error

	Subscribe(
		context.Context,
		string,
		Handler,
	) error
}

Но сразу возникает несколько вопросов.

Что происходит, если consumer умер после получения сообщения?
Что происходит, если broker перезапустился?
Можно ли доставить одно сообщение нескольким consumer'ам?
Что будет, если два consumer'а одновременно пытаются получить одно сообщение?

Чтобы ответить на эти вопросы, будем постепенно усложнять реализацию.

Самая простая очередь

Для начала забудем сделаем очередь в памяти:

type Queue struct {
	mu       sync.Mutex
	messages []Message
}

Добавление сообщения:

func (queue *Queue) Push(message Message) {
	queue.mu.Lock()
	defer queue.mu.Unlock()

	queue.messages = append(queue.messages, message)
}

Получение:

func (queue *Queue) Pop() (Message, bool) {
	queue.mu.Lock()
	defer queue.mu.Unlock()

	if len(queue.messages) == 0 {
		return Message{}, false
	}

	message := queue.messages[0]
	queue.messages = queue.messages[1:]

	return message, true
}

Уже можно сделать простой broker:

type Broker struct {
	queues map[string]*Queue
	mu     sync.RWMutex
}

Producer публикует:

func (b *Broker) Publish(topic string, payload []byte) {
	b.mu.RLock()
	queue := b.queues[topic]
	b.mu.RUnlock()

	queue.Push(Message{
		Payload: payload,
	})
}

А consumer периодически вызывает Pop().

Проблема очевидна: если очередь пустая, приходится либо постоянно опрашивать её, либо блокироваться каким-то другим способом. Для этого гораздо лучше подходят channels.

Channels как транспорт внутри брокера

Можно представить очередь следующим образом:

type Queue struct {
	messages chan Message
}

Создание:

queue := &Queue{
	messages: make(chan Message, 1024),
}

Producer:

queue.messages <- message

Consumer:

message := <-queue.messages

Получается очень простая схема:

Producer │ ▼ channel | ▼ Consumer

Но у такого решения есть серьёзный недостаток. channel хранит сообщения только в памяти.

После:

broker ↓ restart ↓ RAM очищена ↓ сообщения потеряны

Для настоящего брокера этого недостаточно.

Добавляем идентификаторы сообщений

Первое, что понадобится для надёжной доставки, — стабильный ID.

type Message struct {
	ID        uint64
	Topic     string
	Payload   []byte
	Timestamp time.Time
}

ID пригодится для нескольких вещей:

подтверждения обработки;
повторной доставки;
хранения сообщений;
восстановления состояния после перезапуска;
дедупликации.

Простейший генератор:

type IDGenerator struct {
	next atomic.Uint64
}

func (g *IDGenerator) Next() uint64 {
	return g.next.Add(1)
}

Теперь каждое сообщение получает уникальный идентификатор.

Почему обычного Pop недостаточно

Представим:

Broker → Consumer

Broker отправил сообщение:

ID=42

Consumer получил его, но через миллисекунду процесс упал. Если broker уже удалил сообщение из очереди, оно потеряно.

Поэтому получение сообщения и его удаление из очереди — не одно и то же.

Нам нужна модель:

READY │ ▼ DELIVERED │ ├── ACK ─────► DONE │ └── timeout ─► READY

Это одна из ключевых идей message broker.

In-flight сообщения

Добавим структуру для сообщений, которые уже отправлены consumer'у:

type Delivery struct {
	Message  Message
	Consumer string
	Deadline time.Time
}

Broker теперь имеет две коллекции:

type Queue struct {
	ready    []Message
	inFlight map[uint64]Delivery

	mu sync.Mutex
}

Когда consumer получает сообщение:

func (q *Queue) Deliver(consumer string) (Message, bool) {
	q.mu.Lock()
	defer q.mu.Unlock()

	if len(q.ready) == 0 {
		return Message{}, false
	}

	message := q.ready[0]
	q.ready = q.ready[1:]

	q.inFlight[message.ID] = Delivery{
		Message:  message,
		Consumer: consumer,
		Deadline: time.Now().Add(30 * time.Second),
	}

	return message, true
}

Теперь сообщение не исчезает окончательно. Оно находится в inFlight.

ACK

После успешной обработки consumer должен подтвердить сообщение:

func (q *Queue) Ack(id uint64) bool {
	q.mu.Lock()
	defer q.mu.Unlock()

	if _, ok := q.inFlight[id]; !ok {
		return false
	}

	delete(q.inFlight, id)

	return true
}

Теперь жизненный цикл сообщения выглядит так:

publish │ ▼ READY │ ▼ DELIVERED │ ▼ ACK │ ▼ DONE

Если ACK не пришёл, сообщение можно вернуть обратно в очередь.

Retry

Запустим отдельную goroutine, которая периодически проверяет inFlight.

func (q *Queue) retryExpired() {
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()

	for range ticker.C {
		q.mu.Lock()

		now := time.Now()

		for id, delivery := range q.inFlight {
			if now.Before(delivery.Deadline) {
				continue
			}

			q.ready = append(q.ready, delivery.Message)
			delete(q.inFlight, id)
		}
	}
}

Теперь если consumer получил сообщение и не отправил ACK, через некоторое время оно будет доставлено повторно. Это даёт нам at-least-once delivery.

Что такое at-least-once

У нас теперь есть гарантия:

Сообщение будет доставлено как минимум один раз.

Но не обязательно ровно один.

Например:

Broker → Consumer │ ├── обработал сообщение │ └── процесс умер ДО ACK

Broker этого не знает. Через timeout он отправит сообщение снова:

Broker → Consumer │ └── обработал второй раз

Поэтому consumer должен быть готов к повторной обработке.

Например, операция:

increase balance by 100

может быть опасной при повторном выполнении. Лучше использовать idempotency key:

type Message struct {
	ID      uint64
	Payload []byte
}

И хранить уже обработанные ID.

Consumer groups

Теперь рассмотрим другую проблему.

Допустим, у нас три consumer'а:

┌── Consumer 1 Broker ──────┼── Consumer 2 └── Consumer 3

Мы хотим, чтобы одно сообщение обработал только один consumer. То есть:

Message 1 → Consumer 1 Message 2 → Consumer 2 Message 3 → Consumer 3 Message 4 → Consumer 1

Это и есть базовая модель consumer group.

Можно хранить список consumers:

type Consumer struct {
	ID string
}

А broker выбирать следующего consumer'а по round-robin:

type ConsumerGroup struct {
	consumers []Consumer
	next      int
	mu        sync.Mutex
}

Выбор:

func (g *ConsumerGroup) Next() Consumer {
	g.mu.Lock()
	defer g.mu.Unlock()

	consumer := g.consumers[g.next]

	g.next++
                    
	if g.next >= len(g.consumers) {
		g.next = 0
	}

	return consumer
}

Это очень простая реализация. В реальной системе всё сложнее: consumer может умереть, появиться новый consumer, измениться количество partition'ов и так далее.

Persistence

До сих пор после рестарта broker забывает всё. Исправим это.

Самый простой вариант — append-only log. Каждое сообщение записываем в файл:

┌────────────┬──────────────┐ │ message ID │ payload size │ ├────────────┴──────────────┤ │ payload │ └───────────────────────────┘

Например:

type Log struct {
	file *os.File
	mu   sync.Mutex
}

Запись:

func (l *Log) Append(message Message) error {
	l.mu.Lock()
	defer l.mu.Unlock()

	data, err := json.Marshal(message)

	if err != nil {
		return err
	}

	_, err = l.file.Write(append(data, '\n'))

	return err
}

После записи желательно сделать:

l.file.Sync()

если нам нужна гарантия, что данные дошли до диска.

Почему JSON — не лучший формат

Для демонстрационного брокера JSON удобен. Но у него есть недостатки:

большой размер;
медленный парсинг;
нет фиксированного размера записи;
сложнее и затратнее искать сообщения.

Более реалистичный вариант — бинарный формат.

Например:

[4 bytes length][8 bytes ID][payload]

Тогда восстановление журнала выглядит примерно так:

for {
	length, err := readUint32(file)

	if errors.Is(err, io.EOF) {
		break
	}

	if err != nil {
		return err
	}

	payload := make([]byte, length)

	if _, err := io.ReadFull(file, payload); err != nil {
		return err
	}

	restore(payload)
}

При запуске broker читает log и восстанавливает состояние.

Что делать с ACK после рестарта

Здесь появляется интересная проблема.

Предположим:

1. message 42 записано 2. message 42 отправлено consumer 3. consumer отправил ACK 4. broker упал

Если ACK не был сохранён, после восстановления broker может снова отправить 42. Это нормально для at-least-once delivery.

Но если мы хотим более сложные гарантии, нам понадобится отдельный журнал состояния или commit log. И здесь мы постепенно приходим к архитектуре, похожей на ту, которую используют настоящие брокеры.

API брокера

Теперь можно сделать внешний интерфейс.

Например, HTTP API:

POST /topics/orders

Тело:

{
	"order_id": 123
}

Broker создаёт сообщение:

Message{
	ID:      42,
	Topic:   "orders",
	Payload: []byte(...),
}

Consumer может подключаться через TCP или HTTP streaming. Но для большого количества сообщений постоянный HTTP polling будет не самым удачным решением. Поэтому для нашего брокера интереснее сделать собственный TCP-протокол.

Простейший TCP-протокол

Например:

PUBLISH orders 42 128
<128 bytes payload>

ACK 42

SUBSCRIBE orders worker-1

Соответственно:

Producer │ │ PUBLISH ▼ Broker │ │ MESSAGE ▼ Consumer │ │ ACK ▼ Broker

Go очень хорошо подходит для такого сервера. Можно принимать соединения:

func (b *Broker) handleConnection(conn net.Conn) {
	defer conn.Close()

	for {
		command, err := readCommand(conn)
		
		if err != nil {
			return
		}

		switch command.Type {
		case CommandPublish:
			b.publish(command)

		case CommandSubscribe:
			b.subscribe(command)

		case CommandAck:
			b.ack(command)
		}
	}
}

Backpressure

Теперь представим ситуацию:

Producer: 100 000 msg/s Consumer: 10 msg/s

Если broker бесконечно принимает сообщения в память, процесс рано или поздно закончится нехваткой памяти. Нам нужен backpressure.

Один из вариантов — ограничить очередь:

const maxQueueSize = 100_000

Когда очередь заполнена, producer получает ошибку:

QUEUE_FULL

Другой вариант — блокировать публикацию до появления свободного места. Это важная часть дизайна любой очереди.

Dead Letter Queue

А что делать с сообщением, которое постоянно падает?

Например:

Message 42 │ ├── attempt 1 → error ├── attempt 2 → error ├── attempt 3 → error ├── attempt 4 → error └── ...

Бесконечно retry делать нельзя. Добавим максимальное количество попыток:

type Message struct {
	ID       uint64
	Payload  []byte
	Attempts int
}

После, например, пяти неудачных попыток:

orders │ └── Message 42 │ ▼ dead-letter

Dead Letter Queue позволяет отдельно разбирать сообщения, которые невозможно обработать автоматически.

Graceful shutdown

Broker нельзя просто завершить:

os.Exit(0)

Иначе можно потерять сообщения.

При получении SIGTERM нужно:

перестать принимать новые connections;
перестать принимать новые сообщения;
завершить текущие операции;
сохранить необходимые данные;
корректно закрыть consumers;
закрыть файлы;
завершить процесс.

Например:

ctx, stop := signal.NotifyContext(
	context.Background(),
	syscall.SIGTERM,
	syscall.SIGINT,
)

defer stop()

<-ctx.Done()

broker.Shutdown()

В случае брокера сообщений shutdown можно назвать даже отдельной частью архитектуры.

Какая гарантия доставки у нас получилась?

После всех изменений у нас есть примерно следующая модель:

Publish │ ▼ Persistent log │ ▼ Ready queue │ ▼ Consumer │ ├── ACK ─────► Done │ └── timeout ─► Retry │ └── max retries │ ▼ Dead Letter

Гарантия: at-least-once delivery.

Это означает, что сообщения могут доставляться повторно. Для большинства практических систем это вполне разумная модель.

Об exactly-once

На первый взгляд можно сказать: "Давайте просто сделаем exactly-once."

Но проблема в том, что broker контролирует только доставку сообщения. Он не контролирует внешнюю операцию consumer'а.

Broker после timeout отправит сообщение снова. Поэтому exactly-once delivery нельзя получить простым добавлением ещё одного флага.

Заключение

В результате у нас получился небольшой message broker.

Цель статьи была не написать промышленный брокер, а понять, какие задачи вообще приходится решать подобной системе.

Какие функции в теории можно добавить:

partitioning;
несколько broker'ов;
replication;
leader election;
consumer group rebalancing;
batching;
zero-copy передача;
метрики.

«The only way to go fast, is to go well.» — Robert C. Martin

Message broker на Go

Что мы хотим получить

Самая простая очередь

Channels как транспорт внутри брокера

Добавляем идентификаторы сообщений

Почему обычного Pop недостаточно

In-flight сообщения

ACK

Retry

Что такое at-least-once

Consumer groups

Persistence

Почему JSON — не лучший формат

Что делать с ACK после рестарта

API брокера

Простейший TCP-протокол

Backpressure

Dead Letter Queue

Graceful shutdown

Какая гарантия доставки у нас получилась?

Об exactly-once

Заключение