Đồng thờiConcurrency

Publish-Subscribeaka Pub/Sub, Topic-based messaging, Event Bus

Publisher gửi message vào một topic; mọi subscriber của topic đó đều nhận một bản, không ai biết ai.Publishers send messages to a topic; every subscriber of that topic gets a copy, and nobody knows anybody.

Độ khóDifficulty: VừaMedium
Mức phổ biếnUsage: Mức phổ biếnUsage 5/5
Thời gian đọcReading time: ~9 phút đọc~9 min read

Thảo luậnDiscussion

Ví dụ đời thườngReal-life analogy

Một kênh Zalo OA của cửa hàng: chủ shop đăng 'Flash sale 12.12' một lần, ai đã quan tâm kênh đều nhận thông báo. Chủ shop không cần biết danh sách người theo dõi, người theo dõi cũng không cần quen chủ shop. Ai bỏ theo dõi thì thôi nhận, ai mới theo dõi thì nhận từ bài sau.A shop's Zalo Official Account: the owner posts '12.12 Flash Sale' once, and everyone following the channel gets notified. The owner doesn't need the follower list, and followers don't need to know the owner. Unfollow and you stop receiving; follow now and you get the next posts.

Vấn đề và giải phápProblem and solution

Không có patternWithout the pattern

Khi có đơn hàng mới, OrderService phải gọi lần lượt EmailService, LoyaltyService, AnalyticsService, WarehouseService... Mỗi lần thêm một bên quan tâm lại phải sửa OrderService. Một service chậm (SMTP 3 giây) làm chậm cả luồng đặt hàng, một service sập làm đặt hàng lỗi theo. OrderService bị trói chặt vào mọi thứ xung quanh.When an order is placed, OrderService calls EmailService, LoyaltyService, AnalyticsService, WarehouseService one after another... Every new interested party means editing OrderService. One slow service (3-second SMTP) slows down checkout, one crashed service makes checkout fail. OrderService is tied to everything around it.

Với Publish-SubscribeWith Publish-Subscribe

Đưa vào một broker với các topic. OrderService chỉ publish sự kiện order.placed rồi đi tiếp. Mỗi bên quan tâm tự subscribe topic đó và xử lý theo nhịp của mình (bất đồng bộ). Thêm subscriber mới không đụng tới publisher. Broker có thể là channel trong cùng process (Go), hoặc hệ thống ngoài như Kafka, Redis, NATS khi cần đi qua nhiều service.Introduce a broker with topics. OrderService only publishes an order.placed event and moves on. Each interested party subscribes to that topic and processes at its own pace (asynchronously). Adding a subscriber never touches the publisher. The broker can be in-process channels (Go), or an external system like Kafka, Redis or NATS when crossing services.

Thành phần tham giaParticipants

Publisher: Phát message vào một topic, không biết ai sẽ nhận.Emits messages to a topic without knowing who receives them.
Broker: Quản lý topic và danh sách subscriber, phân phát một bản cho mỗi subscriber.Manages topics and subscriber lists, delivers one copy per subscriber.
Topic: Tên kênh logic (order.placed) — điểm hẹn giữa hai bên.A logical channel name (order.placed) — the meeting point for both sides.
Subscriber: Đăng ký topic và xử lý message độc lập, theo tốc độ riêng.Subscribes to a topic and processes messages independently, at its own speed.

Sơ đồ và luồng thực thiDiagram and execution flow

Dùng Tự chạy hoặc phím mũi tên để đi qua từng bước gọi giữa các object. Console hiển thị output tương ứng của chương trình.Use Autoplay or the arrow keys to step through each call between objects. The console shows the matching program output.

1Lúc khởi động, hai worker subscribe topic order.placed; broker tạo cho MỖI worker một channel riêng có buffer.At startup, two workers subscribe to order.placed; the broker creates a SEPARATE buffered channel for each worker.

← →

console

1subscribe order.placed → 2 subscribers

implementsimplementsextendsextendsgọi / dùngcalls / usestạo racreatesgửi qua channelsends via channel1thứ tự chạy (bấm để nhảy bước)execution order (click to jump)

Code: Java Spring Boot ⇄ GoCode: Java Spring Boot ⇄ Go

Cùng một bài toán, hai ngôn ngữ. Rê chuột hoặc focus vào chú thích (số màu cam) để làm nổi bật dòng code tương ứng.Same problem, two languages. Hover or focus an annotation (orange number) to highlight its line.

JavaSpring Boot 3 · Java 21Spring Boot 3 · Java 21

package com.shop.order;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

record OrderPlaced(String orderId, String userId, long total) {}1

@Service
public class OrderService {
    static final String TOPIC = "order.placed";
    private static final Logger log = LoggerFactory.getLogger(OrderService.class);
    private final KafkaTemplate<String, OrderPlaced> kafka;
    private final OrderRepository orders;

    public OrderService(KafkaTemplate<String, OrderPlaced> kafka, OrderRepository orders) {
        this.kafka = kafka;
        this.orders = orders;
    }

    public void place(Order order) {
        orders.save(order);2
        var event = new OrderPlaced(order.id(), order.userId(), order.total());
        // Key = orderId → all events of one order go to the same partition (ordered)
        kafka.send(TOPIC, order.id(), event)3
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("publish failed for order {}", order.id(), ex);
                }
            });
    }
}

GoGo 1.23 · thư viện chuẩnGo 1.23 · standard library

package pubsub

import "sync"

type Event struct {
	Topic   string
	Payload any
}

type Broker struct {
	mu   sync.RWMutex
	subs map[string][]chan Event1
}

func NewBroker() *Broker {
	return &Broker{subs: make(map[string][]chan Event)}
}

// Subscribe returns a dedicated buffered channel for this subscriber.
func (b *Broker) Subscribe(topic string, buf int) <-chan Event {2
	ch := make(chan Event, buf)
	b.mu.Lock()
	b.subs[topic] = append(b.subs[topic], ch)
	b.mu.Unlock()
	return ch
}

// Publish never blocks: a slow subscriber drops events instead of stalling everyone.
func (b *Broker) Publish(topic string, payload any) {
	b.mu.RLock()
	defer b.mu.RUnlock()
	for _, ch := range b.subs[topic] {
		select {3
		case ch <- Event{Topic: topic, Payload: payload}:
		default: // buffer full → drop (count a metric in real code)
		}
	}
}

// Close closes every subscriber channel so their range loops end.
func (b *Broker) Close() {
	b.mu.Lock()4
	defer b.mu.Unlock()
	for topic, chans := range b.subs {
		for _, ch := range chans {
			close(ch)
		}
		delete(b.subs, topic)
	}
}

Khác biệt chính giữa Java và GoKey differences between Java and Go

01Trong một process, Go dựng broker chỉ bằng map + channel mỗi subscriber + RWMutex. Java thường dùng ApplicationEventPublisher của Spring (đồng bộ mặc định, @Async để bất đồng bộ) hoặc đi thẳng ra Kafka/Redis.In-process, Go builds a broker from just a map + one channel per subscriber + an RWMutex. Java usually uses Spring's ApplicationEventPublisher (synchronous by default, @Async for async) or goes straight to Kafka/Redis.
02Subscriber Go là goroutine for ev := range ch — tự kiểm soát vòng lặp. Spring đảo ngược điều khiển: bạn chỉ viết method có @KafkaListener, container lo poll, thread, retry.A Go subscriber is a goroutine running for ev := range ch — you own the loop. Spring inverts control: you just write a method with @KafkaListener, and the container handles polling, threads and retries.
03Backpressure trong Go phải chọn tường minh: chặn, bỏ (select default), hay buffer. Kafka giải quyết bằng cách lưu log trên đĩa — subscriber chậm chỉ bị 'lag' chứ không làm chậm publisher.Backpressure in Go is an explicit choice: block, drop (select default) or buffer. Kafka solves it by storing a log on disk — a slow subscriber just 'lags' without slowing the publisher.
04Ngữ nghĩa nhóm: Kafka dùng groupId, NATS dùng queue group — khác nhóm = pub-sub (mỗi nhóm một bản), cùng nhóm = chia việc.Group semantics: Kafka uses groupId, NATS uses queue groups — different groups = pub-sub (one copy per group), same group = work sharing.

Ứng dụng thực tếReal-world usage

Kafka, RabbitMQ (fanout/topic exchange), Google Pub/SubKafka, RabbitMQ (fanout/topic exchange), Google Pub/Sub
Xương sống của kiến trúc hướng sự kiện: một sự kiện order.placed nuôi hàng chục service downstream.The backbone of event-driven architectures: one order.placed event feeds dozens of downstream services.
Redis Pub/Sub và NATSRedis Pub/Sub and NATS
Nhẹ, độ trễ thấp, không lưu trữ: xoá cache trên mọi instance, broadcast vị trí tài xế tới các WebSocket server.Lightweight, low latency, no storage: evicting caches on every instance, broadcasting driver locations to WebSocket servers.
Spring ApplicationEventPublisherSpring ApplicationEventPublisher
Pub-sub trong một app: publisher.publishEvent(new OrderPlaced(...)) và các @EventListener / @TransactionalEventListener nhận.In-app pub-sub: publisher.publishEvent(new OrderPlaced(...)) received by @EventListener / @TransactionalEventListener methods.
Thông báo đa kênhMulti-channel notifications
Sự kiện 'nạp tiền thành công' ở ví điện tử → push notification, SMS, email, cập nhật hạng thành viên, chống gian lận — mỗi bên một subscriber.A 'top-up succeeded' event in an e-wallet → push notification, SMS, email, membership tier update, fraud checks — one subscriber each.

Khi nào dùng và khi nào khôngWhen to use it, and when not

Nên dùng khiUse it when

Một sự kiện cần được NHIỀU bên độc lập phản ứng, và danh sách đó còn tăng.One event must trigger reactions in MANY independent parties, and that list keeps growing.
Publisher không nên chờ hay phụ thuộc vào sự thành công của các bên nhận.The publisher shouldn't wait for or depend on the receivers succeeding.
Muốn tách service để deploy, scale và lỗi độc lập với nhau.You want services to deploy, scale and fail independently.

Không nên dùng khiAvoid it when

Cần kết quả trả về ngay (kiểm tra tồn kho trước khi cho đặt) — dùng lời gọi đồng bộ request/response.You need an answer right away (check stock before allowing an order) — use a synchronous request/response call.
Mỗi việc chỉ được làm ĐÚNG MỘT lần bởi một worker — đó là Producer-Consumer/work queue, không phải pub-sub.Each job must be done by EXACTLY ONE worker — that's Producer-Consumer/a work queue, not pub-sub.
Hệ thống nhỏ, chỉ 1–2 bên nhận cố định — gọi hàm trực tiếp dễ đọc và dễ debug hơn.A small system with 1–2 fixed receivers — direct calls are easier to read and debug.

Ưu điểmPros

Tách rời hoàn toàn: thêm subscriber không sửa publisher.Full decoupling: adding a subscriber never edits the publisher.
Bất đồng bộ: subscriber chậm hoặc sập không kéo publisher theo.Asynchronous: a slow or crashed subscriber doesn't drag the publisher down.
Dễ mở rộng: mỗi subscriber scale theo tải riêng.Scales well: each subscriber scales with its own load.

Nhược điểmCons

Khó theo dõi luồng: 'ai đang nghe sự kiện này?' — cần tracing, tài liệu sự kiện.Harder to follow the flow: 'who listens to this event?' — you need tracing and event docs.
Phải xử lý trùng lặp (at-least-once), mất message (at-most-once), thứ tự, và nhất quán cuối cùng.You must handle duplicates (at-least-once), losses (at-most-once), ordering, and eventual consistency.
Broker trong bộ nhớ mất message khi crash; broker ngoài là thêm một hạ tầng phải vận hành.An in-memory broker loses messages on a crash; an external broker is one more piece of infra to run.

Anti-pattern: lỗi thường gặpAnti-pattern: common mistake

Publish chặn + subscriber bỏ đi không huỷ đăng kýBlocking publish + subscribers leaving without unsubscribing

Publish gửi chặn vào channel không buffer trong khi giữ khoá: chỉ cần một subscriber chậm là mọi publisher đứng hình. Tệ hơn, subscriber thoát mà không Unsubscribe thì không còn ai đọc channel đó — lần Publish kế tiếp treo vĩnh viễn (deadlock/goroutine leak).Publish does blocking sends on unbuffered channels while holding the lock: one slow subscriber freezes every publisher. Worse, a subscriber that exits without Unsubscribe leaves nobody reading its channel — the next Publish hangs forever (deadlock/goroutine leak).

func (b *Broker) Publish(topic string, payload any) {
	b.mu.Lock()
	defer b.mu.Unlock()
	for _, ch := range b.subs[topic] {
		ch <- Event{topic, payload} // ❌ blocking send under the lock
	}
}

func listenOnce(sub <-chan Event) {
	go func() {
		ev := <-sub
		handle(ev)
		return // ❌ quits without Unsubscribe → next Publish blocks forever
	}()
}

Dễ nhầm vớiEasily confused with

Publish-Subscribe vs Observer

Observer: subject giữ danh sách observer và gọi trực tiếp, thường đồng bộ, cùng process. Pub-sub: có broker/topic ở giữa, hai bên không biết nhau, thường bất đồng bộ và có thể khác process.Observer: the subject keeps its observer list and calls them directly, usually synchronously, in-process. Pub-sub: a broker/topic sits in between, the sides don't know each other, usually async and possibly cross-process.

Publish-Subscribe vs Producer-Consumer

Producer-consumer chia việc: mỗi item cho đúng một consumer. Pub-sub sao chép: mỗi subscriber nhận mọi message.Producer-consumer shares work: each item goes to exactly one consumer. Pub-sub copies: every subscriber gets every message.

Bài tập áp dụngExercise

Thêm Unsubscribe(topic string, ch <-chan Event) vào Broker Go: xoá channel khỏi danh sách và close nó, an toàn khi chạy đồng thời với Publish. Với Java, thêm một subscriber mới 'analytics-service' cho topic order.placed mà không sửa OrderService.Add Unsubscribe(topic string, ch <-chan Event) to the Go Broker: remove the channel from the list and close it, safely concurrent with Publish. In Java, add a new 'analytics-service' subscriber for order.placed without touching OrderService.

Kiểm tra kiến thứcKnowledge check

Câu 1 / 6Question 1 of 6

Khác biệt cốt lõi giữa pub-sub và producer-consumer (work queue)?The core difference between pub-sub and producer-consumer (work queue)?

Phím 1–4 để chọn đáp án, Enter để kiểm tra.Press 1–4 to choose an answer, Enter to check.

Thẻ ghi nhớFlashcards

Ôn tập ngắt quãng (spaced repetition): thẻ được lặp lại sau 1 → 3 → 7 → 14 → 30 ngày để củng cố trí nhớ dài hạn.Spaced repetition: cards are repeated after 1 → 3 → 7 → 14 → 30 days to build long-term memory.

Dùng trong bài toánUsed in problems

5 bài toán Backend dùng pattern này5 backend problems use this pattern

Pattern liên quanRelated patterns

Observer Producer-Consumer Transactional Outbox Fan-out / Fan-in

Bài trướcPrevious

Read-Write Lock

Thảo luậnDiscussion

Publish-Subscribeaka Pub/Sub, Topic-based messaging, Event Bus

VừaMedium

Mức phổ biếnUsage 5/5

~9 phút đọc~9 min read

package com.shop.order; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.stereotype.Service; record OrderPlaced(String orderId, String userId, long total) {}1 @Service public class OrderService { static final String TOPIC = "order.placed"; private static final Logger log = LoggerFactory.getLogger(OrderService.class); private final KafkaTemplate<String, OrderPlaced> kafka; private final OrderRepository orders; public OrderService(KafkaTemplate<String, OrderPlaced> kafka, OrderRepository orders) { this.kafka = kafka; this.orders = orders; } public void place(Order order) { orders.save(order);2 var event = new OrderPlaced(order.id(), order.userId(), order.total()); // Key = orderId → all events of one order go to the same partition (ordered) kafka.send(TOPIC, order.id(), event)3 .whenComplete((result, ex) -> { if (ex != null) { log.error("publish failed for order {}", order.id(), ex); } }); } }

package pubsub import "sync" type Event struct { Topic string Payload any } type Broker struct { mu sync.RWMutex subs map[string][]chan Event1 } func NewBroker() *Broker { return &Broker{subs: make(map[string][]chan Event)} } // Subscribe returns a dedicated buffered channel for this subscriber. func (b *Broker) Subscribe(topic string, buf int) <-chan Event {2 ch := make(chan Event, buf) b.mu.Lock() b.subs[topic] = append(b.subs[topic], ch) b.mu.Unlock() return ch } // Publish never blocks: a slow subscriber drops events instead of stalling everyone. func (b *Broker) Publish(topic string, payload any) { b.mu.RLock() defer b.mu.RUnlock() for _, ch := range b.subs[topic] { select {3 case ch <- Event{Topic: topic, Payload: payload}: default: // buffer full → drop (count a metric in real code) } } } // Close closes every subscriber channel so their range loops end. func (b *Broker) Close() { b.mu.Lock()4 defer b.mu.Unlock() for topic, chans := range b.subs { for _, ch := range chans { close(ch) } delete(b.subs, topic) } }

func (b *Broker) Publish(topic string, payload any) { b.mu.Lock() defer b.mu.Unlock() for _, ch := range b.subs[topic] { ch <- Event{topic, payload} // ❌ blocking send under the lock } } func listenOnce(sub <-chan Event) { go func() { ev := <-sub handle(ev) return // ❌ quits without Unsubscribe → next Publish blocks forever }() }