1. 上下文包
情境 是 Go 中用於並發控制的最重要的套件。情境承載 截止日期, 取消訊號 和 請求範圍的值 透過函數邊界和 goroutine。
1.1.上下文基礎知識
import "context"
// context.Background() - root context, dùng trong main/init/tests
ctx := context.Background()
// context.TODO() - placeholder khi chưa biết dùng context nào
ctx := context.TODO()
// ⚠️ Quy tắc: context LUÔN là parameter đầu tiên
func GetUser(ctx context.Context, id int) (*User, error) {
// ...
}
// KHÔNG lưu context trong struct
// ❌ type Server struct { ctx context.Context }
// ✅ Truyền context qua function parameter
1.2.上下文.WithCancel
func main() {
ctx, cancel := context.WithCancel(context.Background())
defer cancel() // ⚠️ LUÔN gọi cancel để tránh resource leak
go func(ctx context.Context) {
for {
select {
case <-ctx.Done():
fmt.Println("Worker cancelled:", ctx.Err())
return
default:
fmt.Println("Working...")
time.Sleep(500 * time.Millisecond)
}
}
}(ctx)
time.Sleep(2 * time.Second)
cancel() // Signal cancellation
time.Sleep(100 * time.Millisecond)
}
1.3. context.WithTimeout 和 WithDeadline
// WithTimeout: cancel sau duration
func fetchData(ctx context.Context) (string, error) {
ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
defer cancel()
ch := make(chan string, 1)
go func() {
// Simulate slow operation
time.Sleep(5 * time.Second)
ch <- "data"
}()
select {
case data := <-ch:
return data, nil
case <-ctx.Done():
return "", fmt.Errorf("fetch timed out: %w", ctx.Err())
}
}
// WithDeadline: cancel tại thời điểm cụ thể
func processUntil(ctx context.Context) {
deadline := time.Now().Add(10 * time.Second)
ctx, cancel := context.WithDeadline(ctx, deadline)
defer cancel()
if d, ok := ctx.Deadline(); ok {
fmt.Println("Deadline:", d)
}
// ... process
}
func main() {
ctx := context.Background()
data, err := fetchData(ctx)
if err != nil {
fmt.Println("Error:", err) // "fetch timed out: context deadline exceeded"
return
}
fmt.Println(data)
}
1.4.上下文.WithValue
// WithValue: gắn request-scoped data vào context
// ⚠️ Chỉ dùng cho cross-cutting concerns: request ID, auth info, trace ID
// ❌ KHÔNG dùng cho function parameters
type contextKey string
const (
requestIDKey contextKey = "request_id"
userIDKey contextKey = "user_id"
)
func withRequestID(ctx context.Context, requestID string) context.Context {
return context.WithValue(ctx, requestIDKey, requestID)
}
func getRequestID(ctx context.Context) string {
if id, ok := ctx.Value(requestIDKey).(string); ok {
return id
}
return ""
}
// Middleware pattern
func loggingMiddleware(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
requestID := uuid.New().String()
ctx := withRequestID(r.Context(), requestID)
log.Printf("[%s] %s %s", requestID, r.Method, r.URL.Path)
next.ServeHTTP(w, r.WithContext(ctx))
})
}
func handler(w http.ResponseWriter, r *http.Request) {
requestID := getRequestID(r.Context())
fmt.Fprintf(w, "Request ID: %s", requestID)
}
1.5.上下文傳播
// Context lan truyền qua call chain
func handleRequest(ctx context.Context) error {
// Thêm timeout cho toàn bộ request
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
// Gọi service layer
user, err := userService.GetUser(ctx, 123)
if err != nil {
return err
}
// Gọi nhiều services concurrent
g, ctx := errgroup.WithContext(ctx)
var orders []Order
g.Go(func() error {
var err error
orders, err = orderService.GetOrders(ctx, user.ID)
return err
})
var notifications []Notification
g.Go(func() error {
var err error
notifications, err = notifService.GetNotifications(ctx, user.ID)
return err
})
if err := g.Wait(); err != nil {
return err // Nếu 1 goroutine fail, context tự cancel các goroutine khác
}
// ... use orders, notifications
return nil
}
2. 同步包
2.1.同步互斥鎖
// Mutex bảo vệ shared data khỏi race conditions
type SafeCounter struct {
mu sync.Mutex
count map[string]int
}
func NewSafeCounter() *SafeCounter {
return &SafeCounter{
count: make(map[string]int),
}
}
func (c *SafeCounter) Increment(key string) {
c.mu.Lock()
defer c.mu.Unlock()
c.count[key]++
}
func (c *SafeCounter) Get(key string) int {
c.mu.Lock()
defer c.mu.Unlock()
return c.count[key]
}
func main() {
counter := NewSafeCounter()
var wg sync.WaitGroup
for i := 0; i < 1000; i++ {
wg.Add(1)
go func() {
defer wg.Done()
counter.Increment("visits")
}()
}
wg.Wait()
fmt.Println(counter.Get("visits")) // 1000 (always correct)
}
// Detect race conditions:
// go run -race main.go
// go test -race ./...
2.2.同步RWMutex
// RWMutex: nhiều readers đồng thời, nhưng writer exclusive
type Cache struct {
mu sync.RWMutex
data map[string]string
}
func NewCache() *Cache {
return &Cache{data: make(map[string]string)}
}
func (c *Cache) Get(key string) (string, bool) {
c.mu.RLock() // Read lock - nhiều goroutines đọc cùng lúc
defer c.mu.RUnlock()
val, ok := c.data[key]
return val, ok
}
func (c *Cache) Set(key, value string) {
c.mu.Lock() // Write lock - exclusive
defer c.mu.Unlock()
c.data[key] = value
}
func (c *Cache) Delete(key string) {
c.mu.Lock()
defer c.mu.Unlock()
delete(c.data, key)
}
2.3.同步一次
// sync.Once đảm bảo function chỉ chạy 1 lần (thread-safe singleton)
var (
instance *Database
once sync.Once
)
type Database struct {
conn string
}
func GetDatabase() *Database {
once.Do(func() {
fmt.Println("Initializing database...")
instance = &Database{conn: "postgres://localhost/mydb"}
})
return instance
}
func main() {
// Gọi từ nhiều goroutines - chỉ init 1 lần
var wg sync.WaitGroup
for i := 0; i < 10; i++ {
wg.Add(1)
go func() {
defer wg.Done()
db := GetDatabase()
fmt.Println(db.conn)
}()
}
wg.Wait()
// "Initializing database..." chỉ in 1 lần
}
2.4.同步池
// sync.Pool: tái sử dụng objects tạm thời, giảm GC pressure
var bufferPool = sync.Pool{
New: func() any {
return new(bytes.Buffer)
},
}
func processRequest(data []byte) string {
// Lấy buffer từ pool
buf := bufferPool.Get().(*bytes.Buffer)
defer func() {
buf.Reset()
bufferPool.Put(buf) // Trả lại pool
}()
buf.Write(data)
buf.WriteString(" processed")
return buf.String()
}
// Use case phổ biến:
// - JSON encoder/decoder
// - HTTP request/response buffers
// - Temporary byte slices
2.5.同步映射
// sync.Map: concurrent-safe map (không cần mutex)
// Phù hợp khi: keys ổn định, read nhiều hơn write
var cache sync.Map
func main() {
// Store
cache.Store("key1", "value1")
cache.Store("key2", 42)
// Load
val, ok := cache.Load("key1")
if ok {
fmt.Println(val.(string)) // "value1"
}
// LoadOrStore (get or set if not exists)
actual, loaded := cache.LoadOrStore("key3", "default")
fmt.Println(actual, loaded) // "default", false (mới tạo)
// Delete
cache.Delete("key1")
// Range
cache.Range(func(key, value any) bool {
fmt.Printf("%v: %v\n", key, value)
return true // return false để dừng
})
}
3. errgroup (golang.org/x/sync)
import "golang.org/x/sync/errgroup"
// errgroup: chạy nhiều goroutines, collect errors, auto cancellation
func fetchUserData(ctx context.Context, userID int) (*UserProfile, error) {
g, ctx := errgroup.WithContext(ctx)
var user *User
var orders []Order
var reviews []Review
// Fetch user info
g.Go(func() error {
var err error
user, err = getUser(ctx, userID)
return err
})
// Fetch orders (concurrent)
g.Go(func() error {
var err error
orders, err = getOrders(ctx, userID)
return err
})
// Fetch reviews (concurrent)
g.Go(func() error {
var err error
reviews, err = getReviews(ctx, userID)
return err
})
// Wait for all goroutines
if err := g.Wait(); err != nil {
return nil, err // Returns first error, cancels others
}
return &UserProfile{
User: user,
Orders: orders,
Reviews: reviews,
}, nil
}
// errgroup with concurrency limit
func processItems(ctx context.Context, items []Item) error {
g, ctx := errgroup.WithContext(ctx)
g.SetLimit(10) // Max 10 concurrent goroutines
for _, item := range items {
item := item // capture loop variable
g.Go(func() error {
return processItem(ctx, item)
})
}
return g.Wait()
}
4. 進階並發模式
4.1.信號量模式
// Semaphore giới hạn số goroutines chạy đồng thời
type Semaphore struct {
ch chan struct{}
}
func NewSemaphore(max int) *Semaphore {
return &Semaphore{ch: make(chan struct{}, max)}
}
func (s *Semaphore) Acquire() {
s.ch <- struct{}{}
}
func (s *Semaphore) Release() {
<-s.ch
}
func main() {
sem := NewSemaphore(3) // Max 3 concurrent
var wg sync.WaitGroup
for i := 0; i < 10; i++ {
wg.Add(1)
go func(id int) {
defer wg.Done()
sem.Acquire()
defer sem.Release()
fmt.Printf("Worker %d running\n", id)
time.Sleep(time.Second)
}(i)
}
wg.Wait()
}
// Hoặc dùng golang.org/x/sync/semaphore cho weighted semaphore
4.2.速率限制器
// Rate limiter với time.Ticker
func rateLimitedProcess(items []string, rps int) {
ticker := time.NewTicker(time.Second / time.Duration(rps))
defer ticker.Stop()
for _, item := range items {
<-ticker.C // Chờ tick
go func(item string) {
fmt.Println("Processing:", item)
}(item)
}
}
// Token bucket rate limiter
type RateLimiter struct {
tokens chan struct{}
}
func NewRateLimiter(rate int, burst int) *RateLimiter {
rl := &RateLimiter{
tokens: make(chan struct{}, burst),
}
// Fill tokens
for i := 0; i < burst; i++ {
rl.tokens <- struct{}{}
}
// Refill tokens
go func() {
ticker := time.NewTicker(time.Second / time.Duration(rate))
defer ticker.Stop()
for range ticker.C {
select {
case rl.tokens <- struct{}{}:
default: // bucket full
}
}
}()
return rl
}
func (rl *RateLimiter) Allow() bool {
select {
case <-rl.tokens:
return true
default:
return false
}
}
4.3. Or-Done 頻道
// Wrap channel với done signal
func orDone(done <-chan struct{}, ch <-chan int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for {
select {
case <-done:
return
case val, ok := <-ch:
if !ok {
return
}
select {
case out <- val:
case <-done:
return
}
}
}
}()
return out
}
5. 並發最佳實踐
- 總是使用
去跑步比賽檢測競爭條件 - 上下文是第一個參數 每個具有 I/O 的函數
- 總是呼叫cancel() 與
延後取消() - 首選管道 為了溝通， 互斥量 為了保護國家
- 不共享記憶體 — 透過管道發送副本
- 避免 goroutine 洩漏 — 總有辦法讓 goroutine 返回
- 使用錯誤群組 而不是 WaitGroup + 手動錯誤處理
- 緩衝通道 對於已知大小的數據，未緩衝同步
下一篇： HTTP 伺服器和 JSON 處理 — 使用標準庫建立 HTTP 伺服器。