Queue

package main

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

const (
	queueCount = 16      // 队列数量
	bufferSize = 100     // 每个队列的缓冲区大小
	messageCount = 1000  // 要发送的消息总数
)

func main() {
	// 1. 创建16个缓冲队列
	queues := make([]chan int, queueCount)
	for i := range queues {
		queues[i] = make(chan int, bufferSize)
	}

	// 2. 创建WaitGroup用于等待所有消费者完成
	var wg sync.WaitGroup
	wg.Add(queueCount)

	// 3. 为每个队列启动一个消费者goroutine
	for i, queue := range queues {
		go func(queueID int, ch <-chan int) {
			defer wg.Done()
			processQueue(queueID, ch)
		}(i, queue)
	}

	// 4. 生产者发送消息(使用ID取模路由)
	start := time.Now()
	for id := 0; id < messageCount; id++ {
		queueIndex := id % queueCount  // 取模决定路由到哪个队列
		queues[queueIndex] <- id      // 发送到对应队列
	}

	// 5. 关闭所有队列(通知消费者处理完成)
	for _, queue := range queues {
		close(queue)
	}

	// 6. 等待所有消费者完成
	wg.Wait()

	fmt.Printf("All queues processed. Time elapsed: %v\n", time.Since(start))
}

// 处理队列中的消息
func processQueue(queueID int, ch <-chan int) {
	count := 0
	for msg := range ch {
		// 模拟处理消息(这里简单打印)
		if count%100 == 0 { // 每100条打印一次，避免输出太多
			fmt.Printf("Queue %d processing message %d\n", queueID, msg)
		}
		count++
		time.Sleep(10 * time.Millisecond) // 模拟处理耗时
	}
	fmt.Printf("Queue %d processed %d messages\n", queueID, count)
}

queues := make([]chan int, queueCount)
for i := range queues {
    queues[i] = make(chan int, bufferSize)
}

// 通道未关闭时：for range会持续接收新消息，包括其他goroutine新插入的消息
// 通道关闭后：for range会处理完已缓冲的消息后退出，不会接收关闭后新发送的消息(会panic)
// 遍历是阻塞的：如果没有新消息，for range会阻塞等待
for i, queue := range queues {
    // chan T：双向通道 (可读可写)
    // <-chan T：只读通道 (接收操作)
    // chan<- T：只写通道 (发送操作)
    go func(queueID int, ch <-chan int) {
        defer wg.Done()
        processQueue(queueID, ch)
    }(i, queue)
}

for id := 0; id < messageCount; id++ {
    queueIndex := id % queueCount  // 取模决定路由
    queues[queueIndex] <- id      // 发送到对应队列
}

for _, queue := range queues {
    close(queue)
}

wg.Wait()