Đồng thờiConcurrency

Fan-out / Fan-inaka Scatter-Gather, Split-Merge

Chia việc ra nhiều goroutine/thread chạy song song (fan-out), rồi gộp kết quả về một mối (fan-in).Spread work across many goroutines/threads in parallel (fan-out), then merge the results into one stream (fan-in).

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

Thảo luậnDiscussion

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

Cuối tháng, kế toán trưởng nhờ 3 chi nhánh Hà Nội, Đà Nẵng, TP.HCM cùng lúc gửi báo cáo doanh thu (fan-out). Các bản báo cáo về qua một hộp thư chung, đến trước sau lộn xộn, và chỉ một người ngồi cộng sổ (fan-in). Nhanh gấp ba so với gọi từng chi nhánh, và không ai tranh nhau ghi vào cùng một cuốn sổ.At month end, the chief accountant asks the Hanoi, Da Nang and HCMC branches to send their revenue reports at the same time (fan-out). The reports land in one shared inbox in random order, and a single person adds them up (fan-in). Three times faster than calling branches one by one, and nobody fights over the same ledger.

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

Không có patternWithout the pattern

DB đơn hàng của sàn thương mại điện tử được chia 3 shard theo vùng. Báo cáo doanh thu ngày phải query cả 3 shard, mỗi shard mất ~1 giây. Query tuần tự mất 3 giây; còn nếu cho mỗi goroutine tự cộng vào một map dùng chung thì dính data race và Go có thể crash với 'concurrent map writes'.An e-commerce marketplace splits its order DB into 3 regional shards. The daily revenue report must query all 3, each taking ~1 second. Querying sequentially takes 3 seconds; letting each goroutine add into a shared map causes a data race, and Go may crash with 'concurrent map writes'.

Với Fan-out / Fan-inWith Fan-out / Fan-in

Fan-out: khởi động một goroutine cho mỗi shard, mỗi goroutine trả kết quả qua channel riêng. Fan-in: hàm Merge chuyển tất cả channel về một channel duy nhất và tự close khi mọi nguồn đã cạn (WaitGroup). Chỉ một bên đọc channel gộp để cộng tổng → không cần khoá, không race, thời gian ≈ shard chậm nhất.Fan-out: start one goroutine per shard, each returning results on its own channel. Fan-in: a Merge function funnels every channel into a single one and closes it once all sources are drained (WaitGroup). A single reader of the merged channel does the summing → no locks, no races, time ≈ the slowest shard.

Thành phần tham giaParticipants

Splitter: Chia việc và khởi động các nhánh song song (fan-out).Splits the work and starts parallel branches (fan-out).
Branch worker: Xử lý một phần việc và gửi kết quả ra channel riêng.Processes one part and sends results on its own channel.
Merger: Gộp nhiều channel về một (fan-in) và close khi tất cả nguồn đã xong.Merges many channels into one (fan-in) and closes it when every source is done.
Aggregator: Bên đọc duy nhất của channel gộp; tổng hợp kết quả cuối mà không cần khoá.The single reader of the merged channel; builds the final result without locks.

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.

1Fan-out: DailyReport khởi động một goroutine cho mỗi shard. Cả 3 query chạy cùng lúc.Fan-out: DailyReport starts one goroutine per shard. All 3 queries run at the same time.

← →

console

1fan-out: 3 shards

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.report;

import org.springframework.stereotype.Service;
import java.time.LocalDate;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

record Revenue(String shopId, long amount) {}

@Service
public class ShardReportService {
    private final List<ShardClient> shards;    // one bean per regional shard1
    private final ExecutorService ioPool;

    public ShardReportService(List<ShardClient> shards, ExecutorService ioPool) {
        this.shards = shards;
        this.ioPool = ioPool;
    }

    public Map<String, Long> dailyReport(LocalDate day)
            throws InterruptedException, ExecutionException {
        List<Callable<List<Revenue>>> tasks = shards.stream()
                .<Callable<List<Revenue>>>map(s -> () -> s.dailyRevenue(day))2
                .toList();

        Map<String, Long> totals = new HashMap<>();
        for (Future<List<Revenue>> f : ioPool.invokeAll(tasks, 5, TimeUnit.SECONDS)) {3
            for (Revenue r : f.get()) {
                totals.merge(r.shopId(), r.amount(), Long::sum);4
            }
        }
        return totals;
    }
}

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

package report

import (
	"context"
	"sync"
)

// Merge fans in many channels into one.
// The output is closed once every input is drained.
func Merge[T any](ctx context.Context, ins ...<-chan T) <-chan T {1
	out := make(chan T)
	var wg sync.WaitGroup
	wg.Add(len(ins))2
	for _, in := range ins {3
		go func() {
			defer wg.Done()
			for v := range in {
				select {
				case out <- v:
				case <-ctx.Done():
					return
				}
			}
		}()
	}
	go func() {
		wg.Wait()
		close(out)4
	}()
	return out
}

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

01Java có sẵn fan-out + chờ: invokeAll, CompletableFuture.allOf (và StructuredTaskScope đang ở dạng preview). Go tự viết Merge bằng channel + WaitGroup, hoặc dùng errgroup.Java ships fan-out + wait: invokeAll, CompletableFuture.allOf (and StructuredTaskScope in preview). Go writes Merge with channels + a WaitGroup, or uses errgroup.
02Java thường gom kết quả SAU khi tất cả xong (List<Future>). Go stream kết quả ngay khi có — aggregator bắt đầu cộng từ lúc shard nhanh nhất trả về.Java usually gathers results AFTER all are done (List<Future>). Go streams results as they arrive — the aggregator starts summing as soon as the fastest shard replies.
03Cả hai đều tránh data race bằng cách để MỘT bên ghi kết quả: thread gọi invokeAll trong Java, goroutine đọc channel gộp trong Go.Both avoid data races by having ONE party write the result: the invokeAll caller thread in Java, the merged-channel reader in Go.
04Fan-out trong Go còn có dạng thứ hai: nhiều goroutine cùng đọc MỘT channel đầu vào (chia việc). Java tương ứng là nhiều thread cùng take() từ một BlockingQueue.Go has a second fan-out shape: many goroutines reading ONE input channel (work sharing). Java's equivalent is several threads take()-ing from one BlockingQueue.

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

Scatter-gather trên DB phân mảnhScatter-gather over sharded databases
Elasticsearch gửi truy vấn tới mọi shard rồi gộp top-k; Vitess/Citus làm tương tự cho SQL.Elasticsearch sends a query to every shard and merges the top-k; Vitess/Citus do the same for SQL.
Spring Cloud Gateway / BFFSpring Cloud Gateway / BFF
Một API gateway gọi song song nhiều microservice rồi ghép một response cho app mobile.An API gateway calls several microservices in parallel and builds one response for the mobile app.
Blog 'Go Concurrency Patterns: Pipelines'The 'Go Concurrency Patterns: Pipelines' blog
Hàm merge với WaitGroup trong bài viết chính thức của Go là bản mẫu được chép khắp nơi; errgroup gói gọn phần xử lý lỗi.The merge function with a WaitGroup in the official Go blog post is copied everywhere; errgroup wraps up the error handling.
Tìm vé máy bay, so sánh giáFlight search, price comparison
Hỏi giá cùng lúc từ Vietnam Airlines, Vietjet, Bamboo và gộp danh sách chuyến cho khách.Ask Vietnam Airlines, Vietjet and Bamboo for fares at once and merge the flight list for the customer.

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 việc lớn chia được thành nhiều phần độc lập (theo shard, theo vùng, theo nhà cung cấp).A big task splits into independent parts (by shard, region or provider).
Một công đoạn trong pipeline là nút cổ chai và cần chạy nhiều bản song song.One pipeline stage is a bottleneck and needs several parallel copies.
Cần gộp kết quả về một chỗ để tổng hợp mà không muốn dùng khoá.You need to gather results in one place for aggregation without using locks.

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

Các phần phụ thuộc nhau hoặc phải giữ đúng thứ tự tuyệt đối — fan-in làm xáo trộn thứ tự.Parts depend on each other or strict ordering is required — fan-in shuffles the order.
Số phần quá lớn (hàng triệu) mà mỗi phần một goroutine/thread — hãy dùng Worker Pool để giới hạn.Huge numbers of parts (millions) with one goroutine/thread each — use a Worker Pool to bound it.

Ưu điểmPros

Thời gian ≈ nhánh chậm nhất thay vì tổng các nhánh.Time ≈ the slowest branch instead of the sum of all branches.
Tổng hợp an toàn với một bên đọc duy nhất, không cần mutex.Safe aggregation with a single reader — no mutex needed.
Dễ kết hợp với Pipeline: fan-out riêng công đoạn chậm.Composes nicely with Pipeline: fan out only the slow stage.

Nhược điểmCons

Thứ tự kết quả không còn đảm bảo.Result order is no longer guaranteed.
Xử lý lỗi và huỷ phức tạp hơn: một nhánh lỗi thì các nhánh khác dừng hay chạy tiếp? Quên ctx → goroutine leak.Error handling and cancellation get harder: when one branch fails, do the others stop? Forget ctx → goroutine leak.
Close channel gộp sai thời điểm → panic; không ai đọc channel gộp → deadlock.Closing the merged channel at the wrong time → panic; nobody reading it → deadlock.

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

Fan-out nhưng 'fan-in' bằng map dùng chungFanning out but 'fanning in' through a shared map

Mỗi goroutine tự cộng vào cùng một map. Đây là data race: Go có thể dừng chương trình với 'fatal error: concurrent map writes', và go test -race sẽ báo ngay. Hãy gộp qua channel về một bên đọc duy nhất (hoặc ít nhất dùng sync.Mutex).Each goroutine adds into the same map. That's a data race: Go may abort with 'fatal error: concurrent map writes', and go test -race flags it immediately. Funnel results through a channel to a single reader (or at least use a sync.Mutex).

totals := make(map[string]int64)
var wg sync.WaitGroup
for _, s := range shards {
	wg.Add(1)
	go func() {
		defer wg.Done()
		rows, _ := s.DailyRevenue(ctx, day) // ❌ error ignored
		for _, r := range rows {
			totals[r.ShopID] += r.Amount // ❌ concurrent map writes
		}
	}()
}
wg.Wait()

Dễ nhầm vớiEasily confused with

Fan-out / Fan-in vs Worker Pool

Worker Pool: N worker cố định rút việc từ một hàng đợi, nhấn mạnh việc giới hạn. Fan-out/Fan-in: toả việc ra nhiều nhánh (thường theo nguồn dữ liệu) và nhấn mạnh việc GỘP kết quả về một mối.Worker Pool: a fixed N workers pulling from one queue, emphasising the limit. Fan-out/Fan-in: spread work into branches (often per data source), emphasising MERGING results into one stream.

Fan-out / Fan-in vs Publish-Subscribe

Pub-Sub gửi CÙNG một message tới mọi subscriber. Fan-out ở đây chia các PHẦN VIỆC KHÁC NHAU cho các nhánh.Pub-Sub sends the SAME message to every subscriber. Fan-out here hands DIFFERENT pieces of work to different branches.

Bài tập áp dụngExercise

Viết hàm tìm chuyến bay từ 3 hãng song song. Gộp tất cả chuyến về một danh sách, sắp xếp theo giá; hãng nào lỗi thì bỏ qua nhưng ghi log; toàn bộ không quá 2 giây.Write a function that searches flights from 3 airlines in parallel. Merge all flights into one list sorted by price; skip (but log) any airline that fails; the whole thing must finish within 2 seconds.

Kiểm tra kiến thứcKnowledge check

Câu 1 / 6Question 1 of 6

Trong hàm Merge, khi nào được phép close(out)?In Merge, when may out be closed?

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

3 bài toán Backend dùng pattern này3 backend problems use this pattern

Pattern liên quanRelated patterns

Pipeline Worker Pool Future / Promise Cancellation / Context

Bài trướcPrevious

Pipeline

Bài tiếpNext

Cancellation / Context

Thảo luậnDiscussion

Fan-out / Fan-inaka Scatter-Gather, Split-Merge

KhóHard

Mức phổ biếnUsage 4/5

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

package com.shop.report; import org.springframework.stereotype.Service; import java.time.LocalDate; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.*; record Revenue(String shopId, long amount) {} @Service public class ShardReportService { private final List<ShardClient> shards; // one bean per regional shard1 private final ExecutorService ioPool; public ShardReportService(List<ShardClient> shards, ExecutorService ioPool) { this.shards = shards; this.ioPool = ioPool; } public Map<String, Long> dailyReport(LocalDate day) throws InterruptedException, ExecutionException { List<Callable<List<Revenue>>> tasks = shards.stream() .<Callable<List<Revenue>>>map(s -> () -> s.dailyRevenue(day))2 .toList(); Map<String, Long> totals = new HashMap<>(); for (Future<List<Revenue>> f : ioPool.invokeAll(tasks, 5, TimeUnit.SECONDS)) {3 for (Revenue r : f.get()) { totals.merge(r.shopId(), r.amount(), Long::sum);4 } } return totals; } }

package report import ( "context" "sync" ) // Merge fans in many channels into one. // The output is closed once every input is drained. func Merge[T any](ctx context.Context, ins ...<-chan T) <-chan T {1 out := make(chan T) var wg sync.WaitGroup wg.Add(len(ins))2 for _, in := range ins {3 go func() { defer wg.Done() for v := range in { select { case out <- v: case <-ctx.Done(): return } } }() } go func() { wg.Wait() close(out)4 }() return out }

totals := make(map[string]int64) var wg sync.WaitGroup for _, s := range shards { wg.Add(1) go func() { defer wg.Done() rows, _ := s.DailyRevenue(ctx, day) // ❌ error ignored for _, r := range rows { totals[r.ShopID] += r.Amount // ❌ concurrent map writes } }() } wg.Wait()