1
2

// WaitGroup 用于等待一组goroutine完成
var wg sync.WaitGroup

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

func main() {
    fmt.Println("=== 屏障模式：旅游团集合 ===\n")

    // 创建WaitGroup
    var wg sync.WaitGroup

    // 假设有5个景点，5个游客分别去不同景点
    attractions := []string{"故宫", "长城", "天坛", "颐和园", "圆明园"}

    // 每个游客游览一个景点
    for i, attraction := range attractions {
        // 增加一个待等待的goroutine
        wg.Add(1)

        // 启动goroutine
        go func(name string, index int) {
            defer wg.Done() // goroutine完成时调用

            // 模拟游览时间（随机）
            duration := time.Duration(100+rand.Intn(400)) * time.Millisecond
            fmt.Printf("[游客%d] 前往%s游览，预计%d毫秒...\n", index+1, name, duration)

            // 游览
            time.Sleep(duration)

            // 游览完毕，去集合点等
            fmt.Printf("[游客%d] 🏃 已到达集合点，等待其他人...\n", index+1)
        }(attraction, i)
    }

    fmt.Println("[导游] 大家都去玩了，我在这等你们...")

    // 等待所有goroutine完成（屏障点）
    wg.Wait()

    fmt.Println("\n[导游] 太好了！所有人都到齐了！出发去下一个景点！")
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

// 并行计算矩阵乘法的例子
func parallelMatrixMultiply(A, B [][]int, result [][]int, rowCount, colCount, midCount int) {
    var wg sync.WaitGroup

    // 每个goroutine负责计算结果矩阵的一行
    for i := 0; i < rowCount; i++ {
        wg.Add(1)
        go func(row int) {
            defer wg.Done()
            for j := 0; j < colCount; j++ {
                sum := 0
                for k := 0; k < midCount; k++ {
                    sum += A[row][k] * B[k][j]
                }
                result[row][j] = sum
            }
            fmt.Printf("[Worker] 第%d行计算完成\n", row)
        }(i)
    }

    // 屏障：等待所有行计算完成
    wg.Wait()

    fmt.Println("[主程序] 所有行计算完成，矩阵乘法结束")
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

func main() {
    fmt.Println("=== 屏障模式：并行计算 ===\n")

    // 简单的例子：计算1到100的和，分成10组并行计算
    const totalNumbers = 100
    const numWorkers = 10
    const numbersPerWorker = totalNumbers / numWorkers

    results := make([]int, numWorkers)
    var wg sync.WaitGroup

    fmt.Printf("计算1到%d的和，分成%d组并行计算...\n\n", totalNumbers, numWorkers)

    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        start := i * numbersPerWorker
        end := start + numbersPerWorker

        go func(workerID, start, end int) {
            defer wg.Done()

            sum := 0
            for n := start + 1; n <= end; n++ {
                sum += n
            }
            results[workerID] = sum
            fmt.Printf("[Worker %d] 计算 %d 到 %d 的和 = %d\n", workerID, start+1, end, sum)
        }(i, start, end)
    }

    // 屏障等待所有worker完成
    wg.Wait()

    // 合并结果
    total := 0
    for i, r := range results {
        fmt.Printf("第%d组结果: %d\n", i, r)
        total += r
    }

    fmt.Printf("\n最终结果: %d\n", total)
    fmt.Printf("验证: 1+2+...+100 = %d\n", totalNumbers*(totalNumbers+1)/2)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

// ========== 第一步：定义Future结构 ==========

// Future 异步计算的结果
// 这是一个"凭证"，你可以用它来获取最终结果
type Future struct {
    resultChan chan interface{} // 存放结果的channel
    done       bool
    value      interface{}     // 缓存结果
    mu         sync.Mutex
}

// NewFuture 创建一个Future
func NewFuture() *Future {
    return &Future{
        resultChan: make(chan interface{}, 1), // 缓冲1，防止协程泄漏
    }
}

// Resolve 完成Future，设置结果
// 这通常由异步任务调用
func (f *Future) Resolve(value interface{}) {
    f.mu.Lock()
    defer f.mu.Unlock()

    if f.done {
        return // 已经被resolve了
    }

    f.value = value
    f.done = true
    close(f.resultChan) // 通知等待者
}

// Get 获取结果，会阻塞直到结果可用
func (f *Future) Get() interface{} {
    <-f.resultChan // 等待结果

    f.mu.Lock()
    defer f.mu.Unlock()
    return f.value
}

// GetWithTimeout 带超时的获取
func (f *Future) GetWithTimeout(duration time.Duration) (interface{}, error) {
    select {
    case <-f.resultChan:
        f.mu.Lock()
        defer f.mu.Unlock()
        return f.value, nil
    case <-time.After(duration):
        return nil, fmt.Errorf("超时了")
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

// ========== 第二步：定义Promise（用于创建Future） ==========

// Promise 用于创建和管理Future
type Promise struct {
    future *Future
}

// NewPromise 创建一个Promise
func NewPromise() *Promise {
    return &Promise{
        future: NewFuture(),
    }
}

// GetFuture 获取Promise关联的Future
func (p *Promise) GetFuture() *Future {
    return p.future
}

// Resolve 完成Promise
func (p *Promise) Resolve(value interface{}) {
    p.future.Resolve(value)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

// ========== 第三步：封装异步任务 ==========

// AsyncCall 异步执行一个函数，返回Future
func AsyncCall(fn func() interface{}) *Future {
    promise := NewPromise()

    go func() {
        // 异步执行函数
        result := fn()
        // 完成Promise
        promise.Resolve(result)
    }()

    return promise.GetFuture()
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

// ========== 第四步：使用示例 ==========

// 模拟一个耗时的任务
func fetchUserData(userID int) interface{} {
    fmt.Printf("[异步任务] 开始获取用户%d的数据...\n", userID)
    time.Sleep(2 * time.Second) // 模拟耗时操作
    fmt.Printf("[异步任务] 用户%d的数据获取完成！\n", userID)
    return fmt.Sprintf("用户%d的数据: {name: '张三', age: 25}", userID)
}

func main() {
    fmt.Println("=== Future/Promise 模式 ===\n")

    // ===== 场景1：基本用法 =====
    fmt.Println("--- 场景1: 基本用法 ---")

    // 启动异步任务
    future := AsyncCall(func() interface{} {
        return fetchUserData(123)
    })

    fmt.Println("[主程序] 异步任务已启动，我可以先做其他事情...")

    // 模拟做其他事情
    time.Sleep(500 * time.Millisecond)
    fmt.Println("[主程序] 我还在做其他事情...")

    time.Sleep(500 * time.Millisecond)
    fmt.Println("[主程序] 我还在做其他事情...")

    // 获取结果（会阻塞直到结果可用）
    fmt.Println("[主程序] 等待结果...")
    result := future.Get()
    fmt.Printf("[主程序] 收到结果: %s\n\n", result)

    // ===== 场景2：多个异步任务 =====
    fmt.Println("--- 场景2: 多个异步任务 ---")

    // 同时发起多个异步请求
    future1 := AsyncCall(func() interface{} {
        time.Sleep(1 * time.Second)
        return "任务1完成"
    })

    future2 := AsyncCall(func() interface{} {
        time.Sleep(2 * time.Second)
        return "任务2完成"
    })

    future3 := AsyncCall(func() interface{} {
        time.Sleep(500 * time.Millisecond)
        return "任务3完成"
    })

    // 等待所有结果
    fmt.Println("[主程序] 等待所有任务完成...")
    result1 := future1.Get()
    fmt.Printf("[主程序] 收到: %s\n", result1)
    result2 := future2.Get()
    fmt.Printf("[主程序] 收到: %s\n", result2)
    result3 := future3.Get()
    fmt.Printf("[主程序] 收到: %s\n\n", result3)

    // ===== 场景3：超时处理 =====
    fmt.Println("--- 场景3: 超时处理 ---")

    future4 := AsyncCall(func() interface{} {
        time.Sleep(5 * time.Second) // 模拟一个很慢的任务
        return "终于完成了！"
    })

    fmt.Println("[主程序] 等待结果（最多3秒）...")
    result4, err := future4.GetWithTimeout(3 * time.Second)
    if err != nil {
        fmt.Printf("[主程序] ⚠️ 错误: %v\n", err)
    } else {
        fmt.Printf("[主程序] 收到结果: %s\n", result4)
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

func main() {
    // 创建一个可以取消的context
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // 启动异步任务
    resultChan := make(chan string, 1)

    go func() {
        // 模拟耗时任务
        time.Sleep(2 * time.Second)
        resultChan <- "任务完成"
    }()

    // 等待结果或取消
    select {
    case result := <-resultChan:
        fmt.Printf("收到结果: %s\n", result)
    case <-time.After(3 * time.Second):
        fmt.Println("超时")
    case <-ctx.Done():
        fmt.Println("被取消")
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

// ========== 第一步：定义管道的各个阶段 ==========

// Generator 生成器：产生数据
// 这是管道的第一阶段，产生初始数据
func Generator(nums ...int) <-chan int {
    out := make(chan int)

    go func() {
        defer close(out)
        for _, n := range nums {
            out <- n
        }
    }()

    return out
}

// MultiplyStage 乘法阶段：把每个数乘以2
func MultiplyStage(in <-chan int, multiplier int) <-chan int {
    out := make(chan int)

    go func() {
        defer close(out)
        for n := range in {
            result := n * multiplier
            fmt.Printf("[乘%d阶段] %d × %d = %d\n", multiplier, n, multiplier, result)
            out <- result
        }
    }()

    return out
}

// AddStage 加法阶段：把每个数加一个值
func AddStage(in <-chan int, addend int) <-chan int {
    out := make(chan int)

    go func() {
        defer close(out)
        for n := range in {
            result := n + addend
            fmt.Printf("[加%d阶段] %d + %d = %d\n", addend, n, addend, result)
            out <- result
        }
    }()

    return out
}

// FilterStage 过滤阶段：只保留偶数
func FilterStage(in <-chan int) <-chan int {
    out := make(chan int)

    go func() {
        defer close(out)
        for n := range in {
            if n%2 == 0 {
                fmt.Printf("[过滤阶段] %d 是偶数，保留\n", n)
                out <- n
            } else {
                fmt.Printf("[过滤阶段] %d 是奇数，过滤掉\n", n)
            }
        }
    }()

    return out
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

func main() {
    fmt.Println("=== 管道模式：流水线处理 ===\n")

    // ===== 构建管道 =====
    // 数据源：生成 1, 2, 3, 4, 5
    fmt.Println("步骤1: 生成数据 1, 2, 3, 4, 5")
    data := Generator(1, 2, 3, 4, 5)

    // 乘以2
    fmt.Println("\n步骤2: 每个数乘以2")
    multiplied := MultiplyStage(data, 2)

    // 加3
    fmt.Println("\n步骤3: 每个数加3")
    added := AddStage(multiplied, 3)

    // 过滤，只保留偶数
    fmt.Println("\n步骤4: 只保留偶数")
    filtered := FilterStage(added)

    // ===== 最终结果 =====
    fmt.Println("\n========== 最终结果 ==========")
    for result := range filtered {
        fmt.Printf("最终结果: %d\n", result)
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

// ========== 实战：处理日志流 ==========

// LogLine 日志行
type LogLine struct {
    timestamp time.Time
    level     string
    message   string
}

// ParseLogStage 解析日志阶段
func ParseLogStage(in <-chan string) <-chan *LogLine {
    out := make(chan *LogLine)

    go func() {
        defer close(out)
        for line := range in {
            // 简单解析：timestamp | level | message
            parts := strings.Split(line, "|")
            if len(parts) == 3 {
                tm, _ := time.Parse("15:04:05", strings.TrimSpace(parts[0]))
                out <- &LogLine{
                    timestamp: tm,
                    level:     strings.TrimSpace(parts[1]),
                    message:   strings.TrimSpace(parts[2]),
                }
            }
        }
    }()

    return out
}

// FilterByLevelStage 按级别过滤
func FilterByLevelStage(in <-chan *LogLine, minLevel string) <-chan *LogLine {
    levelPriority := map[string]int{
        "DEBUG": 0,
        "INFO":  1,
        "WARN":  2,
        "ERROR": 3,
    }

    minPriority := levelPriority[minLevel]
    out := make(chan *LogLine)

    go func() {
        defer close(out)
        for line := range in {
            if levelPriority[line.level] >= minPriority {
                out <- line
            }
        }
    }()

    return out
}

// FormatOutputStage 格式化输出
func FormatOutputStage(in <-chan *LogLine) <-chan string {
    out := make(chan string)

    go func() {
        defer close(out)
        for line := range in {
            formatted := fmt.Sprintf("[%s] %s: %s",
                line.timestamp.Format("15:04:05"),
                line.level,
                line.message)
            out <- formatted
        }
    }()

    return out
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

func main() {
    fmt.Println("=== 管道模式：日志处理流水线 ===\n")

    // 模拟日志源
    logs := []string{
        "10:00:01|DEBUG|Starting application",
        "10:00:02|INFO|Server listening on :8080",
        "10:00:03|WARN|Connection timeout, retrying...",
        "10:00:04|ERROR|Failed to connect to database",
        "10:00:05|INFO|Retrying connection",
        "10:00:06|DEBUG|Cache miss for key: user_123",
        "10:00:07|WARN|High memory usage: 85%",
        "10:00:08|ERROR|Panic: nil pointer dereference",
    }

    // 创建数据源channel
    logChan := make(chan string, len(logs))
    for _, log := range logs {
        logChan <- log
    }
    close(logChan)

    // 构建管道：解析 -> 过滤WARN及以上 -> 格式化 -> 输出
    fmt.Println("原始日志:")
    for _, l := range logs {
        fmt.Printf("  %s\n", l)
    }
    fmt.Println()

    parsed := ParseLogStage(logChan)
    filtered := FilterByLevelStage(parsed, "WARN")
    formatted := FormatOutputStage(filtered)

    fmt.Println("========== 处理结果（WARN及以上）==========")
    for line := range formatted {
        fmt.Println(line)
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

// URLs 是要下载的URL列表
type URLs []string

// FetchResult 下载结果
type FetchResult struct {
    URL    string
    Status string
    Length int
    Error  error
}

// FanOutFetch 扇出：并行下载多个URL
func FanOutFetch(urls []string, concurrency int) []FetchResult {
    // 创建结果channel
    results := make(chan FetchResult, len(urls))

    // 创建任务channel
    tasks := make(chan string, len(urls))
    for _, url := range urls {
        tasks <- url
    }
    close(tasks)

    // 启动固定数量的worker（扇出）
    var wg sync.WaitGroup
    for i := 0; i < concurrency; i++ {
        wg.Add(1)
        go func(workerID int) {
            defer wg.Done()
            fmt.Printf("[Worker %d] 开始工作\n", workerID)

            for url := range tasks {
                // 模拟下载
                fmt.Printf("[Worker %d] 下载: %s\n", workerID, url)
                time.Sleep(time.Duration(100+rand.Intn(200)) * time.Millisecond)

                // 模拟结果
                result := FetchResult{
                    URL:    url,
                    Status: "200 OK",
                    Length: 1000 + rand.Intn(9000),
                    Error:  nil,
                }
                results <- result
            }
            fmt.Printf("[Worker %d] 工作完成\n", workerID)
        }(i)
    }

    // 等待所有worker完成，然后关闭结果channel（扇入点）
    go func() {
        wg.Wait()
        close(results)
    }()

    // 收集所有结果
    var fetchResults []FetchResult
    for result := range results {
        fetchResults = append(fetchResults, result)
    }

    return fetchResults
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

func main() {
    fmt.Println("=== 扇出扇入模式：并行下载 ===\n")

    urls := []string{
        "https://example.com/page1",
        "https://example.com/page2",
        "https://example.com/page3",
        "https://example.com/page4",
        "https://example.com/page5",
        "https://example.com/page6",
        "https://example.com/page7",
        "https://example.com/page8",
    }

    fmt.Printf("需要下载 %d 个URL，使用3个Worker并行下载\n\n", len(urls))

    start := time.Now()
    results := FanOutFetch(urls, 3)
    elapsed := time.Since(start)

    fmt.Println("\n========== 下载结果 ==========")
    for _, r := range results {
        status := "✅"
        if r.Error != nil {
            status = "❌"
        }
        fmt.Printf("%s %s - %s (%d bytes)\n", status, r.URL, r.Status, r.Length)
    }
    fmt.Printf("\n总耗时: %v\n", elapsed)
    fmt.Printf("平均每个URL耗时: %v\n", elapsed/time.Duration(len(urls)))
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

// ========== MapReduce 风格的 Word Count ==========

// WordCountResult 词频统计结果
type WordCountResult struct {
    Word  string
    Count int
}

// MapStage 映射阶段：统计每个分片的词频
func MapStage(texts []string) <-chan map[string]int {
    out := make(chan map[string]int)

    go func() {
        defer close(out)
        for _, text := range texts {
            // 简单分词
            words := strings.Fields(text)
            wordCount := make(map[string]int)
            for _, word := range words {
                wordCount[strings.ToLower(word)]++
            }
            out <- wordCount
        }
    }()

    return out
}

// ReduceStage 归并阶段：合并多个词频统计
func ReduceStage(maps <-chan map[string]int) map[string]int {
    finalCount := make(map[string]int)

    for m := range maps {
        for word, count := range m {
            finalCount[word] += count
        }
    }

    return finalCount
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

func main() {
    fmt.Println("=== 扇出扇入模式：MapReduce 词频统计 ===\n")

    texts := []string{
        "hello world hello",
        "world of golang",
        "hello golang world",
        "golang is great",
        "hello hello world",
    }

    fmt.Println("原始文本:")
    for i, t := range texts {
        fmt.Printf("  [%d] %s\n", i, t)
    }
    fmt.Println()

    // Map阶段（扇出）
    mapResults := MapStage(texts)

    // Reduce阶段（扇入）
    finalCount := ReduceStage(mapResults)

    fmt.Println("========== 词频统计结果 ==========")
    for word, count := range finalCount {
        fmt.Printf("%s: %d\n", word, count)
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

// ========== 第一步：定义消息和订阅者 ==========

// Message 消息结构
type Message struct {
    Topic   string      // 话题/主题
    Content interface{} // 消息内容
    ID      int64       // 消息ID
}

// Subscriber 订阅者接口
type Subscriber interface {
    // GetTopics 返回订阅者感兴趣的话题
    GetTopics() []string
    // OnMessage 收到消息时的回调
    OnMessage(msg *Message)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

// ========== 第二步：定义消息代理 ==========

// PubSub 消息代理
type PubSub struct {
    mu         sync.RWMutex
    topics     map[string][]chan *Message // topic -> message channel
    subscribers map[string][]Subscriber   // topic -> subscribers
    nextMsgID  int64
}

// NewPubSub 创建消息代理
func NewPubSub() *PubSub {
    return &PubSub{
        topics:     make(map[string][]chan *Message),
        subscribers: make(map[string][]Subscriber),
    }
}

// Subscribe 订阅话题
func (ps *PubSub) Subscribe(topic string, ch chan *Message) {
    ps.mu.Lock()
    defer ps.mu.Unlock()

    ps.topics[topic] = append(ps.topics[topic], ch)
    fmt.Printf("[Broker] 新订阅: topic=%s, 当前订阅者数量=%d\n", topic, len(ps.topics[topic]))
}

// SubscribeWithCallback 使用回调订阅
func (ps *PubSub) SubscribeWithCallback(subscriber Subscriber) {
    ps.mu.Lock()
    defer ps.mu.Unlock()

    for _, topic := range subscriber.GetTopics() {
        ps.subscribers[topic] = append(ps.subscribers[topic], subscriber)
        fmt.Printf("[Broker] 新订阅者注册 topic=%s\n", topic)
    }
}

// Publish 发布消息
func (ps *PubSub) Publish(topic string, content interface{}) {
    ps.mu.Lock()
    ps.nextMsgID++
    msgID := ps.nextMsgID
    msg := &Message{
        Topic:   topic,
        Content: content,
        ID:      msgID,
    }

    // 获取该topic的所有channel
    channels := ps.topics[topic]
    subscribers := ps.subscribers[topic]
    ps.mu.Unlock()

    // 发送到channel
    for _, ch := range channels {
        // 使用非阻塞发送，防止慢订阅者阻塞整个系统
        select {
        case ch <- msg:
            fmt.Printf("[Broker] 消息#%d 已投递到channel: topic=%s\n", msgID, topic)
        default:
            fmt.Printf("[Broker] ⚠️ 消息#%d 投递失败（channel满）: topic=%s\n", msgID, topic)
        }
    }

    // 回调订阅者
    for _, sub := range subscribers {
        sub.OnMessage(msg)
    }
}

// Unsubscribe 取消订阅
func (ps *PubSub) Unsubscribe(topic string, ch chan *Message) {
    ps.mu.Lock()
    defer ps.mu.Unlock()

    channels := ps.topics[topic]
    for i, c := range channels {
        if c == ch {
            ps.topics[topic] = append(channels[:i], channels[i+1:]...)
            fmt.Printf("[Broker] 取消订阅: topic=%s\n", topic)
            return
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

// ========== 第三步：定义具体的订阅者 ==========

// EmailSubscriber 邮件订阅者
type EmailSubscriber struct {
    name  string
    email string
}

func NewEmailSubscriber(name, email string) *EmailSubscriber {
    return &EmailSubscriber{name: name, email: email}
}

func (s *EmailSubscriber) GetTopics() []string {
    return []string{"news", "tech"}
}

func (s *EmailSubscriber) OnMessage(msg *Message) {
    fmt.Printf("[邮件订阅者 %s] 📧 收到消息: topic=%s, content=%v, from=%s\n",
        s.name, msg.Topic, msg.Content, s.email)
}

// SMSSubscriber 短信订阅者
type SMSSubscriber struct {
    name     string
    phoneNum string
}

func NewSMSSubscriber(name, phoneNum string) *SMSSubscriber {
    return &SMSSubscriber{name: name, phoneNum: phoneNum}
}

func (s *SMSSubscriber) GetTopics() []string {
    return []string{"news", "alert"}
}

func (s *SMSSubscriber) OnMessage(msg *Message) {
    fmt.Printf("[短信订阅者 %s] 📱 收到消息: topic=%s, content=%v, phone=%s\n",
        s.name, msg.Topic, msg.Content, s.phoneNum)
}

// PushSubscriber 推送订阅者
type PushSubscriber struct {
    name string
    appID string
}

func NewPushSubscriber(name, appID string) *PushSubscriber {
    return &PushSubscriber{name: name, appID: appID}
}

func (s *PushSubscriber) GetTopics() []string {
    return []string{"tech", "promo"}
}

func (s *PushSubscriber) OnMessage(msg *Message) {
    fmt.Printf("[推送订阅者 %s] 🔔 收到消息: topic=%s, content=%v, appID=%s\n",
        s.name, msg.Topic, msg.Content, s.appID)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

func main() {
    fmt.Println("=== 发布订阅模式：消息系统 ===\n")

    // 创建消息代理
    broker := NewPubSub()

    // ===== 创建channel订阅者 =====
    newsCh := make(chan *Message, 10)
    techCh := make(chan *Message, 10)

    broker.Subscribe("news", newsCh)
    broker.Subscribe("tech", techCh)

    // ===== 创建callback订阅者 =====
    emailSub := NewEmailSubscriber("张三", "zhangsan@example.com")
    smsSub := NewSMSSubscriber("李四", "13800138000")
    pushSub := NewPushSubscriber("王五", "app_12345")

    broker.SubscribeWithCallback(emailSub)
    broker.SubscribeWithCallback(smsSub)
    broker.SubscribeWithCallback(pushSub)

    // 启动订阅者的goroutine来处理消息
    go func() {
        for msg := range newsCh {
            fmt.Printf("[NewsChannel] 📰 处理消息: topic=%s, content=%v\n", msg.Topic, msg.Content)
        }
    }()

    go func() {
        for msg := range techCh {
            fmt.Printf("[TechChannel] 💻 处理消息: topic=%s, content=%v\n", msg.Topic, msg.Content)
        }
    }()

    // 等待goroutine启动
    time.Sleep(100 * time.Millisecond)

    fmt.Println("\n========== 发布消息 ==========\n")

    // ===== 发布各种消息 =====
    broker.Publish("news", "习近平主席发表重要讲话")
    time.Sleep(200 * time.Millisecond)

    broker.Publish("tech", "Go 1.22正式发布！")
    time.Sleep(200 * time.Millisecond)

    broker.Publish("alert", "系统告警：CPU使用率超过90%")
    time.Sleep(200 * time.Millisecond)

    broker.Publish("promo", "双11大促，全场5折！")
    time.Sleep(200 * time.Millisecond)

    // 等待所有消息处理完成
    time.Sleep(500 * time.Millisecond)

    fmt.Println("\n========== 消息发布完成 ==========")
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

// ========== 第一步：定义任务和结果 ==========

// Task 任务结构
type Task struct {
    ID   int
    Data string
}

// TaskResult 任务结果
type TaskResult struct {
    TaskID   int
    Result   string
    Duration time.Duration
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

// ========== 第二步：定义工作者池 ==========

// WorkerPool 工作者池
type WorkerPool struct {
    workerCount int           // worker数量
    taskQueue  chan *Task    // 任务队列
    resultQueue chan *TaskResult // 结果队列
    wg          sync.WaitGroup
}

// NewWorkerPool 创建工作者池
func NewWorkerPool(workerCount int, queueSize int) *WorkerPool {
    return &WorkerPool{
        workerCount: workerCount,
        taskQueue:   make(chan *Task, queueSize),    // 带缓冲的任务队列
        resultQueue: make(chan *TaskResult, queueSize),
    }
}

// Start 启动工作者池
func (wp *WorkerPool) Start() {
    fmt.Printf("[WorkerPool] 启动 %d 个工作者\n", wp.workerCount)

    for i := 0; i < wp.workerCount; i++ {
        wp.wg.Add(1)
        go wp.worker(i)
    }
}

// worker 工作者 goroutine
func (wp *WorkerPool) worker(id int) {
    defer wp.wg.Done()
    fmt.Printf("[Worker %d] 工作者启动\n", id)

    for task := range wp.taskQueue {
        fmt.Printf("[Worker %d] 接收任务 #%d: %s\n", id, task.ID, task.Data)

        // 模拟处理任务
        start := time.Now()
        time.Sleep(time.Duration(100+rand.Intn(200)) * time.Millisecond)
        duration := time.Since(start)

        // 返回结果
        result := &TaskResult{
            TaskID:   task.ID,
            Result:   fmt.Sprintf("Worker %d 处理完成: %s", id, task.Data),
            Duration: duration,
        }

        // 非阻塞发送结果
        select {
        case wp.resultQueue <- result:
            fmt.Printf("[Worker %d] 任务 #%d 结果已发送\n", id, task.ID)
        default:
            fmt.Printf("[Worker %d] ⚠️ 结果队列满，丢弃任务 #%d 结果\n", id, task.ID)
        }
    }

    fmt.Printf("[Worker %d] 工作者退出\n", id)
}

// Submit 提交任务
func (wp *WorkerPool) Submit(task *Task) bool {
    select {
    case wp.taskQueue <- task:
        fmt.Printf("[Pool] 任务 #%d 已提交\n", task.ID)
        return true
    default:
        fmt.Printf("[Pool] ⚠️ 任务队列满，任务 #%d 被拒绝\n", task.ID)
        return false
    }
}

// SubmitAndWait 提交任务并等待结果
func (wp *WorkerPool) SubmitAndWait(task *Task) *TaskResult {
    wp.Submit(task)

    // 等待结果
    select {
    case result := <-wp.resultQueue:
        return result
    case <-time.After(10 * time.Second):
        return nil
    }
}

// Shutdown 关闭工作者池
func (wp *WorkerPool) Shutdown() {
    fmt.Println("[Pool] 关闭工作者池...")
    close(wp.taskQueue) // 关闭任务队列，worker会自动退出
    wp.wg.Wait()
    close(wp.resultQueue)
    fmt.Println("[Pool] 所有工作者已退出")
}

// GetResultQueue 获取结果队列
func (wp *WorkerPool) GetResultQueue() <-chan *TaskResult {
    return wp.resultQueue
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

func main() {
    fmt.Println("=== 工作者池模式 ===\n")

    // 创建工作者池：3个worker，队列容量10
    pool := NewWorkerPool(3, 10)

    // 启动工作者池
    pool.Start()

    // 提交任务
    taskCount := 10
    fmt.Printf("\n提交 %d 个任务...\n\n", taskCount)

    for i := 1; i <= taskCount; i++ {
        task := &Task{
            ID:   i,
            Data: fmt.Sprintf("订单数据-%d", i),
        }
        pool.Submit(task)
        time.Sleep(50 * time.Millisecond) // 模拟任务提交间隔
    }

    // 等待所有结果
    fmt.Println("\n========== 等待结果 ==========")
    resultCount := 0
    for result := range pool.GetResultQueue() {
        resultCount++
        fmt.Printf("✅ 任务 #%d 完成，耗时: %v，结果: %s\n",
            result.TaskID, result.Duration, result.Result)
        if resultCount >= taskCount {
            break
        }
    }

    // 关闭工作者池
    pool.Shutdown()
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39