Go Channel 手撕题详解（西安市场高频考点）

// 实际项目案例：任务分发器
// 主协程收到请求后，分发到多个 worker 处理
// 这里可以用 channel 实现请求队列和结果收集

package main

import (
	"fmt"
	"sync"
)

// 交替打印奇偶数
func printOddEven(n int) {
	oddCh := make(chan struct{})
	evenCh := make(chan struct{})

	var wg sync.WaitGroup
	wg.Add(2)

	// 奇数打印器
	go func() {
		defer wg.Done()
		for i := 1; i <= n; i += 2 {
			fmt.Printf("%d ", i)
			<-oddCh  // 等待自己的信号
			if i < n {
				evenCh <- struct{}{}  // 通知偶数 goroutine
			}
		}
	}()

	// 偶数打印器
	go func() {
		defer wg.Done()
		for i := 2; i <= n; i += 2 {
			<-evenCh  // 等待自己的信号
			fmt.Printf("%d ", i)
			if i < n {
				oddCh <- struct{}{}  // 通知奇数 goroutine
			}
		}
	}()

	// 启动：先让奇数 goroutine 打印第一个数字
	oddCh <- struct{}{}
	wg.Wait()
	fmt.Println()
}

func main() {
	printOddEven(10) // 输出: 1 2 3 4 5 6 7 8 9 10
}

// 项目案例：消息队列消费者
// 从 Kafka 消费消息时，限制并发处理数，避免数据库连接池耗尽
type MessageHandler struct {
	queue chan Message
	sem   semaphore  // 信号量控制并发
}

package main

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

// BoundedQueue 有界队列
type BoundedQueue struct {
	items    chan interface{}
	empty    chan struct{}
	mu       sync.Mutex
	size     int
}

// NewBoundedQueue 创建有界队列
func NewBoundedQueue(size int) *BoundedQueue {
	bq := &BoundedQueue{
		items:  make(chan interface{}, size),
		empty:  make(chan struct{}, size),
		size:   size,
	}
	// 初始化空槽位
	for i := 0; i < size; i++ {
		bq.empty <- struct{}{}
	}
	return bq
}

// Put 添加元素，队列满时阻塞
func (bq *BoundedQueue) Put(item interface{}) error {
	select {
	case <-bq.empty:
		// 有空槽位，直接放入
		bq.items <- item
		return nil
	case <-time.After(time.Second * 5):
		return fmt.Errorf("timeout waiting for space")
	}
}

// Get 获取元素，超时返回 false
func (bq *BoundedQueue) Get(timeout time.Duration) (interface{}, bool) {
	select {
	case item := <-bq.items:
		bq.empty <- struct{}{}  // 归还空槽位
		return item, true
	case <-time.After(timeout):
		return nil, false
	}
}

// Len 当前队列长度
func (bq *BoundedQueue) Len() int {
	select {
	case item := <-bq.items:
		// 临时取出再放回去
		bq.items <- item
		return len(bq.items)
	default:
		return len(bq.items)
	}
}

func main() {
	bq := NewBoundedQueue(3)
	
	// 生产者
	go func() {
		for i := 0; i < 5; i++ {
			bq.Put(i)
			fmt.Printf("Put %d\n", i)
		}
	}()
	
	// 消费者
	go func() {
		for i := 0; i < 5; i++ {
			item, ok := bq.Get(time.Second * 2)
			if ok {
				fmt.Printf("Get %v\n", item)
			} else {
				fmt.Println("Timeout waiting for item")
			}
		}
	}()
	
	time.Sleep(time.Second * 10)
}

// 项目案例：批量订单处理
// 1. Fan-out：将订单列表分发给 N 个 worker 并发处理
// 2. Worker 内部：每个 worker 再 fan-out 给多个子任务
// 3. Fan-in：收集所有 worker 的结果，生成汇总报告

// Fan-out: 从单一 channel 读取数据分发给多个消费者
func fanOut(data []int, workers int) []<-chan int {
	channels := make([]chan int, workers)
	for i := 0; i < workers; i++ {
		channels[i] = make(chan int)
	}
	
	// 启动 worker
	for i := 0; i < workers; i++ {
		go func(ch <-chan int, idx int) {
			for v := range ch {
				fmt.Printf("Worker %d got: %d\n", idx, v)
			}
		}(channels[i], i)
	}
	
	// 分发数据
	go func() {
		for _, v := range data {
			// 轮询分发给不同的 worker
			for i := 0; i < workers; i++ {
				channels[i] <- v
				break
			}
		}
		// 关闭所有 channel
		for _, ch := range channels {
			close(ch)
		}
	}()
	
	return channels
}

// Fan-in: 将多个 channel 合并为一个
func fanIn(channels ...<-chan int) <-chan int {
	out := make(chan int)
	
	var wg sync.WaitGroup
	wg.Add(len(channels))
	
	for _, ch := range channels {
		go func(c <-chan int) {
			defer wg.Done()
			for v := range c {
				out <- v
			}
		}(ch)
	}
	
	go func() {
		wg.Wait()
		close(out)
	}()
	
	return out
}

package main

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

// Semaphore 基于 channel 实现
type Semaphore struct {
	sem chan struct{}
}

func NewSemaphore(limit int) *Semaphore {
	return &Semaphore{
		sem: make(chan struct{}, limit),
	}
}

func (s *Semaphore) Acquire() {
	s.sem <- struct{}{}
}

func (s *Semaphore) Release() {
	<-s.sem
}

func main() {
	sem := NewSemaphore(3)  // 最多3个并发
	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 started\n", id)
			time.Sleep(time.Second)
			fmt.Printf("Worker %d finished\n", id)
		}(i)
	}
	
	wg.Wait()
}

package main

import (
	"fmt"
)

// Generate 产生自然数序列 2, 3, 4, ...
func Generate(ch chan<- int) {
	for i := 2; ; i++ {
		ch <- i
	}
}

// Sieve 筛掉能被 prime 整除的数
func Sieve(in <-chan int, prime int, out chan<- int) {
	for {
		if v := <-in; v%prime != 0 {
			out <- v
		}
	}
}

func main() {
	ch := make(chan int)
	go Generate(ch)  // 启动生成器
	
	for i := 0; i < 10; i++ {
		prime := <-ch
		fmt.Printf("Prime: %d\n", prime)
		ch1 := make(chan int)
		go Sieve(ch, prime, ch1)  // 链式连接
		ch = ch1
	}
}

package main

import (
	"fmt"
	"time"
)

// Timeout 带超时的函数执行
func Timeout(d time.Duration, fn func()) bool {
	done := make(chan struct{})
	go func() {
		fn()
		close(done)
	}()
	
	select {
	case <-done:
		return true
	case <-time.After(d):
		return false
	}
}

// SelectWithTimeout 带超时的 channel 读取
func SelectWithTimeout(ch <-chan int, timeout time.Duration) (int, bool) {
	select {
	case v := <-ch:
		return v, true
	case <-time.After(timeout):
		return 0, false
	}
}

func main() {
	// 测试 1：正常完成
	result := Timeout(time.Second, func() {
		time.Sleep(time.Millisecond * 500)
		fmt.Println("Operation completed")
	})
	fmt.Printf("Result: %v\n", result)  // true
	
	// 测试 2：超时
	result = Timeout(time.Millisecond*100, func() {
		time.Sleep(time.Second)
		fmt.Println("This won't print")
	})
	fmt.Printf("Result: %v\n", result)  // false
}

type hchan struct {
	qcount   uint           // 队列中元素总数
	dataqsiz uint           // 环形缓冲区大小
	buf      unsafe.Pointer // 环形缓冲区
	elemsize uint16         // 元素大小
	closed   uint32         // 是否已关闭
	elemtype *_type         // 元素类型
	sendx    uint           // 发送索引
	recvx    uint           // 接收索引
	recvq    waitq          // 等待接收的 G 队列
	sendq    waitq          // 等待发送的 G 队列
	lock     mutex          // 保护上述所有字段
}

// 伪代码：select 的实现逻辑
func sigselect(ncase int, PCSlice, cases *hchan) (sel int, gotOK bool) {
	// 1. 找出所有可执行的 case
	var selectedCases [maxSelCase]*sudog
	var shuffledIndices [maxSelCase]int
	var randomIndex int
	
	for i := 0; i < ncase; i++ {
		c := cases[i]
		if c == nil {
			continue
		}
		// 检查是否可以发送或接收
		if canSend(c.ch) || canRecv(c.ch) {
			selectedCases[randomIndex] = c
			shuffledIndices[randomIndex] = i
			randomIndex++
		}
	}
	
	// 2. 随机选择一个候选 case
	if randomIndex > 0 {
		idx := fastrand() % uint32(randomIndex)
		return shuffledIndices[idx], true
	}
	
	// 3. 没有候选：如果有 default，执行 default
	for i := 0; i < ncase; i++ {
		if cases[i].dir == _SELECTdefault {
			return i, false
		}
	}
	
	// 4. 没有 default：挂起当前 goroutine
	gopark(...)
	return 0, false
}