ShowDocs
搜索文章Ctrl K

外观

← 返回Go列表
排错

Go 后端面试复盘：语言、并发与服务实践 · 29.2 有界并发执行任务至29.8 优雅关闭模板 ​

返回分篇目录

前置：Go 系统学习，并完成一个HTTP 工程练习。本页用于面试前查漏补缺，不按章节替代语言课程。

29.2 有界并发执行任务 ​

go
func RunAll(ctx context.Context, jobs []Job, limit int) error {
	if limit <= 0 {
		return fmt.Errorf("invalid limit: %d", limit)
	}

	sem := make(chan struct{}, limit)
	errCh := make(chan error, 1)
	var wg sync.WaitGroup

	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

launchLoop:
	for _, job := range jobs {
		job := job

		select {
		case sem <- struct{}{}:
		case <-ctx.Done():
			break launchLoop
		}

		wg.Add(1)
		go func() {
			defer wg.Done()
			defer func() { <-sem }()

			if err := process(ctx, job); err != nil {
				select {
				case errCh <- err:
					cancel()
				default:
				}
			}
		}()
	}

	wg.Wait()
	close(errCh)

	if err := <-errCh; err != nil {
		return err
	}
	return ctx.Err()
}

说明：上例固定等待所有已启动任务返回；process必须配合context停止，否则wg.Wait仍可能阻塞。取消和可用令牌同时就绪时仍可能再启动任务，不保证取消发生后绝无新任务开始。生产代码按业务选择任务组合同，不能将“发出取消信号”当作已经完成收拢。

29.3 Worker Pool ​

go
func StartWorkers(
	ctx context.Context,
	workers int,
	jobs <-chan Job,
	results chan<- Result,
) {
	var wg sync.WaitGroup
	wg.Add(workers)

	for i := 0; i < workers; i++ {
		go func() {
			defer wg.Done()
			for {
				select {
				case <-ctx.Done():
					return
				case job, ok := <-jobs:
					if !ok {
						return
					}
					result := handle(job)
					select {
					case results <- result:
					case <-ctx.Done():
						return
					}
				}
			}
		}()
	}

	go func() {
		wg.Wait()
		close(results)
	}()
}

29.4 带 TTL 的简单缓存 ​

go
type entry[V any] struct {
	value     V
	expiresAt time.Time
}

type Cache[K comparable, V any] struct {
	mu    sync.RWMutex
	items map[K]entry[V]
}

func NewCache[K comparable, V any]() *Cache[K, V] {
	return &Cache[K, V]{items: make(map[K]entry[V])}
}

func (c *Cache[K, V]) Get(key K) (V, bool) {
	c.mu.RLock()
	e, ok := c.items[key]
	c.mu.RUnlock()

	if !ok || !time.Now().Before(e.expiresAt) {
		var zero V
		return zero, false
	}
	return e.value, true
}

func (c *Cache[K, V]) Set(key K, value V, ttl time.Duration) {
	c.mu.Lock()
	c.items[key] = entry[V]{
		value:     value,
		expiresAt: time.Now().Add(ttl),
	}
	c.mu.Unlock()
}

func (c *Cache[K, V]) Delete(key K) {
	c.mu.Lock()
	delete(c.items, key)
	c.mu.Unlock()
}

生产缓存还需考虑容量上限、淘汰算法、批量清理、时钟、singleflight、指标和大对象。

29.5 生产者消费者安全关闭 ​

go
func Produce(ctx context.Context, out chan<- Item, items []Item) error {
	defer close(out)
	for _, item := range items {
		select {
		case out <- item:
		case <-ctx.Done():
			return ctx.Err()
		}
	}
	return nil
}

func Consume(ctx context.Context, in <-chan Item) error {
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case item, ok := <-in:
			if !ok {
				return nil
			}
			if err := handle(item); err != nil {
				return err
			}
		}
	}
}

29.6 指数退避加抖动 ​

go
func Retry(
	ctx context.Context,
	maxAttempts int,
	baseDelay time.Duration,
	fn func(context.Context) error,
) error {
	if maxAttempts <= 0 || baseDelay <= 0 || baseDelay > 30*time.Second {
		return errors.New("positive attempts and baseDelay in (0,30s] required")
	}
	maxDelay := baseDelay
	var lastErr error

	for attempt := 0; attempt < maxAttempts; attempt++ {
		if err := ctx.Err(); err != nil { return errors.Join(lastErr, err) }
		if err := fn(ctx); err == nil {
			return nil
		} else {
			lastErr = err
			if !isRetryable(err) {
				return err
			}
		}

		if attempt == maxAttempts-1 {
			break
		}

		jitter, err := rand.Int(rand.Reader, big.NewInt(int64(maxDelay)+1))
		if err != nil {
			return errors.Join(lastErr, err)
		}

		timer := time.NewTimer(time.Duration(jitter.Int64()))
		select {
		case <-ctx.Done():
			timer.Stop()
			return errors.Join(lastErr, ctx.Err())
		case <-timer.C:
		}
		maxDelay = min(30*time.Second, maxDelay*2)
	}

	return lastErr
}

本片段将退避上限固定30秒，正值输入先校验，再饱和倍增，避免左移溢出及0次尝试误报成功。30秒只是教学合同，不是生产推荐；总时间预算由ctx控制，fn仍需配合取消。isRetryable由业务提供，crypto/rand与math/big须显式导入。

29.7 HTTP Client 安全请求模板 ​

go
func FetchJSON[T any](
	ctx context.Context,
	client *http.Client,
	url string,
	maxBody int64,
) (T, error) {
	var zero T
	if client == nil || maxBody < 0 || maxBody == math.MaxInt64 {
		return zero, errors.New("client required and maxBody must allow maxBody+1")
	}

	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
	if err != nil {
		return zero, fmt.Errorf("create request: %w", err)
	}

	resp, err := client.Do(req)
	if err != nil {
		return zero, fmt.Errorf("send request: %w", err)
	}
	defer resp.Body.Close()

	body := io.LimitReader(resp.Body, maxBody+1)
	data, err := io.ReadAll(body)
	if err != nil {
		return zero, fmt.Errorf("read response: %w", err)
	}
	if int64(len(data)) > maxBody {
		return zero, fmt.Errorf("response body too large")
	}
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		return zero, fmt.Errorf("unexpected status %d: %s",
			resp.StatusCode, strings.TrimSpace(string(data)))
	}

	var result T
	if err := json.Unmarshal(data, &result); err != nil {
		return zero, fmt.Errorf("decode response: %w", err)
	}
	return result, nil
}

真实系统还需做 URL/SSRF 校验、错误体脱敏、Content-Type 校验和可观测性。

29.8 优雅关闭模板 ​

go
func main() {
	srv := &http.Server{
		Addr:              ":8080",
		Handler:           newHandler(),
		ReadHeaderTimeout: 5 * time.Second,
		IdleTimeout:       60 * time.Second,
	}

	errCh := make(chan error, 1)
	go func() {
		if err := srv.ListenAndServe(); !errors.Is(err, http.ErrServerClosed) {
			errCh <- err
		}
	}()

	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
	defer signal.Stop(sigCh)

	select {
	case sig := <-sigCh:
		log.Printf("received signal: %v", sig)
	case err := <-errCh:
		log.Fatalf("server error: %v", err)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	if err := srv.Shutdown(ctx); err != nil {
		log.Printf("shutdown error: %v", err)
	}
}

继续阅读