Go Goroutine 管理与调度详解（GMP 模型）

package main

import (
	"fmt"
	"sync"
)

// Task 任务接口
type Task func() error

// WorkerPool 工作池
type WorkerPool struct {
	tasks    chan Task
	wg       sync.WaitGroup
	numWorkers int
}

// NewWorkerPool 创建工作池
func NewWorkerPool(numWorkers int, bufferSize int) *WorkerPool {
	return &WorkerPool{
		tasks:      make(chan Task, bufferSize),
		numWorkers: numWorkers,
	}
}

// Start 启动工作池
func (wp *WorkerPool) Start() {
	for i := 0; i < wp.numWorkers; i++ {
		wp.wg.Add(1)
		go func(id int) {
			defer wp.wg.Done()
			for task := range wp.tasks {
				if err := task(); err != nil {
					fmt.Printf("Worker %d: task error: %v\n", id, err)
				}
			}
		}(i)
	}
}

// Submit 提交任务
func (wp *WorkerPool) Submit(task Task) error {
	select {
	case wp.tasks <- task:
		return nil
	default:
		return fmt.Errorf("task queue full")
	}
}

// Stop 停止工作池
func (wp *WorkerPool) Stop() {
	close(wp.tasks)
	wp.wg.Wait()
}

type GracefulWorkerPool struct {
	tasks      chan Task
	wg         sync.WaitGroup
	stopOnce   sync.Once
	stopChan   chan struct{}
}

func NewGracefulWorkerPool(numWorkers, bufferSize int) *GracefulWorkerPool {
	return &GracefulWorkerPool{
		tasks:    make(chan Task, bufferSize),
		stopChan: make(chan struct{}),
	}
}

func (wp *GracefulWorkerPool) Start() {
	for i := 0; i < wp.numWorkers; i++ {
		wp.wg.Add(1)
		go func(id int) {
			defer wp.wg.Done()
			for {
				select {
				case task, ok := <-wp.tasks:
					if !ok {
						return
					}
					if err := task(); err != nil {
						fmt.Printf("Worker %d error: %v\n", id, err)
					}
				case <-wp.stopChan:
					// 处理完剩余任务后退出
					for task := range wp.tasks {
						task()
					}
					return
				}
			}
		}(i)
	}
}

// Stop 优雅停止：等待所有任务完成
func (wp *GracefulWorkerPool) Stop(timeout time.Duration) error {
	close(wp.tasks)
	
	done := make(chan struct{})
	go func() {
		wp.wg.Wait()
		close(done)
	}()
	
	select {
	case <-done:
		return nil
	case <-time.After(timeout):
		return fmt.Errorf("timeout waiting for workers to finish")
	}
}

import "golang.org/x/sync/errgroup"

func fetchURLs(urls []string) ([]string, error) {
	g, ctx := errgroup.WithContext(context.Background())
	
	results := make([]string, len(urls))
	
	for i, url := range urls {
		i, url := i, url // 捕获循环变量
		g.Go(func() error {
			resp, err := http.GetContext(ctx, url)
			if err != nil {
				return err
			}
			defer resp.Body.Close()
			
			body, err := io.ReadAll(resp.Body)
			if err != nil {
				return err
			}
			
			results[i] = string(body)
			return nil
		})
	}
	
	if err := g.Wait(); err != nil {
		return nil, err
	}
	
	return results, nil
}

func fetchWithCancel(urls []string) error {
	g, ctx := errgroup.WithContext(context.Background())
	
	for _, url := range urls {
		url := url
		g.Go(func() error {
			// 任何 goroutine 出错都会取消所有
			resp, err := http.GetContext(ctx, url)
			if err != nil {
				return err
			}


### 应用场景

在实际项目中，这个知识点经常应用于：

1. **典型场景一**：说明具体应用场景
2. **典型场景二**：说明另一个应用场景
3. **面试要点**：提醒面试中可能被问到的点

			defer resp.Body.Close()
			_, err = io.ReadAll(resp.Body)
			return err
		})
	}
	
	return g.Wait()
}

type cancelCtx struct {
	Context
	done     chan struct{}
	children map[canceler]struct{}
	err      error
	mu       sync.Mutex
}

func (c *cancelCtx) cancel(removeFromChildren bool, err error) {
	if err == nil {
		panic("context: internal error: nil error")
	}
	c.mu.Lock()
	c.err = err
	close(c.done)
	for child := range c.children {
		child.cancel(false, err)
	}
	if removeFromChildren {
		delete(c.children, c)
	}
	c.mu.Unlock()
}

func WithCancel(parent Context) (ctx Context, cancel CancelFunc) {
	c := newCancelCtx(parent)
	propagateCancel(parent, &c)
	return &c, func() { c.cancel(true, nil) }
}

func propagateCancel(parent Context, child canceler) {
	if parent.Done() == nil {
		return
	}
	if p, ok := parentCancelCtx(parent); ok {
		p.mu.Lock()
		if p.err == nil {
			if p.children == nil {
				p.children = make(map[canceler]struct{})
			}
			p.children[child] = struct{}{}
		}
		p.mu.Unlock()
		return
	}
	if child.Cancel() == nil {
		return
	}
	propagateCancel(parent, child)
}

type WaitGroup struct {
	noCopy noCopy
	state1 [3]uint32
}

func (wg *WaitGroup) Add(delta int) {
	statep, semap := wg.state()
	// ... 原子操作实现
}

func (wg *WaitGroup) Done() {
	wg.Add(-1)
}

func (wg *WaitGroup) Wait() {
	statep, semap := wg.state()
	// ... 等待逻辑
	<-semap
}

type CountDownLatch struct {
	mu    sync.Mutex
	cond  *sync.Cond
	count int
}

func NewCountDownLatch(count int) *CountDownLatch {
	cd := &CountDownLatch{
		count: count,
	}
	cd.cond = sync.NewCond(&cd.mu)
	return cd
}

func (cd *CountDownLatch) Await() {
	cd.mu.Lock()
	defer cd.mu.Unlock()
	for cd.count > 0 {
		cd.cond.Wait()
	}
}

func (cd *CountDownLatch) CountDown() {
	cd.mu.Lock()
	defer cd.mu.Unlock()
	cd.count--
	if cd.count == 0 {
		cd.cond.Broadcast()
	}
}

type Barrier struct {
	mu      sync.Mutex
	cond    *sync.Cond
	parties int
	count   int
	phase   int
}

func NewBarrier(parties int) *Barrier {
	b := &Barrier{
		parties: parties,
	}
	b.cond = sync.NewCond(&b.mu)
	return b
}

func (b *Barrier) Await() int {
	b.mu.Lock()
	defer b.mu.Unlock()
	
	b.count++
	if b.count == b.parties {
		b.phase++
		b.count = 0
		b.cond.Broadcast()
		return b.phase
	}
	
	for b.count != b.parties {
		b.cond.Wait()
	}
	
	b.phase++
	return b.phase
}

func gracefulShutdown(srv *http.Server, sigs ...os.Signal) error {
	ctx, cancel := context.With

### 应用场景

在实际项目中，这个知识点经常应用于：

1. **典型场景一**：说明具体应用场景
2. **典型场景二**：说明另一个应用场景
3. **面试要点**：提醒面试中可能被问到的点

Timeout(context.Background(), 30*time.Second)
	defer cancel()
	
	// 监听信号
	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, sigs...)
	
	go func() {
		<-sigChan
		cancel()
	}()
	
	if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
		return err
	}
	
	return srv.Shutdown(ctx)
}

func gracefulShutdownWithTimeout(srv *http.Server, timeout time

### 应用场景

在实际项目中，这个知识点经常应用于：

1. **典型场景一**：说明具体应用场景
2. **典型场景二**：说明另一个应用场景
3. **面试要点**：提醒面试中可能被问到的点

.Duration) error {
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()
	
	go func() {
		<-sigChan
		cancel()
	}()
	
	if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
		return err
	}
	
	return srv.Shutdown(ctx)
}

import "go.uber.org/goleak"

func TestMain(m *testing.M) {
	goleak.VerifyTestMain(m)
}

func detectLeakedGoroutines() {
	p := runtime.Stack(true)
	fmt.Printf("Current goroutines:\n%s\n", p)
	
	// 定期检查
	ticker := time.NewTicker(1 * time.Second)
	defer ticker.Stop()
	
	for range ticker.C {
		stack := runtime.Stack(true)
		if strings.Contains(stack, "makechan") || 
		   strings.Contains(stack, "chan receive") {
			fmt.Println("Possible leak detected!")
		}
	}
}

// Go 1.14+ 自动检测 cgroup CPU limit
func main() {
	fmt.Printf("GOMAXPROCS: %d\n", runtime.GOMAXPROCS(0))
	fmt.Printf("NumCPU: %d\n", runtime.Nu

### 应用场景

在实际项目中，这个知识点经常应用于：

1. **典型场景一**：说明具体应用场景
2. **典型场景二**：说明另一个应用场景
3. **面试要点**：提醒面试中可能被问到的点

mCPU())
}

import _ "go.uber.org/automaxprocs/maxprocs"

func init() {
	maxprocs.Set() // 自动设置 GOMAXPROCS
}