线程池大小调优

概念定义

池把到达流转换为并行执行。旋钮是 worker 数——在上下文切换开销与内存和并发能力之间交易。正确取值取决于 worker 在计算还是在等待。

工作机制

CPU 密集：线程数 = 核数 × 目标利用率。I/O 密集：按 核数 × (1 + 等待时间/服务时间) 估算，再按下游容忍度封顶。队列策略决定过载形态：有界队列加拒绝把突发变成可见错误，无界队列把问题藏成延迟。

使用场景

先画像等待-计算比再谈数字。为不同工作形态分池——CPU 一个、远程调用一个——等待才不会饿死计算。按目标并发压测，盯 p99 与队列深度。

常见误区

“线程越多吞吐越高”在超过核数后反转：吞吐下降、延迟上升。为均值负载调的固定池在峰值翻车——考虑自适应或分级队列。受限运行时上每线程栈内存会封死池上限。孤立地调一个池会 destabilize 整条链：worker 饱和在连接池上等待，表象是线程不够。

代码示例

package main

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

type Job struct {
	ID    int
	Value int
}

type Result struct {
	JobID int
	Out   int
	Err   error
}

func worker(ctx context.Context, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
	defer wg.Done()
	for {
		select {
		case <-ctx.Done():
			return
		case job, ok := <-jobs:
			if !ok {
				return
			}
			if job.Value < 0 {
				results <- Result{JobID: job.ID, Err: fmt.Errorf("job %d: negative value %d", job.ID, job.Value)}
				continue
			}
			select {
			case <-ctx.Done():
				return
			case results <- Result{JobID: job.ID, Out: job.Value * job.Value}:
			}
		}
	}
}

func run(ctx context.Context, jobs []Job, workers int) ([]Result, error) {
	if workers < 1 {
		return nil, fmt.Errorf("workers must be at least 1, got %d", workers)
	}
	if len(jobs) == 0 {
		return nil, nil
	}

	jobCh := make(chan Job)
	resCh := make(chan Result, len(jobs))

	var wg sync.WaitGroup
	for i := 0; i < workers; i++ {
		wg.Add(1)
		go worker(ctx, jobCh, resCh, &wg)
	}

	go func() {
		defer close(jobCh)
		for _, job := range jobs {
			select {
			case <-ctx.Done():
				return
			case jobCh <- job:
			}
		}
	}()

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

	results := make([]Result, 0, len(jobs))
	for result := range resCh {
		results = append(results, result)
	}
	if err := ctx.Err(); err != nil {
		return results, err
	}
	return results, nil
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()

	jobs := []Job{
		{ID: 1, Value: 4},
		{ID: 2, Value: 9},
		{ID: 3, Value: -1},
		{ID: 4, Value: 16},
		{ID: 5, Value: 25},
		{ID: 6, Value: 36},
	}

	results, err := run(ctx, jobs, 3)
	if err != nil {
		fmt.Println("stopped early:", err)
	}

	failed := 0
	for _, result := range results {
		if result.Err != nil {
			failed++
			fmt.Println("error:", result.Err)
			continue
		}
		fmt.Printf("job %d -> %d\n", result.JobID, result.Out)
	}
	fmt.Printf("%d results, %d failed\n", len(results), failed)
}

import java.util.concurrent.*;
import java.util.List;

public class ThreadPoolExecutor {
    public static void main(String[] args) throws InterruptedException {
        try (ExecutorService executor = Executors.newFixedThreadPool(4)) {
            List<Callable<String>> tasks = List.of(
                () -> { Thread.sleep(500); return "Task 1 done"; },
                () -> { Thread.sleep(300); return "Task 2 done"; },
                () -> { Thread.sleep(700); return "Task 3 done"; },
                () -> { Thread.sleep(200); return "Task 4 done"; }
            );

            List<Future<String>> futures = executor.invokeAll(tasks);

            for (Future<String> future : futures) {
                try {
                    System.out.println(future.get());
                } catch (ExecutionException e) {
                    System.err.println("Task failed: " + e.getCause());
                }
            }
        }
        System.out.println("All tasks completed");
    }
}