信号量

概念定义

信号量持有一个非负的许可计数。获取会使其递减，若将变为负数则阻塞；释放会使其递增并唤醒等待者。二元信号量等同于一把锁，而计数信号量把并发上限设为 n。与互斥锁不同，信号量没有归属——任何线程都可释放许可——这让它既适合用于生产者与消费者之间的信号，也适合用于限定资源。

为什么重要

无界并发正是服务耗尽连接、内存或下游配额的方式。信号量在使用点把隐式限制变为显式限制，让突发流量在边界处排队而不是级联成故障。在成千上万任务共享少量资源的异步运行时中，它是让并发量与容量相匹配的基本原语。

工程实践

把许可数设为受保护资源的真实容量，绝不能超过它。始终在 finally 式代码块中释放，使错误路径不会泄漏许可，并使用超时或有界等待，让调用方在资源饱和时不会永久挂起。当生产者与消费者需要会合时，一对信号量能干净地表达交接，但除非实现保证，否则不要假设等待者按序被服务。

代码示例

export class Semaphore {
  constructor(max) {
    if (!Number.isInteger(max) || max < 1) throw new RangeError("max must be a positive integer");
    this.max = max;
    this.active = 0;
    this.waiters = [];
  }
  get inFlight() {
    return this.active;
  }
  async acquire() {
    if (this.active < this.max) {
      this.active += 1;
      return this.makeRelease();
    }
    await new Promise((resolve) => this.waiters.push(resolve));
    return this.makeRelease();
  }
  makeRelease() {
    let released = false;
    return () => {
      if (released) return;
      released = true;
      this.active -= 1;
      const next = this.waiters.shift();
      if (next) {
        this.active += 1;
        next();
      }
    };
  }
  async run(task) {
    const release = await this.acquire();
    try {
      return await task();
    } finally {
      release();
    }
  }
  async runAll(tasks) {
    return Promise.all(tasks.map((t) => this.run(t)));
  }
}

package main

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

type WeightedSemaphore struct {
	mu       sync.Mutex
	cond     *sync.Cond
	capacity int64
	used     int64
}

func NewWeightedSemaphore(capacity int64) (*WeightedSemaphore, error) {
	if capacity < 1 {
		return nil, fmt.Errorf("capacity must be positive")
	}
	sem := &WeightedSemaphore{capacity: capacity}
	sem.cond = sync.NewCond(&sem.mu)
	return sem, nil
}

func (s *WeightedSemaphore) Acquire(ctx context.Context, weight int64) error {
	if weight < 1 || weight > s.capacity {
		return fmt.Errorf("weight %d outside 1..%d", weight, s.capacity)
	}

	done := make(chan struct{})
	go func() {
		select {
		case <-ctx.Done():
			s.mu.Lock()
			s.cond.Broadcast()
			s.mu.Unlock()
		case <-done:
		}
	}()
	defer close(done)

	s.mu.Lock()
	defer s.mu.Unlock()
	for s.used+weight > s.capacity {
		if err := ctx.Err(); err != nil {
			return err
		}
		s.cond.Wait()
	}
	if err := ctx.Err(); err != nil {
		return err
	}
	s.used += weight
	return nil
}

func (s *WeightedSemaphore) Release(weight int64) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if weight < 1 || weight > s.used {
		panic(fmt.Sprintf("release %d with %d in use", weight, s.used))
	}
	s.used -= weight
	s.cond.Broadcast()
}

func main() {
	sem, err := NewWeightedSemaphore(3)
	if err != nil {
		fmt.Println(err)
		return
	}

	jobs := []struct {
		name   string
		weight int64
	}{
		{"small", 1},
		{"medium", 2},
		{"large", 3},
		{"small-2", 1},
	}

	var wg sync.WaitGroup
	for _, job := range jobs {
		job := job
		wg.Add(1)
		go func() {
			defer wg.Done()
			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
			defer cancel()
			if err := sem.Acquire(ctx, job.weight); err != nil {
				fmt.Printf("%s: %v\n", job.name, err)
				return
			}
			time.Sleep(20 * time.Millisecond)
			sem.Release(job.weight)
			fmt.Printf("%s finished (weight %d)\n", job.name, job.weight)
		}()
	}
	wg.Wait()
	fmt.Println("all jobs done")
}