概念概述
发布订阅消息（Publish-Subscribe Messaging）指的是：发布订阅通过代理与主题解耦生产者与消费者，让多个独立消费者对同一事件做出反应，便于扇出。
从更严格的定义看，Publish-subscribe is a messaging pattern in which producers publish messages to a channel without knowledge of subscribers, and the messaging infrastructure delivers each message to all interested consumers. 该概念在工程实践中的价值，在于它能帮助团队用统一的语言描述一个具体的质量属性或设计约束，从而在评审与协作中减少歧义。
为什么需要它
It is used for event notification, fan-out to independent services, and integration where producers should not know their consumers. 它与 Message Broker、Layered Architecture、Onion Architecture 等概念经常一起出现，用于解决相近的工程问题。理解它的适用场景，有助于在多种备选方案之间做出有依据的取舍，而不是凭直觉选择技术方案。
常见误解
- publish-subscribe guarantees each message is processed exactly once
- it is identical to a queue with a single consumer
需要避免的反模式
- assuming ordering across topics is preserved
- using publish-subscribe for request-reply where a direct call would fit
实践要点
- Fan-out is the defining property: one published message reaches every subscription, so adding a consumer does not require changing the publisher.
- Delivery semantics vary by broker, and at-least-once delivery means consumers must be idempotent to tolerate duplicates.
- It is asynchronous and decoupled, which improves resilience but makes end-to-end tracing and error propagation harder than in a direct call.
参考来源
https://en.wikipedia.org/wiki/Publish%E2%80%93subscribe_pattern
代码示例
package main
import (
"bufio"
"fmt"
"net/http"
"net/http/httptest"
"strings"
"sync"
"time"
)
type Broker struct {
mu sync.Mutex
clients map[chan string]struct{}
}
func NewBroker() *Broker {
return &Broker{clients: make(map[chan string]struct{})}
}
func (b *Broker) Subscribe() chan string {
events := make(chan string, 8)
b.mu.Lock()
b.clients[events] = struct{}{}
b.mu.Unlock()
return events
}
func (b *Broker) Unsubscribe(events chan string) {
b.mu.Lock()
delete(b.clients, events)
b.mu.Unlock()
}
func (b *Broker) Broadcast(event string) {
b.mu.Lock()
defer b.mu.Unlock()
for events := range b.clients {
select {
case events <- event:
default:
}
}
}
func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
flusher, ok := w.(http.Flusher)
if !ok {
http.Error(w, "streaming unsupported", http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
w.WriteHeader(http.StatusOK)
flusher.Flush()
events := b.Subscribe()
defer b.Unsubscribe(events)
for {
select {
case <-r.Context().Done():
return
case event := <-events:
fmt.Fprintf(w, "data: %s\n\n", event)
flusher.Flush()
}
}
}
func main() {
broker := NewBroker()
server := httptest.NewServer(broker)
defer server.Close()
go func() {
time.Sleep(20 * time.Millisecond)
broker.Broadcast("deploy started")
time.Sleep(10 * time.Millisecond)
broker.Broadcast("deploy finished")
}()
response, err := http.Get(server.URL)
if err != nil {
panic(err)
}
defer response.Body.Close()
fmt.Println("content-type:", response.Header.Get("Content-Type"))
scanner := bufio.NewScanner(response.Body)
received := 0
for scanner.Scan() && received < 2 {
line := scanner.Text()
if strings.HasPrefix(line, "data: ") {
fmt.Println("event:", strings.TrimPrefix(line, "data: "))
received++
}
}
}export class EventEmitter {
constructor() {
this.listeners = new Map();
}
on(event, handler) {
const set = this.listeners.get(event) ?? new Set();
set.add(handler);
this.listeners.set(event, set);
return () => this.off(event, handler);
}
once(event, handler) {
const off = this.on(event, (payload) => {
off();
handler(payload);
});
return off;
}
off(event, handler) {
const set = this.listeners.get(event);
if (set) set.delete(handler);
}
emit(event, payload) {
const set = this.listeners.get(event);
if (!set) return 0;
for (const handler of [...set]) handler(payload);
return set.size;
}
}