Golang基于CAS实现高性能可调整信号量库semaphore

信号量概述

信号量(Semaphore)是一种常用的并发控制机制，用于限制对共享资源的并发访问数量。在Golang中，我们可以基于CAS(Compare-And-Swap)原子操作实现高性能的信号量。

基于CAS的信号量实现

下面是一个基于CAS实现的高性能可调整信号量的完整代码示例：

package semaphore

import (
	"sync/atomic"
)

// Semaphore 基于CAS实现的可调整信号量
type Semaphore struct {
	permits int32     // 总许可数
	channel chan struct{} // 用于阻塞和唤醒goroutine的通道
}

// NewSemaphore 创建一个新的信号量
func NewSemaphore(permits int) *Semaphore {
	if permits <= 0 {
		permits = 1 // 默认最少1个许可
	}
	return &Semaphore{
		permits: int32(permits),
		channel: make(chan struct{}, permits),
	}
}

// Acquire 获取一个许可，如果没有可用许可则阻塞
func (s *Semaphore) Acquire() {
	s.channel <- struct{}{}
	atomic.AddInt32(&s.permits, -1)
}

// TryAcquire 尝试获取一个许可，立即返回是否成功
func (s *Semaphore) TryAcquire() bool {
	select {
	case s.channel <- struct{}{}:
		atomic.AddInt32(&s.permits, -1)
		return true
	default:
		return false
	}
}

// Release 释放一个许可
func (s *Semaphore) Release() {
	<-s.channel
	atomic.AddInt32(&s.permits, 1)
}

// AvailablePermits 返回当前可用许可数
func (s *Semaphore) AvailablePermits() int {
	return int(atomic.LoadInt32(&s.permits))
}

// DrainPermits 获取并返回所有当前可用许可
func (s *Semaphore) DrainPermits() int {
	available := s.AvailablePermits()
	for i := 0; i < available; i++ {
		s.Acquire()
	}
	return available
}

// ReducePermits 减少信号量的总许可数
func (s *Semaphore) ReducePermits(reduction int) {
	if reduction <= 0 {
		return
	}
	atomic.AddInt32(&s.permits, -int32(reduction))
	for i := 0; i < reduction; i++ {
		s.Acquire()
	}
}

// IncreasePermits 增加信号量的总许可数
func (s *Semaphore) IncreasePermits(increase int) {
	if increase <= 0 {
		return
	}
	atomic.AddInt32(&s.permits, int32(increase))
	for i := 0; i < increase; i++ {
		s.Release()
	}
}

使用示例

package main

import (
	"fmt"
	"sync"
	"time"
	
	"./semaphore" // 替换为你的semaphore包路径
)

func worker(id int, sem *semaphore.Semaphore, wg *sync.WaitGroup) {
	defer wg.Done()
	
	sem.Acquire()
	fmt.Printf("Worker %d acquired semaphore\n", id)
	time.Sleep(time.Second) // 模拟工作
	sem.Release()
	fmt.Printf("Worker %d released semaphore\n", id)
}

func main() {
	const (
		totalWorkers = 10
		initialPermits = 3
	)
	
	sem := semaphore.NewSemaphore(initialPermits)
	var wg sync.WaitGroup
	
	// 启动10个worker，但只有3个能同时运行
	for i := 1; i <= totalWorkers; i++ {
		wg.Add(1)
		go worker(i, sem, &wg)
	}
	
	// 等待2秒后增加许可数
	time.Sleep(2 * time.Second)
	fmt.Println("\nIncreasing permits to 5...")
	sem.IncreasePermits(2)
	
	// 等待2秒后减少许可数
	time.Sleep(2 * time.Second)
	fmt.Println("\nReducing permits to 2...")
	sem.ReducePermits(3)
	
	wg.Wait()
	fmt.Println("All workers completed")
}

性能优化说明

1. CAS原子操作：使用atomic包提供的原子操作来保证permits字段的线程安全，避免了锁的开销。

2. 通道缓冲：使用带缓冲的channel来实现阻塞和唤醒机制，比基于条件变量的实现更高效。

3. 动态调整：支持运行时动态调整信号量的许可数量，适用于需要弹性控制并发度的场景。

4. 非阻塞尝试：提供TryAcquire方法实现非阻塞获取许可，适用于需要快速失败(fast-fail)的场景。

适用场景

1. 限制数据库连接池并发数
2. 控制API请求并发度
3. 资源池管理
4. 限流保护系统

这个实现结合了CAS的高性能和channel的简洁性，同时提供了动态调整信号量大小的能力，适合大多数需要并发控制的场景。