1
package bulkhead
2

3
import (
4
  "context"
5
  "fmt"
6
)
7

8
// Bulkhead 基于信号量的舱壁隔离
9
type Bulkhead struct {
10
  sem chan struct{} // 带缓冲 channel 充当信号量
11
}
12

13
// New 创建一个最大并发数为 maxConcurrent 的舱壁
14
func New(maxConcurrent int) *Bulkhead {
15
  return &Bulkhead{
16
    sem: make(chan struct{}, maxConcurrent),
17
  }
18
}
19

20
// Execute 在舱壁保护下执行函数
21
// 如果并发数已满，返回错误而非阻塞
22
func (b *Bulkhead) Execute(ctx context.Context, fn func() error) error {
23
  select {
24
  case b.sem <- struct{}{}:
25
    // 成功获取许可
26
    defer func() { <-b.sem }()
27
    return fn()
28
  default:
29
    // 并发数已满，直接拒绝
30
    return fmt.Errorf("bulkhead: 并发数已满，请求被拒绝")
31
  }
32
}
33

34
// ExecuteWithWait 等待可用许可或超时
35
func (b *Bulkhead) ExecuteWithWait(ctx context.Context, fn func() error) error {
36
  select {
37
  case b.sem <- struct{}{}:
38
    defer func() { <-b.sem }()
39
    return fn()
40
  case <-ctx.Done():
41
    return fmt.Errorf("bulkhead: 等待超时，%w", ctx.Err())
42
  }
43
}

1
package main
2

3
import (
4
  "context"
5
  "fmt"
6
  "net/http"
7
  "time"
8

9
  "bulkhead"
10
)
11

12
func main() {
13
  // 给下游服务 A 分配 20 个并发
14
  bhA := bulkhead.New(20)
15
  // 给下游服务 B 分配 10 个并发（B 历史表现差，少分配点）
16
  bhB := bulkhead.New(10)
17

18
  http.HandleFunc("/api", func(w http.ResponseWriter, r *http.Request) {
19
    // 调用服务 A，受舱壁保护
20
    err := bhA.Execute(r.Context(), func() error {
21
      resp, err := http.Get("http://service-a/endpoint")
22
      if err != nil {
23
        return err
24
      }
25
      defer resp.Body.Close()
26
      // 处理响应...
27
      return nil
28
    })
29
    if err != nil {
30
      // 舱壁拒绝，返回降级响应
31
      w.WriteHeader(http.StatusServiceUnavailable)
32
      fmt.Fprint(w, "服务繁忙，请稍后重试")
33
      return
34
    }
35

36
    // 调用服务 B，带超时等待
37
    ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second)
38
    defer cancel()
39

40
    err = bhB.ExecuteWithWait(ctx, func() error {
41
      resp, err := http.Get("http://service-b/endpoint")
42
      if err != nil {
43
        return err
44
      }
45
      defer resp.Body.Close()
46
      // 处理响应...
47
      return nil
48
    })
49
    if err != nil {
50
      // 降级处理
51
      w.WriteHeader(http.StatusServiceUnavailable)
52
      fmt.Fprint(w, "服务 B 繁忙，已降级")
53
      return
54
    }
55

56
    w.WriteHeader(http.StatusOK)
57
  })
58
}

1
class SemaphoreBulkhead {
2
  private active = 0;
3

4
  constructor(private maxConcurrent: number) {}
5

6
  async execute<T>(fn: () => Promise<T>): Promise<T> {
7
    // 并发数已满，直接拒绝
8
    if (this.active >= this.maxConcurrent) {
9
      throw new Error("bulkhead: 并发数已满，请求被拒绝");
10
    }
11

12
    this.active++;
13
    try {
14
      return await fn();
15
    } finally {
16
      this.active--;
17
    }
18
  }
19
}

1
class QueueBulkhead {
2
  private active = 0;
3
  private queue: Array<() => void> = [];
4

5
  constructor(
6
    private maxConcurrent: number,
7
    private maxWait: number = 5000, // 最大等待时间（毫秒）
8
  ) {}
9

10
  async execute<T>(fn: () => Promise<T>): Promise<T> {
11
    // 等待可用许可
12
    await this.acquire();
13

14
    try {
15
      return await fn();
16
    } finally {
17
      this.release();
18
    }
19
  }
20

21
  private acquire(): Promise<void> {
22
    if (this.active < this.maxConcurrent) {
23
      this.active++;
24
      return Promise.resolve();
25
    }
26

27
    // 超过最大等待数，直接拒绝
28
    if (this.queue.length >= this.maxConcurrent) {
29
      return Promise.reject(new Error("bulkhead: 等待队列已满"));
30
    }
31

32
    // 排队等待，带超时
33
    return new Promise<void>((resolve, reject) => {
34
      const timer = setTimeout(() => {
35
        const idx = this.queue.indexOf(waiter);
36
        if (idx !== -1) this.queue.splice(idx, 1);
37
        reject(new Error("bulkhead: 等待超时"));
38
      }, this.maxWait);
39

40
      const waiter = () => {
41
        clearTimeout(timer);
42
        resolve();
43
      };
44

45
      this.queue.push(waiter);
46
    });
47
  }
48

49
  private release(): void {
50
    this.active--;
51

52
    // 唤醒下一个等待的请求
53
    if (this.queue.length > 0) {
54
      this.active++;
55
      const next = this.queue.shift()!;
56
      next();
57
    }
58
  }
59
}

1
// 给不同下游分配不同的舱壁
2
const userBulkhead = new QueueBulkhead(20);
3
const orderBulkhead = new QueueBulkhead(10);
4

5
async function fetchUser(id: string) {
6
  return userBulkhead.execute(async () => {
7
    const resp = await fetch(`/api/users/${id}`);
8
    if (!resp.ok) throw new Error(`用户服务返回 ${resp.status}`);
9
    return resp.json();
10
  });
11
}
12

13
async function fetchOrder(id: string) {
14
  return orderBulkhead.execute(async () => {
15
    const resp = await fetch(`/api/orders/${id}`);
16
    if (!resp.ok) throw new Error(`订单服务返回 ${resp.status}`);
17
    return resp.json();
18
  });
19
}