Go语言并发编程高级特性

func sayHello() {
    fmt.Println("Hello, goroutine!")
}

func main() {
    go sayHello()  // 创建一个goroutine，异步执行sayHello函数
    fmt.Println("Hello, main!")
    time.Sleep(time.Second)  // 等待goroutine执行完毕
}

// 创建一个无缓冲的channel
b := make(chan int)

// 创建一个带缓冲的channel，缓冲大小为10
buf := make(chan int, 10)

ch := make(chan int)
go func() {
    ch <- 42  // 向channel发送数据
}()

ch := make(chan int)
go func() {
    ch <- 42
}()

value := <-ch  // 从channel接收数据
fmt.Println(value)  // 输出：42

value, ok := <-ch
if ok {
    fmt.Printf("Received value: %d\n", value)
} else {
    fmt.Println("Channel is closed")
}

ch := make(chan int)
go func() {
    for i := 0; i < 5; i++ {
        ch <- i
    }
    close(ch)  // 关闭channel
}()

// 从channel接收数据，直到channel关闭
for value := range ch {
    fmt.Println(value)
}

func main() {
    ch := make(chan int)
    
    go func() {
        fmt.Println("Sender: Sending data...")
        ch <- 42  // 阻塞，直到有goroutine接收数据
        fmt.Println("Sender: Data sent")
    }()
    
    time.Sleep(time.Second)  // 等待一段时间
    
    fmt.Println("Receiver: Waiting for data...")
    value := <-ch  // 阻塞，直到有goroutine发送数据
    fmt.Printf("Receiver: Received data: %d\n", value)
}

func main() {
    ch := make(chan int, 2)  // 创建一个带缓冲的channel，缓冲大小为2
    
    // 发送数据到channel，缓冲区未满，不会阻塞
    ch <- 1
    ch <- 2
    fmt.Println("Sent 1 and 2")
    
    // 缓冲区已满，发送操作会阻塞
    // ch <- 3  // 取消注释会导致程序阻塞
    
    // 接收数据，缓冲区非空，不会阻塞
    fmt.Println(<-ch)  // 输出：1
    fmt.Println(<-ch)  // 输出：2
    
    // 缓冲区为空，接收操作会阻塞
    // fmt.Println(<-ch)  // 取消注释会导致程序阻塞
}

ch := make(chan int, 10)  // 创建一个带缓冲的channel，缓冲大小为10

ch := make(chan int)

// 创建一个只能发送的channel
sendCh := chan<- int(ch)

// 创建一个只能接收的channel
recvCh := <-chan int(ch)

ch := make(chan int)  // 双向channel

sendCh := chan<- int(ch)  // 转换为只能发送的channel
recvCh := <-chan int(ch)  // 转换为只能接收的channel

// 错误：不能将只能发送的channel转换为双向channel
// ch2 := chan int(sendCh)

// 错误：不能将只能接收的channel转换为双向channel
// ch3 := chan int(recvCh)

select {
case <-ch1:
    // 处理从ch1接收数据的情况
case data := <-ch2:
    // 处理从ch2接收数据的情况
case ch3 <- value:
    // 处理向ch3发送数据的情况
case <-time.After(time.Second):
    // 处理超时的情况
default:
    // 当所有的case都不可执行时，执行default分支
}

func main() {
    ch := make(chan int)
    
    go func() {
        time.Sleep(2 * time.Second)  // 模拟长时间操作
        ch <- 42
    }()
    
    select {
    case value := <-ch:
        fmt.Printf("Received value: %d\n", value)
    case <-time.After(1 * time.Second):
        fmt.Println("Timeout")
    }
}

func main() {
    ch := make(chan int)
    
    // 非阻塞发送
    select {
    case ch <- 42:
        fmt.Println("Sent data")
    default:
        fmt.Println("Cannot send data")
    }
    
    // 非阻塞接收
    select {
    case value := <-ch:
        fmt.Printf("Received value: %d\n", value)
    default:
        fmt.Println("Cannot receive data")
    }
}

func main() {
    ch1 := make(chan int)
    ch2 := make(chan string)
    
    go func() {
        time.Sleep(1 * time.Second)
        ch1 <- 42
    }()
    
    go func() {
        time.Sleep(2 * time.Second)
        ch2 <- "hello"
    }()
    
    // 同时监听两个channel
    for i := 0; i < 2; i++ {
        select {
        case value := <-ch1:
            fmt.Printf("Received from ch1: %d\n", value)
        case value := <-ch2:
            fmt.Printf("Received from ch2: %s\n", value)
        }
    }
}

var counter int
var mutex sync.Mutex

func increment() {
    mutex.Lock()  // 获取锁
    defer mutex.Unlock()  // 确保锁被释放
    counter++
}

func main() {
    var wg sync.WaitGroup
    
    // 创建1000个goroutine，每个goroutine都调用increment函数
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            increment()
        }()
    }
    
    wg.Wait()  // 等待所有goroutine执行完毕
    fmt.Printf("Final counter value: %d\n", counter)  // 输出：1000
}

var data map[string]string
var rwMutex sync.RWMutex

func readData(key string) string {
    rwMutex.RLock()  // 获取读锁
    defer rwMutex.RUnlock()  // 确保读锁被释放
    return data[key]
}

func writeData(key, value string) {
    rwMutex.Lock()  // 获取写锁
    defer rwMutex.Unlock()  // 确保写锁被释放
    data[key] = value
}

func main() {
    data = make(map[string]string)
    
    // 写入一些数据
    writeData("key1", "value1")
    writeData("key2", "value2")
    
    var wg sync.WaitGroup
    
    // 创建10个goroutine读取数据
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            fmt.Println(readData("key1"))
        }()
    }
    
    // 创建2个goroutine写入数据
    for i := 0; i < 2; i++ {
        wg.Add(1)
        go func(i int) {
            defer wg.Done()
            writeData(fmt.Sprintf("key%d", i+3), fmt.Sprintf("value%d", i+3))
        }(i)
    }
    
    wg.Wait()  // 等待所有goroutine执行完毕
}

func main() {
    var wg sync.WaitGroup
    
    // 创建10个goroutine
    for i := 0; i < 10; i++ {
        wg.Add(1)  // 增加等待组的计数
        go func(id int) {
            defer wg.Done()  // 减少等待组的计数
            fmt.Printf("Goroutine %d is running\n", id)
            time.Sleep(time.Second)  // 模拟工作
            fmt.Printf("Goroutine %d is done\n", id)
        }(i)
    }
    
    fmt.Println("Waiting for all goroutines to finish...")
    wg.Wait()  // 等待所有goroutine执行完毕
    fmt.Println("All goroutines are done")
}

var queue []int
var mutex sync.Mutex
var notEmpty = sync.NewCond(&mutex)
var notFull = sync.NewCond(&mutex)
const maxQueueSize = 5

func producer() {
    for i := 0; i < 10; i++ {
        mutex.Lock()
        
        // 等待队列不满
        for len(queue) == maxQueueSize {
            notFull.Wait()
        }
        
        // 将数据添加到队列
        queue = append(queue, i)
        fmt.Printf("Produced: %d, Queue: %v\n", i, queue)
        
        mutex.Unlock()
        
        // 通知消费者队列不为空
        notEmpty.Signal()
        
        time.Sleep(time.Millisecond * 500)  // 模拟生产过程
    }
}

func consumer() {
    for i := 0; i < 10; i++ {
        mutex.Lock()
        
        // 等待队列不为空
        for len(queue) == 0 {
            notEmpty.Wait()
        }
        
        // 从队列中取出数据
        data := queue[0]
        queue = queue[1:]
        fmt.Printf("Consumed: %d, Queue: %v\n", data, queue)
        
        mutex.Unlock()
        
        // 通知生产者队列不满
        notFull.Signal()
        
        time.Sleep(time.Millisecond * 1000)  // 模拟消费过程
    }
}

func main() {
    go producer()
    go consumer()
    
    time.Sleep(time.Second * 10)  // 等待生产者和消费者执行完毕
}

var counter int32

func increment() {
    atomic.AddInt32(&counter, 1)  // 原子地增加counter的值
}

func main() {
    var wg sync.WaitGroup
    
    // 创建1000个goroutine，每个goroutine都调用increment函数
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            increment()
        }()
    }
    
    wg.Wait()  // 等待所有goroutine执行完毕
    fmt.Printf("Final counter value: %d\n", atomic.LoadInt32(&counter))  // 输出：1000
}

var m sync.Map

func main() {
    var wg sync.WaitGroup
    
    // 创建10个goroutine写入数据
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(i int) {
            defer wg.Done()
            m.Store(fmt.Sprintf("key%d", i), fmt.Sprintf("value%d", i))
        }(i)
    }
    
    wg.Wait()  // 等待所有写入goroutine执行完毕
    
    // 创建10个goroutine读取数据
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(i int) {
            defer wg.Done()
            value, ok := m.Load(fmt.Sprintf("key%d", i))
            if ok {
                fmt.Printf("key%d: %s\n", i, value)
            }
        }(i)
    }
    
    wg.Wait()  // 等待所有读取goroutine执行完毕
    
    // 遍历sync.Map
    m.Range(func(key, value interface{}) bool {
        fmt.Printf("%s: %s\n", key, value)
        return true  // 返回true继续遍历，返回false停止遍历
    })
}

var bufferPool = sync.Pool{
    New: func() interface{} {
        // 创建一个新的缓冲区
        return make([]byte, 1024)
    },
}

func processData(data []byte) {
    // 从对象池获取一个缓冲区
    buffer := bufferPool.Get().([]byte)
    defer bufferPool.Put(buffer)  // 使用完毕后归还缓冲区
    
    // 处理数据...
    copy(buffer, data)
    // ...
}

func main() {
    for i := 0; i < 1000; i++ {
        data := make([]byte, 512)  // 模拟需要处理的数据
        processData(data)
    }
}

type Context interface {
    Deadline() (deadline time.Time, ok bool)
    Done() <-chan struct{}
    Err() error
    Value(key interface{}) interface{}
}

func main() {
    // 创建根上下文
    ctx := context.Background()
    
    // 创建可取消的上下文
    cancelCtx, cancel := context.WithCancel(ctx)
    defer cancel()  // 确保在函数结束时取消上下文
    
    // 创建带有超时时间的上下文
    timeoutCtx, timeoutCancel := context.WithTimeout(ctx, 5*time.Second)
    defer timeoutCancel()  // 确保在函数结束时取消上下文
    
    // 创建带有值的上下文
    valueCtx := context.WithValue(ctx, "key", "value")
    
    // 使用上下文
    go processWithContext(cancelCtx)
    go processWithContext(timeoutCtx)
    go processWithContext(valueCtx)
    
    time.Sleep(10 * time.Second)  // 等待一段时间
}

func processWithContext(ctx context.Context) {
    // 检查上下文是否被取消
    select {
    case <-ctx.Done():
        fmt.Printf("Context canceled: %v\n", ctx.Err())
        return
    default:
        // 上下文未被取消，继续处理
    }
    
    // 获取上下文中的值
    if value := ctx.Value("key"); value != nil {
        fmt.Printf("Context value: %v\n", value)
    }
    
    // 模拟处理过程
    time.Sleep(2 * time.Second)
    
    // 再次检查上下文是否被取消
    select {
    case <-ctx.Done():
        fmt.Printf("Context canceled during processing: %v\n", ctx.Err())
        return
    default:
        // 上下文未被取消，处理完成
        fmt.Println("Processing completed")
    }
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    
    go func() {
        // 模拟用户取消操作
        time.Sleep(3 * time.Second)
        fmt.Println("User canceled the operation")
        cancel()
    }()
    
    // 执行长时间运行的操作
    err := longRunningOperation(ctx)
    if err != nil {
        fmt.Printf("Operation failed: %v\n", err)
    }
}

func longRunningOperation(ctx context.Context) error {
    for i := 0; i < 10; i++ {
        select {
        case <-ctx.Done():
            return ctx.Err()  // 上下文被取消，返回错误
        default:
            // 继续处理
            fmt.Printf("Processing step %d\n", i+1)
            time.Sleep(time.Second)  // 模拟处理过程
        }
    }
    return nil
}

func main() {
    // 创建一个带有5秒超时的上下文
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()
    
    // 执行可能超时的操作
    err := operationWithTimeout(ctx)
    if err != nil {
        fmt.Printf("Operation failed: %v\n", err)
    }
}

func operationWithTimeout(ctx context.Context) error {
    // 创建一个通道来接收操作结果
    resultCh := make(chan error, 1)
    
    go func() {
        // 模拟长时间运行的操作
        time.Sleep(10 * time.Second)  // 操作需要10秒，但超时时间为5秒
        resultCh <- nil
    }()
    
    // 等待操作完成或超时
    select {
    case result := <-resultCh:
        return result
    case <-ctx.Done():
        return ctx.Err()  // 上下文超时，返回错误
    }
}

func main() {
    // 创建一个带有请求ID的上下文
    ctx := context.WithValue(context.Background(), "requestID", "123456")
    
    // 处理请求
    handleRequest(ctx)
}

func handleRequest(ctx context.Context) {
    // 获取请求ID
    requestID, ok := ctx.Value("requestID").(string)
    if ok {
        fmt.Printf("Handling request with ID: %s\n", requestID)
    }
    
    // 调用其他函数处理请求
    processRequest(ctx)
}

func processRequest(ctx context.Context) {
    // 获取请求ID
    requestID, ok := ctx.Value("requestID").(string)
    if ok {
        fmt.Printf("Processing request with ID: %s\n", requestID)
    }
    
    // 处理请求...
}

type Task struct {
    ID      int
    Payload interface{}
}

func worker(id int, tasks <-chan Task, results chan<- int) {
    for task := range tasks {
        fmt.Printf("Worker %d processing task %d\n", id, task.ID)
        // 模拟处理任务
        time.Sleep(time.Millisecond * 500)
        // 将结果发送到结果通道
        results <- task.ID * 2
    }
}

func main() {
    const numWorkers = 5
    const numTasks = 10
    
    // 创建任务通道和结果通道
    tasks := make(chan Task, numTasks)
    results := make(chan int, numTasks)
    
    // 启动工作者
    for i := 0; i < numWorkers; i++ {
        go worker(i+1, tasks, results)
    }
    
    // 提交任务
    for i := 0; i < numTasks; i++ {
        tasks <- Task{
            ID:      i + 1,
            Payload: fmt.Sprintf("data-%d", i+1),
        }
    }
    close(tasks)  // 关闭任务通道，表示没有更多任务
    
    // 收集结果
    for i := 0; i < numTasks; i++ {
        result := <-results
        fmt.Printf("Received result: %d\n", result)
    }
    close(results)  // 关闭结果通道
}

type DynamicWorkerPool struct {
    tasks        chan Task
    results      chan int
    maxWorkers   int
    activeWorkers int32
    wg           sync.WaitGroup
    mu           sync.Mutex
}

func NewDynamicWorkerPool(maxWorkers int, bufferSize int) *DynamicWorkerPool {
    return &DynamicWorkerPool{
        tasks:      make(chan Task, bufferSize),
        results:    make(chan int, bufferSize),
        maxWorkers: maxWorkers,
    }
}

func (p *DynamicWorkerPool) Start() {
    // 启动一些初始工作者
    for i := 0; i < min(p.maxWorkers, 5); i++ {  // 初始启动5个工作者或maxWorkers（取较小值）
        p.startWorker(i + 1)
    }
    
    // 启动一个监控goroutine，根据任务数量动态调整工作者数量
    go p.monitor()
}

func (p *DynamicWorkerPool) startWorker(id int) {
    p.mu.Lock()
    defer p.mu.Unlock()
    
    if p.activeWorkers >= int32(p.maxWorkers) {
        return  // 已达到最大工作者数量
    }
    
    atomic.AddInt32(&p.activeWorkers, 1)
    p.wg.Add(1)
    
    go func() {
        defer func() {
            atomic.AddInt32(&p.activeWorkers, -1)
            p.wg.Done()
        }()
        
        for task := range p.tasks {
            fmt.Printf("Worker %d processing task %d\n", id, task.ID)
            // 模拟处理任务
            time.Sleep(time.Millisecond * 500)
            // 将结果发送到结果通道
            p.results <- task.ID * 2
        }
    }()
}

func (p *DynamicWorkerPool) monitor() {
    ticker := time.NewTicker(time.Second)  // 每秒检查一次
    defer ticker.Stop()
    
    for range ticker.C {
        taskCount := len(p.tasks)
        workerCount := atomic.LoadInt32(&p.activeWorkers)
        
        fmt.Printf("Tasks: %d, Workers: %d\n", taskCount, workerCount)
        
        // 如果任务数量大于工作者数量的2倍，且未达到最大工作者数量，则增加工作者
        if taskCount > int(workerCount)*2 && workerCount < int32(p.maxWorkers) {
            neededWorkers := min(p.maxWorkers, int(workerCount)+taskCount/2)
            for i := int(workerCount); i < neededWorkers; i++ {
                p.startWorker(i + 1)
            }
        }
    }
}

func (p *DynamicWorkerPool) Submit(task Task) {
    p.tasks <- task
}

func (p *DynamicWorkerPool) Stop() {
    close(p.tasks)  // 关闭任务通道，不再接受新任务
    p.wg.Wait()     // 等待所有工作者处理完任务
    close(p.results)  // 关闭结果通道
}

func main() {
    pool := NewDynamicWorkerPool(20, 100)  // 创建一个最大工作者数量为20的动态工作池
    pool.Start()
    
    // 提交100个任务
    for i := 0; i < 100; i++ {
        pool.Submit(Task{
            ID:      i + 1,
            Payload: fmt.Sprintf("data-%d", i+1),
        })
        time.Sleep(time.Millisecond * 10)  // 稍微延迟提交任务，以便观察动态调整
    }
    
    // 等待一段时间，让工作池有时间处理任务
    time.Sleep(time.Second * 10)
    
    // 停止工作池
    pool.Stop()
    
    // 收集结果
    for result := range pool.results {
        fmt.Printf("Received result: %d\n", result)
    }
}

func main() {
    ch := make(chan int)  // 无缓冲的channel
    var wg sync.WaitGroup
    
    // 启动生产者
    wg.Add(1)
    go func() {
        defer wg.Done()
        for i := 0; i < 10; i++ {
            fmt.Printf("Producer: producing %d\n", i)
            ch <- i  // 发送数据到channel
            fmt.Printf("Producer: produced %d\n", i)
            time.Sleep(time.Millisecond * 300)  // 模拟生产过程
        }
        close(ch)  // 关闭channel，表示没有更多数据
    }()
    
    // 启动消费者
    wg.Add(1)
    go func() {
        defer wg.Done()
        for data := range ch {
            fmt.Printf("Consumer: consuming %d\n", data)
            time.Sleep(time.Millisecond * 500)  // 模拟消费过程
        }
    }()
    
    wg.Wait()  // 等待生产者和消费者执行完毕
}

func main() {
    ch := make(chan int, 5)  // 带缓冲的channel，缓冲区大小为5
    var wg sync.WaitGroup
    
    // 启动生产者
    wg.Add(1)
    go func() {
        defer wg.Done()
        for i := 0; i < 10; i++ {
            fmt.Printf("Producer: producing %d\n", i)
            ch <- i  // 发送数据到channel，如果缓冲区未满，则不会阻塞
            fmt.Printf("Producer: produced %d\n", i)
            time.Sleep(time.Millisecond * 300)  // 模拟生产过程
        }
        close(ch)  // 关闭channel，表示没有更多数据
    }()
    
    // 启动消费者
    wg.Add(1)
    go func() {
        defer wg.Done()
        for data := range ch {
            fmt.Printf("Consumer: consuming %d\n", data)
            time.Sleep(time.Millisecond * 500)  // 模拟消费过程
        }
    }()
    
    wg.Wait()  // 等待生产者和消费者执行完毕
}

func main() {
    ch := make(chan int, 10)  // 带缓冲的channel，缓冲区大小为10
    var wg sync.WaitGroup
    
    // 启动3个生产者
    for i := 0; i < 3; i++ {
        wg.Add(1)
        go func(producerID int) {
            defer wg.Done()
            for j := 0; j < 5; j++ {
                data := producerID*100 + j
                fmt.Printf("Producer %d: producing %d\n", producerID, data)
                ch <- data
                fmt.Printf("Producer %d: produced %d\n", producerID, data)
                time.Sleep(time.Millisecond * 300)  // 模拟生产过程
            }
        }(i)
    }
    
    // 启动一个goroutine，等待所有生产者完成后关闭channel
    go func() {
        wg.Wait()  // 等待所有生产者完成
        close(ch)  // 关闭channel，表示没有更多数据
    }()
    
    // 重置等待组，用于等待消费者
    wg = sync.WaitGroup{}
    
    // 启动2个消费者
    for i := 0; i < 2; i++ {
        wg.Add(1)
        go func(consumerID int) {
            defer wg.Done()
            for data := range ch {
                fmt.Printf("Consumer %d: consuming %d\n", consumerID, data)
                time.Sleep(time.Millisecond * 500)  // 模拟消费过程
            }
        }(i)
    }
    
    wg.Wait()  // 等待所有消费者执行完毕
}

type Broker struct {
    subscribers map[string][]chan string
    mu          sync.RWMutex
}

func NewBroker() *Broker {
    return &Broker{
        subscribers: make(map[string][]chan string),
    }
}

// Subscribe 订阅一个主题
func (b *Broker) Subscribe(topic string) <-chan string {
    ch := make(chan string, 10)  // 带缓冲的channel
    
    b.mu.Lock()
    defer b.mu.Unlock()
    
    b.subscribers[topic] = append(b.subscribers[topic], ch)
    
    return ch
}

// Unsubscribe 取消订阅一个主题
func (b *Broker) Unsubscribe(topic string, ch <-chan string) {
    b.mu.Lock()
    defer b.mu.Unlock()
    
    if subscribers, ok := b.subscribers[topic]; ok {
        for i, subscriber := range subscribers {
            if subscriber == ch {
                // 从切片中删除该订阅者
                subscribers[i] = subscribers[len(subscribers)-1]
                b.subscribers[topic] = subscribers[:len(subscribers)-1]
                close(subscriber)
                break
            }
        }
        
        // 如果没有订阅者了，删除该主题
        if len(b.subscribers[topic]) == 0 {
            delete(b.subscribers, topic)
        }
    }
}

// Publish 发布一条消息到指定主题
func (b *Broker) Publish(topic string, message string) {
    b.mu.RLock()
    subscribers, ok := b.subscribers[topic]
    b.mu.RUnlock()
    
    if ok {
        // 向所有订阅者发送消息
        for _, subscriber := range subscribers {
            select {
            case subscriber <- message:
                // 消息发送成功
            default:
                // 缓冲区满，丢弃消息
                fmt.Printf("Warning: buffer full for subscriber, message dropped\n")
            }
        }
    }
}

func main() {
    broker := NewBroker()
    
    // 订阅者1订阅"news"主题
    newsCh1 := broker.Subscribe("news")
    defer broker.Unsubscribe("news", newsCh1)
    
    // 订阅者2订阅"news"主题
    newsCh2 := broker.Subscribe("news")
    defer broker.Unsubscribe("news", newsCh2)
    
    // 订阅者3订阅"tech"主题
    techCh := broker.Subscribe("tech")
    defer broker.Unsubscribe("tech", techCh)
    
    // 启动订阅者goroutine
    var wg sync.WaitGroup
    
    wg.Add(3)
    
    // 订阅者1处理消息
    go func() {
        defer wg.Done()
        for message := range newsCh1 {
            fmt.Printf("Subscriber 1 received news: %s\n", message)
        }
    }()
    
    // 订阅者2处理消息
    go func() {
        defer wg.Done()
        for message := range newsCh2 {
            fmt.Printf("Subscriber 2 received news: %s\n", message)
        }
    }()
    
    // 订阅者3处理消息
    go func() {
        defer wg.Done()
        for message := range techCh {
            fmt.Printf("Subscriber 3 received tech: %s\n", message)
        }
    }()
    
    // 发布者发布消息
    broker.Publish("news", "Breaking news: Go 1.18 released!")
    broker.Publish("tech", "New technology trends for 2023")
    broker.Publish("news", "Another news update")
    
    // 等待一段时间，让订阅者有时间处理消息
    time.Sleep(time.Second)
    
    // 取消订阅者1的订阅
    broker.Unsubscribe("news", newsCh1)
    
    // 再次发布消息
    broker.Publish("news", "This message should only be received by subscriber 2")
    
    // 等待一段时间，让订阅者有时间处理消息
    time.Sleep(time.Second)
    
    fmt.Println("All messages sent, waiting for subscribers to finish...")
    wg.Wait()  // 等待所有订阅者处理完毕
}

func fanOut(tasks <-chan Task, numWorkers int) []<-chan Result {
    resultChannels := make([]<-chan Result, 0, numWorkers)
    
    for i := 0; i < numWorkers; i++ {
        resultCh := make(chan Result)
        resultChannels = append(resultChannels, resultCh)
        
        go func() {
            defer close(resultCh)
            for task := range tasks {
                // 处理任务
                result := processTask(task)
                resultCh <- result
            }
        }()
    }
    
    return resultChannels
}

func fanIn(resultChannels []<-chan Result) <-chan Result {
    var wg sync.WaitGroup
    mergedResults := make(chan Result)
    
    // 为每个结果channel启动一个goroutine
    wg.Add(len(resultChannels))
    for _, resultCh := range resultChannels {
        go func(ch <-chan Result) {
            defer wg.Done()
            for result := range ch {
                mergedResults <- result
            }
        }(resultCh)
    }
    
    // 启动一个goroutine，等待所有结果channel处理完毕后关闭mergedResults
    go func() {
        wg.Wait()
        close(mergedResults)
    }()
    
    return mergedResults
}

type Task struct {
    ID     int
    Number int
}

type Result struct {
    TaskID int
    Square int
}

func processTask(task Task) Result {
    // 模拟处理任务
    time.Sleep(time.Millisecond * 100)
    return Result{
        TaskID: task.ID,
        Square: task.Number * task.Number,
    }
}

func main() {
    const numTasks = 100
    const numWorkers = 10
    
    // 创建任务channel
    tasks := make(chan Task, numTasks)
    
    // 提交任务
    for i := 0; i < numTasks; i++ {
        tasks <- Task{
            ID:     i,
            Number: i + 1,
        }
    }
    close(tasks)  // 关闭任务channel，表示没有更多任务
    
    // Fan-Out：启动工作者并行处理任务
    startTime := time.Now()
    resultChannels := make([]<-chan Result, 0, numWorkers)
    for i := 0; i < numWorkers; i++ {
        resultCh := make(chan Result)
        resultChannels = append(resultChannels, resultCh)
        
        go func() {
            defer close(resultCh)
            for task := range tasks {
                result := processTask(task)
                resultCh <- result
            }
        }()
    }
    
    // Fan-In：合并所有工作者的结果
    var wg sync.WaitGroup
    mergedResults := make(chan Result)
    
    wg.Add(len(resultChannels))
    for _, resultCh := range resultChannels {
        go func(ch <-chan Result) {
            defer wg.Done()
            for result := range ch {
                mergedResults <- result
            }
        }(resultCh)
    }
    
    go func() {
        wg.Wait()
        close(mergedResults)
    }()
    
    // 计算结果
    sum := 0
    count := 0
    for result := range mergedResults {
        sum += result.Square
        count++
    }
    
    duration := time.Since(startTime)
    fmt.Printf("Processed %d tasks in %v\n", count, duration)
    fmt.Printf("Sum of squares from 1 to %d: %d\n", numTasks, sum)
    
    // 验证结果（1²+2²+...+n² = n(n+1)(2n+1)/6）
    expectedSum := numTasks * (numTasks + 1) * (2*numTasks + 1) / 6
    fmt.Printf("Expected sum: %d\n", expectedSum)
    fmt.Printf("Result is %s\n", map[bool]string{true: "correct", false: "incorrect"}[sum == expectedSum])
}

func main() {
    errs := make(chan error, 10)  // 用于收集错误的channel
    var wg sync.WaitGroup
    
    // 启动多个goroutine
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            if err := doWork(id); err != nil {
                errs <- fmt.Errorf("worker %d error: %w", id, err)
            }
        }(i)
    }
    
    // 启动一个goroutine，等待所有工作者完成后关闭错误channel
    go func() {
        wg.Wait()
        close(errs)
    }()
    
    // 收集和处理错误
    for err := range errs {
        fmt.Printf("Error: %v\n", err)
    }
}

func doWork(id int) error {
    // 模拟工作过程，随机产生错误
    time.Sleep(time.Millisecond * 100)
    if id%3 == 0 {
        return fmt.Errorf("random failure")
    }
    return nil
}

### 18.2 errgroup包

Go语言的扩展包`golang.org/x/sync/errgroup`提供了一种更方便的方式来管理一组goroutine并处理它们的错误。`errgroup.Group`结构体封装了等待组和错误处理的逻辑，可以帮助我们更简洁地编写并发代码。

```go
import (
    "fmt"
    "time"
    
    "golang.org/x/sync/errgroup"
)

func main() {
    g, ctx := errgroup.WithContext(context.Background())
    
    // 启动多个goroutine
    for i := 0; i < 10; i++ {
        id := i  // 捕获循环变量
        g.Go(func() error {
            // 检查上下文是否被取消
            select {
            case <-ctx.Done():
                return ctx.Err()
            default:
                // 继续执行
            }
            
            // 执行任务
            err := doWorkWithContext(ctx, id)
            if err != nil {
                // 如果有一个goroutine返回错误，其他goroutine也会被取消
                return fmt.Errorf("worker %d error: %w", id, err)
            }
            return nil
        })
    }
    
    // 等待所有goroutine执行完毕，并获取第一个返回的错误
    if err := g.Wait(); err != nil {
        fmt.Printf("Error: %v\n", err)
    } else {
        fmt.Println("All workers completed successfully")
    }
}

func doWorkWithContext(ctx context.Context, id int) error {
    // 模拟工作过程
    select {
    case <-time.After(time.Millisecond * 100):
        // 工作完成
        if id%3 == 0 {
            return fmt.Errorf("random failure")
        }
        return nil
    case <-ctx.Done():
        // 上下文被取消
        return ctx.Err()
    }
}

var counter int

func main() {
    var wg sync.WaitGroup
    
    // 创建1000个goroutine，每个goroutine都增加计数器的值
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            counter++  // 竞态条件：多个goroutine同时修改counter
        }()
    }
    
    wg.Wait()  // 等待所有goroutine执行完毕
    fmt.Printf("Final counter value: %d\n", counter)  // 可能不是1000
}

go run -race main.go

var counter int
var mutex sync.Mutex

func increment() {
    mutex.Lock()
    defer mutex.Unlock()
    counter++
}

var data map[string]string
var rwMutex sync.RWMutex

func readData(key string) string {
    rwMutex.RLock()
    defer rwMutex.RUnlock()
    return data[key]
}

func writeData(key, value string) {
    rwMutex.Lock()
    defer rwMutex.Unlock()
    data[key] = value
}

var counter int32

func increment() {
    atomic.AddInt32(&counter, 1)
}

func main() {
    counterChan := make(chan int)
    resultChan := make(chan int)
    
    // 启动一个goroutine来维护计数器
    go func() {
        counter := 0
        for {
            select {
            case <-counterChan:
                counter++
            case resultChan <- counter:
                // 发送计数器的值
            }
        }
    }()
    
    var wg sync.WaitGroup
    
    // 创建1000个goroutine，每个goroutine都通过channel增加计数器的值
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            counterChan <- 1  // 通过channel增加计数器的值
        }()
    }
    
    wg.Wait()  // 等待所有goroutine执行完毕
    
    // 获取最终的计数器值
    finalCounter := <-resultChan
    fmt.Printf("Final counter value: %d\n", finalCounter)  // 应该是1000
}

// 不好的做法：使用一个全局锁来保护所有资源
var globalMutex sync.Mutex
var resourceA, resourceB int

// 好的做法：使用多个锁来保护不同的资源
var mutexA, mutexB sync.Mutex
var resourceA, resourceB int

func updateResources() {
    // 只在必要的时间持有锁
    mutexA.Lock()
    resourceA++
    mutexA.Unlock()
    
    // 执行不需要锁的操作
    time.Sleep(time.Millisecond)
    
    mutexB.Lock()
    resourceB++
    mutexB.Unlock()
}

var counter int64

func increment() {
    atomic.AddInt64(&counter, 1)
}

func getCounter() int64 {
    return atomic.LoadInt64(&counter)
}

var bufferPool = sync.Pool{
    New: func() interface{} {
        return make([]byte, 4096)  // 创建一个4KB的缓冲区
    },
}

func processData(data []byte) {
    // 从对象池获取缓冲区
    buffer := bufferPool.Get().([]byte)
    defer bufferPool.Put(buffer)  // 使用完毕后归还缓冲区
    
    // 确保缓冲区足够大
    if cap(buffer) < len(data) {
        buffer = make([]byte, len(data))  // 如果缓冲区不够大，创建一个新的
    } else {
        buffer = buffer[:len(data)]  // 调整缓冲区的长度
    }
    
    // 处理数据
    copy(buffer, data)
    // ...
}

import (
    _ "net/http/pprof"
    "net/http"
)

func main() {
    // 启动pprof HTTP服务器
    go func() {
        http.ListenAndServe("localhost:6060", nil)
    }()
    
    // 程序的主要逻辑
    // ...
}

# 分析CPU性能
go tool pprof http://localhost:6060/debug/pprof/profile

# 分析内存性能
go tool pprof http://localhost:6060/debug/pprof/heap

# 分析锁竞争
go tool pprof http://localhost:6060/debug/pprof/mutex

func TestCounter_Concurrent(t *testing.T) {
    counter := NewCounter()
    var wg sync.WaitGroup
    
    // 启动1000个goroutine，每个goroutine都增加计数器的值100次
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for j := 0; j < 100; j++ {
                counter.Increment()
            }
        }()
    }
    
    wg.Wait()  // 等待所有goroutine执行完毕
    
    // 验证计数器的值是否正确
    expected := int64(1000 * 100)
    actual := counter.Get()
    if actual != expected {
        t.Errorf("Expected counter value %d, but got %d", expected, actual)
    }
}

// Counter 是一个并发安全的计数器
type Counter struct {
    value int64
}

// NewCounter 创建一个新的计数器
func NewCounter() *Counter {
    return &Counter{}
}

// Increment 增加计数器的值
func (c *Counter) Increment() {
    atomic.AddInt64(&c.value, 1)
}

// Get 获取计数器的当前值
func (c *Counter) Get() int64 {
    return atomic.LoadInt64(&c.value)
}

go test -race

// 有竞态条件的计数器
type BadCounter struct {
    value int
}

// Increment 增加计数器的值（有竞态条件）
func (c *BadCounter) Increment() {
    c.value++
}

// Get 获取计数器的当前值（有竞态条件）
func (c *BadCounter) Get() int {
    return c.value
}

func TestBadCounter_Race(t *testing.T) {
    counter := &BadCounter{}
    var wg sync.WaitGroup
    
    // 启动多个goroutine，同时访问计数器
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            counter.Increment()
        }()
    }
    
    wg.Wait()
    t.Logf("Final counter value: %d", counter.Get())
}

package workerpool

import (
    "sync"
    "time"
)

// Task 表示工作池中的一个任务
type Task func()

// WorkerPool 表示一个工作池
type WorkerPool struct {
    tasks        chan Task
    results      chan error
    maxWorkers   int
    activeWorkers int32
    wg           sync.WaitGroup
    mu           sync.Mutex
    stopped      bool
}

// NewWorkerPool 创建一个新的工作池
func NewWorkerPool(maxWorkers int, taskBuffer int) *WorkerPool {
    return &WorkerPool{
        tasks:      make(chan Task, taskBuffer),
        results:    make(chan error, taskBuffer),
        maxWorkers: maxWorkers,
    }
}

// Start 启动工作池
func (p *WorkerPool) Start() {
    p.mu.Lock()
    defer p.mu.Unlock()
    
    if p.stopped {
        return
    }
    
    // 启动初始工作者
    initialWorkers := p.maxWorkers
    if initialWorkers > 10 {
        initialWorkers = 10  // 初始工作者数量不超过10
    }
    
    for i := 0; i < initialWorkers; i++ {
        p.startWorker()
    }
    
    // 启动监控goroutine，根据任务数量动态调整工作者数量
    go p.monitor()
    
    // 启动结果收集goroutine
    go p.collectResults()
}

// startWorker 启动一个工作者
func (p *WorkerPool) startWorker() {
    p.mu.Lock()
    defer p.mu.Unlock()
    
    if p.stopped || p.activeWorkers >= int32(p.maxWorkers) {
        return
    }
    
    atomic.AddInt32(&p.activeWorkers, 1)
    p.wg.Add(1)
    
    go func() {
        defer func() {
            atomic.AddInt32(&p.activeWorkers, -1)
            p.wg.Done()
        }()
        
        for task := range p.tasks {
            if p.stopped {
                return
            }
            
            // 执行任务
            task()
            
            // 发送成功结果
            select {
            case p.results <- nil:
            default:
                // 结果通道已满，忽略结果
            }
        }
    }()
}

// monitor 监控任务队列，动态调整工作者数量
func (p *WorkerPool) monitor() {
    ticker := time.NewTicker(100 * time.Millisecond)
    defer ticker.Stop()
    
    for range ticker.C {
        p.mu.Lock()
        if p.stopped {
            p.mu.Unlock()
            return
        }
        
        taskCount := len(p.tasks)
        workerCount := atomic.LoadInt32(&p.activeWorkers)
        
        // 如果任务数量大于工作者数量的2倍，且未达到最大工作者数量，则增加工作者
        if taskCount > int(workerCount)*2 && workerCount < int32(p.maxWorkers) {
            neededWorkers := min(p.maxWorkers, int(workerCount)+taskCount/2)
            for i := int(workerCount); i < neededWorkers; i++ {
                p.startWorker()
            }
        }
        p.mu.Unlock()
    }
}

// collectResults 收集任务执行结果
func (p *WorkerPool) collectResults() {
    for range p.results {
        // 这里可以处理任务执行结果，例如记录日志、统计成功/失败次数等
    }
}

// Submit 提交一个任务到工作池
func (p *WorkerPool) Submit(task Task) error {
    p.mu.Lock()
    defer p.mu.Unlock()
    
    if p.stopped {
        return fmt.Errorf("worker pool is stopped")
    }
    
    select {
    case p.tasks <- task:
        return nil
    default:
        return fmt.Errorf("task queue is full")
    }
}

// Stop 停止工作池
func (p *WorkerPool) Stop() {
    p.mu.Lock()
    defer p.mu.Unlock()
    
    if p.stopped {
        return
    }
    
    p.stopped = true
    close(p.tasks)  // 关闭任务通道，不再接受新任务
    
    // 等待所有工作者完成
    go func() {
        p.wg.Wait()
        close(p.results)  // 关闭结果通道
    }()
}

// Stats 返回工作池的当前状态
func (p *WorkerPool) Stats() (activeWorkers int32, pendingTasks int) {
    return atomic.LoadInt32(&p.activeWorkers), len(p.tasks)
}

func min(a, b int) int {
    if a < b {
        return a
    }
    return b
}