Go语言信号量实际应用示例

发布时间: 2025-04-24 23:09

↑

# Go语言信号量实际应用示例

信号量是一种强大的并发控制机制，在Go语言中有着广泛的应用。本文将通过实际示例展示如何在Go程序中使用信号量解决各种并发问题。

## 使用标准库中的信号量

Go语言标准库中并没有直接提供信号量的实现，但我们可以通过`sync.Mutex`和`sync.Cond`组合实现简单的信号量，或者使用第三方库如`golang.org/x/sync/semaphore`。

### 使用golang.org/x/sync/semaphore

```go
package main

import (
	"context"
	"fmt"
	"log"
	"runtime"
	"time"

"golang.org/x/sync/semaphore"
)

func main() {
	// 创建一个权重为5的信号量，表示最多允许5个并发操作
	weight := int64(5)
	sem := semaphore.NewWeighted(weight)
	ctx := context.Background()

// 模拟10个并发任务
	for i := 0; i < 10; i++ {
		// 尝试获取信号量
		if err := sem.Acquire(ctx, 1); err != nil {
			log.Printf("Failed to acquire semaphore: %v\n", err)
			break
		}

// 启动goroutine执行任务
		go func(id int) {
			defer sem.Release(1)
			// 模拟任务执行
			fmt.Printf("Worker %d: 开始执行任务\n", id)
			time.Sleep(2 * time.Second)
			fmt.Printf("Worker %d: 任务完成\n", id)
		}(i)
	}

// 等待所有信号量被释放
	if err := sem.Acquire(ctx, weight); err != nil {
		log.Printf("Failed to acquire semaphore: %v\n", err)
	}
	fmt.Println("所有任务已完成")
}
```

这个示例展示了如何使用信号量限制并发任务的数量。无论有多少个任务需要执行，同时运行的任务数量不会超过信号量的权重值。

## 使用信号量实现资源池

信号量非常适合实现资源池，限制对有限资源的并发访问：

```go
package main

import (
	"context"
	"fmt"
	"math/rand"
	"sync"
	"time"

"golang.org/x/sync/semaphore"
)

// 数据库连接池示例
type DBConnection struct {
	id int
}

func (db *DBConnection) Query() int {
	// 模拟查询
	time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
	return rand.Intn(100)
}

type ConnectionPool struct {
	conns     []*DBConnection
	sem       *semaphore.Weighted
	available []bool
	mu        sync.Mutex
}

func NewConnectionPool(size int) *ConnectionPool {
	pool := &ConnectionPool{
		conns:     make([]*DBConnection, size),
		sem:       semaphore.NewWeighted(int64(size)),
		available: make([]bool, size),
	}

// 初始化连接
	for i := 0; i < size; i++ {
		pool.conns[i] = &DBConnection{id: i}
		pool.available[i] = true
	}

return pool
}

func (p *ConnectionPool) Acquire(ctx context.Context) (*DBConnection, error) {
	// 获取信号量
	if err := p.sem.Acquire(ctx, 1); err != nil {
		return nil, err
	}

// 分配连接
	p.mu.Lock()
	var conn *DBConnection
	for i, available := range p.available {
		if available {
			p.available[i] = false
			conn = p.conns[i]
			break
		}
	}
	p.mu.Unlock()

return conn, nil
}

func (p *ConnectionPool) Release(conn *DBConnection) {
	p.mu.Lock()
	// 查找连接并标记为可用
	for i, c := range p.conns {
		if c.id == conn.id {
			p.available[i] = true
			break
		}
	}
	p.mu.Unlock()

// 释放信号量
	p.sem.Release(1)
}

func main() {
	// 创建一个有5个连接的连接池
	pool := NewConnectionPool(5)
	ctx := context.Background()

// 模拟20个并发请求
	var wg sync.WaitGroup
	for i := 0; i < 20; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()

// 获取连接
			conn, err := pool.Acquire(ctx)
			if err != nil {
				fmt.Printf("请求 %d: 无法获取连接: %v\n", id, err)
				return
			}
			fmt.Printf("请求 %d: 获取连接 %d\n", id, conn.id)

// 使用连接
			result := conn.Query()
			fmt.Printf("请求 %d: 查询结果: %d\n", id, result)

// 释放连接
			pool.Release(conn)
			fmt.Printf("请求 %d: 释放连接 %d\n", id, conn.id)
		}(i)
	}

wg.Wait()
	fmt.Println("所有请求已完成")
}
```

## 使用信号量实现限流器

信号量可以用来实现简单的限流器，控制请求速率：

```go
package main

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

"golang.org/x/sync/semaphore"
)

// 限流器实现
type RateLimiter struct {
	sem     *semaphore.Weighted
	interval time.Duration
}

func NewRateLimiter(rate int, interval time.Duration) *RateLimiter {
	return &RateLimiter{
		sem:     semaphore.NewWeighted(int64(rate)),
		interval: interval,
	}
}

func (r *RateLimiter) Allow(ctx context.Context) error {
	// 尝试获取信号量
	err := r.sem.Acquire(ctx, 1)
	if err != nil {
		return err
	}

// 启动一个goroutine在指定时间后释放信号量
	go func() {
		time.Sleep(r.interval)
		r.sem.Release(1)
	}()

return nil
}

func main() {
	// 创建一个限流器，每秒允许5个请求
	limiter := NewRateLimiter(5, time.Second)
	ctx := context.Background()

// 模拟20个并发请求
	var wg sync.WaitGroup
	for i := 0; i < 20; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()

// 尝试通过限流器
			start := time.Now()
			err := limiter.Allow(ctx)
			elapsed := time.Since(start)

if err != nil {
				fmt.Printf("请求 %d: 被限流: %v\n", id, err)
				return
			}

fmt.Printf("请求 %d: 通过限流器，等待时间: %v\n", id, elapsed)

// 模拟处理请求
			time.Sleep(100 * time.Millisecond)
			fmt.Printf("请求 %d: 处理完成\n", id)
		}(i)
	}

wg.Wait()
	fmt.Println("所有请求已完成")
}
```

## 使用信号量实现工作池

信号量可以用来控制工作池中的并发工作者数量：

```go
package main

import (
	"context"
	"fmt"
	"math/rand"
	"sync"
	"time"

"golang.org/x/sync/semaphore"
)

// 任务定义
type Task struct {
	id int
}

// 工作池实现
type WorkerPool struct {
	tasks   chan Task
	sem     *semaphore.Weighted
	wg      sync.WaitGroup
	workers int
}

func NewWorkerPool(workers int, queueSize int) *WorkerPool {
	return &WorkerPool{
		tasks:   make(chan Task, queueSize),
		sem:     semaphore.NewWeighted(int64(workers)),
		workers: workers,
	}
}

func (p *WorkerPool) Start(ctx context.Context) {
	for i := 0; i < p.workers; i++ {
		p.wg.Add(1)
		go func(workerID int) {
			defer p.wg.Done()

for {
				select {
				case task, ok := <-p.tasks:
					if !ok {
						return // 通道已关闭，退出
					}

// 获取信号量
					err := p.sem.Acquire(ctx, 1)
					if err != nil {
						fmt.Printf("工作者 %d: 无法获取信号量: %v\n", workerID, err)
						return
					}

// 处理任务
					fmt.Printf("工作者 %d: 开始处理任务 %d\n", workerID, task.id)
					time.Sleep(time.Duration(rand.Intn(500)+500) * time.Millisecond) // 模拟处理时间
					fmt.Printf("工作者 %d: 完成任务 %d\n", workerID, task.id)

// 释放信号量
					p.sem.Release(1)

case <-ctx.Done():
					fmt.Printf("工作者 %d: 收到取消信号，退出\n", workerID)
					return
				}
			}
		}(i)
	}
}

func (p *WorkerPool) Submit(task Task) {
	p.tasks <- task
}

func (p *WorkerPool) Stop() {
	close(p.tasks)
	p.wg.Wait()
}

func main() {
	// 创建一个有5个工作者的工作池，任务队列大小为20
	pool := NewWorkerPool(5, 20)
	ctx := context.Background()

// 启动工作池
	pool.Start(ctx)

// 提交20个任务
	for i := 0; i < 20; i++ {
		pool.Submit(Task{id: i})
	}

// 等待一段时间让任务处理
	time.Sleep(5 * time.Second)

// 停止工作池
	pool.Stop()
	fmt.Println("工作池已停止")
}
```

## 使用信号量实现并发控制的最佳实践

1. **选择合适的信号量实现**：
   - 对于简单场景，可以使用`golang.org/x/sync/semaphore`
   - 对于复杂场景，可能需要自定义实现或使用其他并发原语组合

2. **合理设置信号量大小**：
   - 太小会限制并发性能
   - 太大会导致资源过度使用
   - 可以根据CPU核心数、内存大小或外部资源限制来设置

3. **处理获取信号量超时**：
   - 使用带超时的context避免无限等待
   - 实现优雅的降级策略

4. **确保正确释放信号量**：
   - 使用defer确保在各种情况下都能释放信号量
   - 避免重复释放或忘记释放

## 总结

信号量是Go语言并发编程中的强大工具，可以用于限制并发数量、实现资源池、控制访问速率等多种场景。通过本文的示例，我们展示了信号量在实际应用中的多种用法，希望能帮助读者更好地理解和使用信号量解决并发问题。

在Go语言中，虽然信号量不是直接暴露的原语，但通过标准库或第三方库提供的实现，我们可以轻松地在程序中使用信号量控制并发，提高程序的性能和稳定性。

元素码农