Đồng thờiConcurrency

Pipelineaka Pipes and Filters, Stream Processing

Chia việc xử lý thành chuỗi các công đoạn nối nhau, mỗi công đoạn chạy song song với công đoạn khác.Split processing into a chain of connected stages, each running concurrently with the others.

Độ khóDifficulty: VừaMedium
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

Dây chuyền làm bánh mì: người thứ nhất xẻ bánh, người thứ hai phết pate, người thứ ba cho thịt và rau, người cuối gói giấy. Khi người cuối đang gói ổ số 1 thì người đầu đã xẻ tới ổ số 4. Không ai phải làm hết mọi việc, và cả dây chuyền chạy cùng lúc.A bánh mì assembly line: the first person slices the bread, the second spreads pâté, the third adds meat and herbs, the last wraps it. While the last person wraps sandwich #1, the first is already slicing #4. Nobody does everything, and the whole line works at once.

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

Không có patternWithout the pattern

Ngân hàng cần nhập file sao kê 2 triệu dòng: đọc dòng → parse → quy đổi ngoại tệ sang VND → ghi DB. Code kiểu 'đọc hết file vào List, rồi parse hết, rồi quy đổi hết' tốn vài GB RAM và CPU, mạng, đĩa thay nhau ngồi chơi: lúc đọc file thì DB rảnh, lúc ghi DB thì CPU rảnh.A bank must import a 2-million-line statement file: read line → parse → convert foreign currency to VND → write to DB. Code that 'reads the whole file into a List, then parses everything, then converts everything' eats gigabytes of RAM, and CPU, network and disk take turns idling: while reading the file the DB sits idle, while writing the DB the CPU sits idle.

Với PipelineWith Pipeline

Tách thành các công đoạn nhỏ, mỗi công đoạn nhận dữ liệu từ đầu vào, xử lý TỪNG phần tử và đẩy sang công đoạn sau. Các công đoạn nối với nhau bằng channel (Go) hoặc stream/Flux (Java). Dữ liệu chảy từng phần tử một nên RAM ổn định, các công đoạn chạy chồng lên nhau, và công đoạn chậm tự làm chậm các công đoạn phía trước (backpressure).Split the work into small stages; each takes input, processes ONE item at a time and pushes it to the next stage. Stages are connected by channels (Go) or streams/Flux (Java). Data flows item by item so memory stays flat, stages overlap in time, and a slow stage automatically slows the stages before it (backpressure).

Thành phần tham giaParticipants

Source: Sinh dữ liệu ban đầu (đọc file, đọc Kafka, query DB) và báo hết dữ liệu.Produces the initial data (file, Kafka, DB query) and signals the end.
Stage: Nhận từ đầu vào, biến đổi/lọc một phần tử, gửi ra đầu ra; đóng đầu ra khi đầu vào hết.Receives from its input, transforms/filters one item, sends to its output; closes its output when the input ends.
Channel / Stream: Đường ống nối hai công đoạn liền kề.The pipe connecting two adjacent stages.
Sink: Công đoạn cuối tiêu thụ kết quả (ghi DB, gửi API) và 'kéo' cả dây chuyền chạy.The final stage consuming results (write DB, call an API) and 'pulling' the whole line.

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.

1Source đọc dòng 1 của file sao kê và gửi vào lines. Chỉ một dòng nằm trong bộ nhớ, không phải 2 triệu dòng.The source reads line 1 of the statement and sends it into lines. Only one line is in memory, not 2 million.

← →

console

1read: FT2401,120,USD

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.bank.importer;

import org.springframework.stereotype.Service;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Optional;
import java.util.stream.Stream;

record Txn(String ref, long amount, String currency, long vnd) {
    static Optional<Txn> parse(String line) {1
        String[] f = line.split(",");
        if (f.length != 3) return Optional.empty();
        try {
            return Optional.of(new Txn(f[0], Long.parseLong(f[1]), f[2], 0));
        } catch (NumberFormatException e) {
            return Optional.empty();
        }
    }

    Txn withVnd(long rate) { return new Txn(ref, amount, currency, amount * rate); }
}

@Service
public class StatementImporter {
    private final FxRates fx;
    private final TxnRepository repo;

    public StatementImporter(FxRates fx, TxnRepository repo) {
        this.fx = fx;
        this.repo = repo;
    }

    public void importFile(Path file) throws IOException {
        try (Stream<String> lines = Files.lines(file)) {         // source (lazy)2
            lines.map(Txn::parse).flatMap(Optional::stream)       // stage 1: parse3
                 .filter(t -> fx.supports(t.currency()))          // stage 2: validate
                 .map(t -> t.withVnd(fx.rate(t.currency())))      // stage 3: enrich
                 .forEach(repo::save);                            // sink4
        }
    }
}

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

package importer

import (
	"bufio"
	"context"
	"io"
)

// Source emits one line at a time and closes out at EOF or on cancel.
func Source(ctx context.Context, r io.Reader) <-chan string {
	out := make(chan string)
	go func() {
		defer close(out)1
		sc := bufio.NewScanner(r)
		for sc.Scan() {
			select {
			case out <- sc.Text():
			case <-ctx.Done():2
				return
			}
		}
	}()
	return out
}

// Stage turns a per-item function into a pipeline stage.
// fn returns ok=false to drop an item.
func Stage[In, Out any](ctx context.Context, in <-chan In, fn func(In) (Out, bool)) <-chan Out {3
	out := make(chan Out)
	go func() {
		defer close(out)
		for v := range in {4
			res, ok := fn(v)
			if !ok {
				continue
			}
			select {
			case out <- res:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

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

01Go: mỗi công đoạn là một goroutine thật, chạy song song, nối bằng channel. Java Stream: các công đoạn chỉ là phép biến đổi lười chạy trên MỘT thread (trừ khi .parallel()).Go: each stage is a real goroutine running concurrently, linked by channels. Java Streams: stages are lazy transforms running on ONE thread (unless .parallel()).
02Báo hết dữ liệu: Go close channel lan dần qua từng công đoạn. Java Stream kết thúc khi nguồn hết; Reactor phát tín hiệu onComplete/onError xuống dưới.End of data: Go's close cascades stage by stage. A Java Stream ends when its source is exhausted; Reactor sends onComplete/onError downstream.
03Backpressure: channel không buffer/buffer nhỏ chặn bên gửi một cách tự nhiên. Reactor dùng giao thức request(n) của Reactive Streams.Backpressure: unbuffered/small channels naturally block the sender. Reactor uses the Reactive Streams request(n) protocol.
04Huỷ giữa chừng: Go truyền ctx vào mọi công đoạn và select trên ctx.Done(). Reactor huỷ subscription và Flux.using tự dọn tài nguyên.Cancelling midway: Go passes ctx to every stage and selects on ctx.Done(). Reactor cancels the subscription and Flux.using cleans up resources.

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

Java Stream APIJava Stream API
list.stream().filter().map().collect() là pipeline lười trong bộ nhớ — mỗi phần tử đi qua hết các công đoạn rồi mới tới phần tử sau.list.stream().filter().map().collect() is a lazy in-memory pipeline — each element passes through every stage before the next one.
Unix pipe và io.PipeUnix pipes and io.Pipe
cat log | grep ERROR | sort | uniq -c — mỗi lệnh là một công đoạn chạy song song. Go có io.Pipe, io.MultiWriter, gzip.Writer để xếp chồng stream.cat log | grep ERROR | sort | uniq -c — each command is a concurrently running stage. Go has io.Pipe, io.MultiWriter, gzip.Writer to stack streams.
Spring Batch, Reactor, Kafka StreamsSpring Batch, Reactor, Kafka Streams
Spring Batch: ItemReader → ItemProcessor → ItemWriter theo chunk. Reactor/WebFlux và Kafka Streams là pipeline có backpressure.Spring Batch: ItemReader → ItemProcessor → ItemWriter in chunks. Reactor/WebFlux and Kafka Streams are pipelines with backpressure.
ETL và xử lý ảnh/videoETL and image/video processing
Nhập sao kê ngân hàng, đồng bộ đơn hàng sang kho, tải ảnh → resize → watermark → upload S3.Importing bank statements, syncing orders to a warehouse, download image → resize → watermark → upload to S3.

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

Nên dùng khiUse it when

Dữ liệu lớn hoặc vô hạn cần xử lý qua nhiều bước nối tiếp.Large or unbounded data must go through several sequential steps.
Các bước dùng tài nguyên khác nhau (đĩa, CPU, mạng, DB) và có thể chạy chồng lên nhau.Steps use different resources (disk, CPU, network, DB) and can overlap.
Muốn RAM ổn định và các bước dễ tái sử dụng, dễ test riêng.You want flat memory use and reusable, separately testable steps.

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

Dữ liệu nhỏ và xử lý nhanh — một vòng for đơn giản dễ đọc hơn, goroutine/channel chỉ thêm overhead.Small, fast data — a plain for loop is clearer; goroutines/channels only add overhead.
Các bước cần nhìn toàn bộ dữ liệu cùng lúc (sort toàn cục, join phức tạp) — pipeline không giúp gì nhiều.Steps need the whole dataset at once (global sort, complex joins) — a pipeline doesn't help much.

Ưu điểmPros

Các công đoạn chạy chồng nhau → tận dụng CPU, mạng, đĩa cùng lúc.Stages overlap → CPU, network and disk are used at the same time.
Bộ nhớ ổn định nhờ xử lý từng phần tử; có backpressure tự nhiên.Flat memory by processing item by item; natural backpressure.
Mỗi công đoạn nhỏ, tái sử dụng được, thêm/bớt công đoạn dễ dàng.Small, reusable stages; easy to add or remove one.

Nhược điểmCons

Thông lượng bị giới hạn bởi công đoạn chậm nhất (nút cổ chai) — cần fan-out ở công đoạn đó.Throughput is capped by the slowest stage (bottleneck) — you need fan-out at that stage.
Xử lý lỗi và huỷ phức tạp: quên ctx hoặc quên close là goroutine leak hoặc deadlock.Error handling and cancellation are tricky: forgetting ctx or a close means goroutine leaks or deadlocks.
Gửi con trỏ/slice qua channel rồi vẫn sửa nó ở công đoạn trước → data race.Sending a pointer/slice through a channel and still mutating it upstream → data race.

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

Công đoạn không lắng nghe ctx — goroutine leak khi sink bỏ cuộcStages ignoring ctx — goroutine leaks when the sink gives up

Sink gặp lỗi ở dòng 10 và return. Không ai đọc channel nữa, nhưng công đoạn phía trước vẫn chặn mãi ở out <- ... vì không có select với ctx.Done(). Mỗi lần import lỗi lại rò thêm vài goroutine giữ file đang mở.The sink fails on row 10 and returns. Nobody reads the channel any more, but the upstream stage blocks forever on out <- ... because there's no select on ctx.Done(). Every failed import leaks a few goroutines holding open files.

func parseStage(in <-chan string) <-chan Txn {
	out := make(chan Txn)
	go func() {
		defer close(out)
		for line := range in {
			t, _ := parse(line)
			out <- t // ❌ blocks forever once the sink has returned
		}
	}()
	return out
}

Dễ nhầm vớiEasily confused with

Pipeline vs Chain of Responsibility

Chain of Responsibility: một request đi qua chuỗi handler và có thể dừng ở handler nào đó, chạy tuần tự. Pipeline: MỌI phần tử đi qua MỌI công đoạn, các công đoạn chạy đồng thời trên một dòng dữ liệu.Chain of Responsibility: one request travels a chain of handlers and may stop at one, sequentially. Pipeline: EVERY item passes EVERY stage, and stages run concurrently over a stream.

Pipeline vs Producer-Consumer

Producer-Consumer là một khớp nối duy nhất giữa hai bên. Pipeline là nhiều khớp Producer-Consumer nối tiếp: mỗi công đoạn vừa là consumer của công đoạn trước vừa là producer cho công đoạn sau.Producer-Consumer is a single link between two sides. A Pipeline chains many such links: each stage consumes from the previous one and produces for the next.

Bài tập áp dụngExercise

Công đoạn quy đổi tỷ giá gọi API mất 50ms/dòng và trở thành nút cổ chai. Hãy làm nó nhanh lên ~8 lần mà không đổi các công đoạn khác.The FX stage calls an API taking 50ms per row and has become the bottleneck. Make it ~8× faster without changing the other stages.

Kiểm tra kiến thứcKnowledge check

Câu 1 / 5Question 1 of 5

Trong pipeline Go, ai chịu trách nhiệm close channel đầu ra của một công đoạn?In a Go pipeline, who is responsible for closing a stage's output channel?

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

4 bài toán Backend dùng pattern này4 backend problems use this pattern

Pattern liên quanRelated patterns

Fan-out / Fan-in Producer-Consumer Chain of Responsibility Cancellation / Context

Thảo luậnDiscussion

Pipelineaka Pipes and Filters, Stream Processing

VừaMedium

Mức phổ biếnUsage 4/5

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

package com.bank.importer; import org.springframework.stereotype.Service; import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; import java.util.Optional; import java.util.stream.Stream; record Txn(String ref, long amount, String currency, long vnd) { static Optional<Txn> parse(String line) {1 String[] f = line.split(","); if (f.length != 3) return Optional.empty(); try { return Optional.of(new Txn(f[0], Long.parseLong(f[1]), f[2], 0)); } catch (NumberFormatException e) { return Optional.empty(); } } Txn withVnd(long rate) { return new Txn(ref, amount, currency, amount * rate); } } @Service public class StatementImporter { private final FxRates fx; private final TxnRepository repo; public StatementImporter(FxRates fx, TxnRepository repo) { this.fx = fx; this.repo = repo; } public void importFile(Path file) throws IOException { try (Stream<String> lines = Files.lines(file)) { // source (lazy)2 lines.map(Txn::parse).flatMap(Optional::stream) // stage 1: parse3 .filter(t -> fx.supports(t.currency())) // stage 2: validate .map(t -> t.withVnd(fx.rate(t.currency()))) // stage 3: enrich .forEach(repo::save); // sink4 } } }

package importer import ( "bufio" "context" "io" ) // Source emits one line at a time and closes out at EOF or on cancel. func Source(ctx context.Context, r io.Reader) <-chan string { out := make(chan string) go func() { defer close(out)1 sc := bufio.NewScanner(r) for sc.Scan() { select { case out <- sc.Text(): case <-ctx.Done():2 return } } }() return out } // Stage turns a per-item function into a pipeline stage. // fn returns ok=false to drop an item. func Stage[In, Out any](ctx context.Context, in <-chan In, fn func(In) (Out, bool)) <-chan Out {3 out := make(chan Out) go func() { defer close(out) for v := range in {4 res, ok := fn(v) if !ok { continue } select { case out <- res: case <-ctx.Done(): return } } }() return out }

func parseStage(in <-chan string) <-chan Txn { out := make(chan Txn) go func() { defer close(out) for line := range in { t, _ := parse(line) out <- t // ❌ blocks forever once the sink has returned } }() return out }