Менеджер публикаций‑подписок

Условие

Требуется реализовать объект‑менеджер PubSub, предоставляющий два публичных метода:

• Subscribe(id string) chan any — регистрирует подписку на канал с заданным идентификатором и возвращает канал, из которого подписчик может получать сообщения.
• Publish(id string, msg any) — отправляет сообщение в канал с указанным id; все активные подписчики должны получить копию сообщения. Пример использования:

m := PubSubManager{}
msgChan := m.Subscribe("id1")
msgChan2 := m.Subscribe("id1")

m.Publish("id1", "hello world")
m.Publish("id1", "hi")
m.Publish("id1", "test")

log.Println(<-msgChan, <-msgChan, <-msgChan)   // prints: hello world, hi, test
log.Println(<-msgChan2, <-msgChan2, <-msgChan2) // prints: hello world, hi, test

type PubSubManager struct {
    // ...
}

func (m *PubSubManager) Publish(id string, msg any) {
    // ...
}

func (m *PubSubManager) Subscribe(id string) chan any {
    // ...
}

Ответ

Каждая подписка должна иметь отдельный канал, иначе сообщения будут распределяться между читателями, а не рассылаться всем. Ограниченный буфер допускает пример из условия и создаёт backpressure.

Простой вариант с импортом sync, буфером 16 и сериализованной публикацией:

type PubSubManager struct {
    mu sync.Mutex
    subscribers map[string][]chan any
}
func (m *PubSubManager) Subscribe(id string) chan any {
    m.mu.Lock()
    defer m.mu.Unlock()
    if m.subscribers == nil { m.subscribers = make(map[string][]chan any) }
    ch := make(chan any, 16)
    m.subscribers[id] = append(m.subscribers[id], ch)
    return ch
}
func (m *PubSubManager) Publish(id string, msg any) {
    m.mu.Lock()
    defer m.mu.Unlock()
    for _, ch := range m.subscribers[id] { ch <- msg }
}

Подписчики только читают и не закрывают каналы. После заполнения буфера Publish блокируется на медленном читателе, включая другие темы; здесь это явная простая политика, не масштабируемый брокер. Для эксплуатации нужны Unsubscribe/Close, отмена и изоляция очередей. Передаётся копия значения интерфейса, не глубокая копия объекта.