Go语言并发编程详解：对比Java CompletableFuture

Go语言以其简洁高效的并发模型著称，通过Goroutines和Channels提供了比Java CompletableFuture更底层但更灵活的并发编程能力。以下是Go语言中实现类似CompletableFuture功能的完整指南。

一、核心概念对比

二、Go语言并发核心机制

1. Goroutines（协程）

Go的并发基础，比Java线程轻量得多：

package main

import (
    "fmt"
    "time"
)

func main() {
    // 启动一个goroutine
    go func() {
        fmt.Println("Hello from goroutine!")
    }()
    
    // 主goroutine需要等待
    time.Sleep(100 * time.Millisecond)
}

特点：

• 内存开销：初始约2KB（Java线程约1MB）
• 调度：由Go运行时管理，M:N调度模型
• 生命周期：当函数返回时自动结束

2. Channels（通道）

Goroutines之间的通信桥梁：

func main() {
    // 创建无缓冲通道
    ch := make(chan string)
    
    go func() {
        time.Sleep(1 * time.Second)
        ch <- "Hello from goroutine"
    }()
    
    // 从通道接收数据
    msg := <-ch
    fmt.Println(msg)
}

通道类型：

• 无缓冲通道：同步通信，发送和接收必须同时发生
• 有缓冲通道：异步通信，缓冲区满时阻塞

// 无缓冲
ch1 := make(chan int)

// 有缓冲（容量10）
ch2 := make(chan int, 10)

三、实现CompletableFuture功能

1. 基本异步任务（类似supplyAsync）

package main

import (
    "fmt"
    "time"
)

// 模拟异步任务
func asyncTask(id string) <-chan string {
    ch := make(chan string)
    
    go func() {
        defer close(ch) // 关闭通道
        fmt.Printf("Starting task %s\n", id)
        time.Sleep(2 * time.Second)
        ch <- fmt.Sprintf("Result from %s", id)
    }()
    
    return ch
}

func main() {
    resultCh := asyncTask("Task1")
    result := <-resultCh
    fmt.Println(result)
}

2. 任务链式调用（类似thenApply）

func chainTasks() {
    // 第一个任务
    task1 := func() <-chan string {
        ch := make(chan string)
        go func() {
            defer close(ch)
            time.Sleep(1 * time.Second)
            ch <- "Hello"
        }()
        return ch
    }
    
    // 第二个任务，依赖第一个
    task2 := func(input <-chan string) <-chan string {
        ch := make(chan string)
        go func() {
            defer close(ch)
            data := <-input
            time.Sleep(1 * time.Second)
            ch <- fmt.Sprintf("%s World", data)
        }()
        return ch
    }
    
    // 执行链式调用
    result1 := task1()
    result2 := task2(result1)
    
    finalResult := <-result2
    fmt.Println(finalResult) // 输出: Hello World
}

3. 多任务组合（类似allOf）

func allOfExample() {
    // 创建多个任务
    task1 := func() <-chan string {
        ch := make(chan string)
        go func() {
            defer close(ch)
            time.Sleep(2 * time.Second)
            ch <- "Result1"
        }()
        return ch
    }
    
    task2 := func() <-chan string {
        ch := make(chan string)
        go func() {
            defer close(ch)
            time.Sleep(1 * time.Second)
            ch <- "Result2"
        }()
        return ch
    }
    
    task3 := func() <-chan string {
        ch := make(chan string)
        go func() {
            defer close(ch)
            time.Sleep(3 * time.Second)
            ch <- "Result3"
        }()
        return ch
    }
    
    // 启动所有任务
    ch1 := task1()
    ch2 := task2()
    ch3 := task3()
    
    // 等待所有任务完成
    results := make([]string, 3)
    results[0] = <-ch1
    results[1] = <-ch2
    results[2] = <-ch3
    
    fmt.Println("All results:", results)
}

4. 任一任务完成（类似anyOf）

func anyOfExample() {
    task1 := func() <-chan string {
        ch := make(chan string)
        go func() {
            defer close(ch)
            time.Sleep(3 * time.Second)
            ch <- "Fast result"
        }()
        return ch
    }
    
    task2 := func() <-chan string {
        ch := make(chan string)
        go func() {
            defer close(ch)
            time.Sleep(5 * time.Second)
            ch <- "Slow result"
        }()
        return ch
    }
    
    ch1 := task1()
    ch2 := task2()
    
    // 使用select获取最快的结果
    select {
    case result := <-ch1:
        fmt.Println("Got result:", result)
    case result := <-ch2:
        fmt.Println("Got result:", result)
    }
}

5. 超时控制

func timeoutExample() {
    task := func() <-chan string {
        ch := make(chan string)
        go func() {
            defer close(ch)
            time.Sleep(4 * time.Second) // 模拟长时间任务
            ch <- "Task completed"
        }()
        return ch
    }
    
    ch := task()
    
    // 设置2秒超时
    select {
    case result := <-ch:
        fmt.Println("Success:", result)
    case <-time.After(2 * time.Second):
        fmt.Println("Timeout: Task took too long")
    }
}

6. 上下文控制（推荐方式）

func contextTimeoutExample() {
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()
    
    task := func(ctx context.Context) <-chan string {
        ch := make(chan string)
        go func() {
            defer close(ch)
            
            // 模拟工作
            select {
            case <-time.After(3 * time.Second):
                ch <- "Task completed"
            case <-ctx.Done():
                ch <- "Task cancelled"
                return
            }
        }()
        return ch
    }
    
    resultCh := task(ctx)
    result := <-resultCh
    fmt.Println(result)
}

四、高级并发模式

1. Worker Pool（工作池）

func workerPool() {
    const numWorkers = 3
    const numJobs = 5
    
    jobs := make(chan int, numJobs)
    results := make(chan int, numJobs)
    
    // 启动worker
    for w := 1; w <= numWorkers; w++ {
        go func(workerID int) {
            for job := range jobs {
                fmt.Printf("Worker %d started job %d\n", workerID, job)
                time.Sleep(time.Second)
                fmt.Printf("Worker %d finished job %d\n", workerID, job)
                results <- job * 2
            }
        }(w)
    }
    
    // 发送工作
    for j := 1; j <= numJobs; j++ {
        jobs <- j
    }
    close(jobs)
    
    // 收集结果
    for a := 1; a <= numJobs; a++ {
        <-results
    }
}

2. 任务扇出/扇入（Fan-out/Fan-in）

func fanOutIn() {
    // 扇出：一个生产者，多个消费者
    producer := func() <-chan int {
        ch := make(chan int, 100)
        go func() {
            defer close(ch)
            for i := 1; i <= 10; i++ {
                ch <- i
            }
        }()
        return ch
    }
    
    // 消费者
    consumer := func(in <-chan int) <-chan int {
        out := make(chan int)
        go func() {
            defer close(out)
            for val := range in {
                out <- val * val
            }
        }()
        return out
    }
    
    // 生产数据
    data := producer()
    
    // 多个消费者
    c1 := consumer(data)
    c2 := consumer(data)
    c3 := consumer(data)
    
    // 扇入：合并结果
    merge := func(cs ...<-chan int) <-chan int {
        var wg sync.WaitGroup
        out := make(chan int)
        
        output := func(c <-chan int) {
            defer wg.Done()
            for n := range c {
                out <- n
            }
        }
        
        wg.Add(len(cs))
        for _, c := range cs {
            go output(c)
        }
        
        go func() {
            wg.Wait()
            close(out)
        }()
        
        return out
    }
    
    // 合并所有结果
    merged := merge(c1, c2, c3)
    for result := range merged {
        fmt.Println("Result:", result)
    }
}

3. 错误处理

func errorHandling() {
    type Result struct {
        Data  string
        Error error
    }
    
    task := func() <-chan Result {
        ch := make(chan Result, 1)
        go func() {
            defer close(ch)
            
            // 模拟可能出错的任务
            if time.Now().UnixNano()%2 == 0 {
                ch <- Result{Error: fmt.Errorf("task failed")}
            } else {
                ch <- Result{Data: "Success"}
            }
        }()
        return ch
    }
    
    resultCh := task()
    result := <-resultCh
    
    if result.Error != nil {
        fmt.Printf("Task failed: %v\n", result.Error)
    } else {
        fmt.Printf("Task succeeded: %s\n", result.Data)
    }
}

五、完整实战：电商订单处理

package main

import (
    "context"
    "fmt"
    "sync"
    "time"
)

type Order struct {
    ID     string
    UserID string
    Items  []string
}

type User struct {
    ID    string
    Name  string
    Email string
}

type Product struct {
    ID    string
    Name  string
    Price float64
}

func (p *Product) Discount() float64 {
    return p.Price * 0.9
}

// 获取用户信息
func getUser(ctx context.Context, userID string) <-chan User {
    ch := make(chan User, 1)
    go func() {
        defer close(ch)
        
        select {
        case <-ctx.Done():
            return
        case <-time.After(300 * time.Millisecond):
            ch <- User{
                ID:    userID,
                Name:  "Alice",
                Email: "alice@example.com",
            }
        }
    }()
    return ch
}

// 获取商品信息
func getProducts(ctx context.Context, productIDs []string) <-chan []Product {
    ch := make(chan []Product, 1)
    go func() {
        defer close(ch)
        
        select {
        case <-ctx.Done():
            return
        case <-time.After(200 * time.Millisecond):
            products := make([]Product, 0, len(productIDs))
            for _, id := range productIDs {
                products = append(products, Product{
                    ID:    id,
                    Name:  fmt.Sprintf("Product %s", id),
                    Price: 100.0,
                })
            }
            ch <- products
        }
    }()
    return ch
}

// 计算价格
func calculatePrice(ctx context.Context, user User, products []Product) <-chan float64 {
    ch := make(chan float64, 1)
    go func() {
        defer close(ch)
        
        select {
        case <-ctx.Done():
            return
        case <-time.After(100 * time.Millisecond):
            total := 0.0
            for _, p := range products {
                total += p.Discount()
            }
            ch <- total
        }
    }()
    return ch
}

// 处理支付
func processPayment(ctx context.Context, amount float64) <-chan string {
    ch := make(chan string, 1)
    go func() {
        defer close(ch)
        
        select {
        case <-ctx.Done():
            ch <- "Payment cancelled"
            return
        case <-time.After(250 * time.Millisecond):
            ch <- fmt.Sprintf("Payment successful: $%.2f", amount)
        }
    }()
    return ch
}

// 电商订单处理主流程
func processOrder(order Order) {
    // 创建上下文，设置总超时
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()
    
    var wg sync.WaitGroup
    
    // 1. 获取用户信息
    userCh := getUser(ctx, order.UserID)
    
    // 2. 获取商品信息
    productsCh := getProducts(ctx, order.Items)
    
    // 3. 等待用户和商品信息
    var user User
    var products []Product
    
    wg.Add(2)
    go func() {
        defer wg.Done()
        u, ok := <-userCh
        if !ok || ctx.Err() != nil {
            fmt.Println("Failed to get user info")
            return
        }
        user = u
    }()
    
    go func() {
        defer wg.Done()
        p, ok := <-productsCh
        if !ok || ctx.Err() != nil {
            fmt.Println("Failed to get products info")
            return
        }
        products = p
    }()
    
    wg.Wait()
    
    // 检查上下文是否已取消
    if ctx.Err() != nil {
        fmt.Printf("Order processing cancelled: %v\n", ctx.Err())
        return
    }
    
    // 4. 计算价格
    priceCh := calculatePrice(ctx, user, products)
    price, ok := <-priceCh
    if !ok || ctx.Err() != nil {
        fmt.Println("Failed to calculate price")
        return
    }
    
    // 5. 处理支付
    paymentCh := processPayment(ctx, price)
    paymentResult, ok := <-paymentCh
    if !ok || ctx.Err() != nil {
        fmt.Println("Payment failed")
        return
    }
    
    // 6. 输出结果
    fmt.Printf("Order %s processed successfully!\n", order.ID)
    fmt.Printf("User: %s (%s)\n", user.Name, user.Email)
    fmt.Printf("Products: %v\n", products)
    fmt.Printf("Total: $%.2f\n", price)
    fmt.Printf("Payment: %s\n", paymentResult)
}

func main() {
    order := Order{
        ID:     "ORD-001",
        UserID: "USR-001",
        Items:  []string{"P001", "P002", "P003"},
    }
    
    processOrder(order)
}

六、最佳实践总结

1. 通道使用原则

// ✅ 正确：发送方关闭通道
func producer() <-chan int {
    ch := make(chan int)
    go func() {
        defer close(ch) // 发送方负责关闭
        for i := 0; i < 10; i++ {
            ch <- i
        }
    }()
    return ch
}

// ❌ 错误：接收方尝试关闭
func consumer(ch <-chan int) {
    for val := range ch {
        fmt.Println(val)
    }
    // close(ch) // 不能关闭只读通道
}

2. 上下文最佳实践

func serviceCall(ctx context.Context) error {
    // 检查上下文是否已取消
    select {
    case <-ctx.Done():
        return ctx.Err()
    default:
    }
    
    // 执行业务逻辑
    time.Sleep(100 * time.Millisecond)
    
    return nil
}

3. 错误处理模式

type Result struct {
    Data  interface{}
    Error error
}

func asyncOperation() <-chan Result {
    ch := make(chan Result, 1)
    go func() {
        defer close(ch)
        
        // 模拟操作
        if err := someWork(); err != nil {
            ch <- Result{Error: err}
            return
        }
        
        ch <- Result{Data: "success"}
    }()
    return ch
}

4. 资源清理

func cleanupExample() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel() // 确保清理
    
    // 启动后台任务
    go func() {
        for {
            select {
            case <-ctx.Done():
                fmt.Println("Cleaning up resources...")
                return
            default:
                // 执行工作
                time.Sleep(100 * time.Millisecond)
            }
        }
    }()
    
    // 模拟主程序运行
    time.Sleep(2 * time.Second)
}

七、与Java CompletableFuture对比总结

选择建议：

• 简单任务：Go语言更简洁
• 复杂任务链：Java CompletableFuture的链式调用更直观
• 高并发场景：Go语言性能优势明显
• 团队熟悉度：根据团队技术栈选择

Go语言的并发模型虽然需要更多手动管理，但提供了更大的灵活性和更好的性能，特别适合高并发、微服务等场景。