Golang 并发限制器：支持超时和动态优先级的 concurrency-limiter

在 Go 语言中，管理并发是一个常见需求。concurrency-limiter 是一个支持超时和动态优先级的并发限制插件库，可以帮助开发者更好地控制并发执行的任务。

基本概念

concurrency-limiter 提供了以下核心功能：

• 限制最大并发数
• 支持任务优先级
• 支持超时控制
• 动态调整并发限制

安装

go get github.com/your-repo/concurrency-limiter

基本使用示例

package main

import (
	"context"
	"fmt"
	"time"
	"github.com/your-repo/concurrency-limiter"
)

func main() {
	// 创建一个最大并发数为3的限流器
	limiter := concurrency_limiter.NewLimiter(3)

	// 模拟10个任务
	for i := 0; i < 10; i++ {
		taskID := i
		go func() {
			// 获取许可，如果没有可用许可会阻塞
			token, err := limiter.Acquire(context.Background())
			if err != nil {
				fmt.Printf("Task %d failed to acquire token: %v\n", taskID, err)
				return
			}
			defer token.Release() // 确保释放许可

			// 执行任务
			fmt.Printf("Task %d is running\n", taskID)
			time.Sleep(time.Second)
			fmt.Printf("Task %d completed\n", taskID)
		}()
	}

	// 等待所有任务完成
	time.Sleep(5 * time.Second)
}

带超时的使用示例

func main() {
	limiter := concurrency_limiter.NewLimiter(2)
	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
	defer cancel()

	for i := 0; i < 5; i++ {
		taskID := i
		go func() {
			token, err := limiter.Acquire(ctx)
			if err != nil {
				fmt.Printf("Task %d timed out waiting for token\n", taskID)
				return
			}
			defer token.Release()

			fmt.Printf("Task %d started\n", taskID)
			time.Sleep(time.Second)
			fmt.Printf("Task %d finished\n", taskID)
		}()
	}

	time.Sleep(3 * time.Second)
}

带优先级的使用示例

func main() {
	// 创建支持优先级的限流器
	limiter := concurrency_limiter.NewPriorityLimiter(3)

	for i := 0; i < 10; i++ {
		taskID := i
		priority := i % 3 // 优先级0-2
		go func() {
			token, err := limiter.AcquireWithPriority(context.Background(), priority)
			if err != nil {
				fmt.Printf("Task %d failed to acquire token\n", taskID)
				return
			}
			defer token.Release()

			fmt.Printf("Task %d (priority %d) started\n", taskID, priority)
			time.Sleep(time.Second)
			fmt.Printf("Task %d completed\n", taskID)
		}()
	}

	time.Sleep(5 * time.Second)
}

动态调整并发限制

func main() {
	limiter := concurrency_limiter.NewLimiter(2)

	// 初始限制为2
	for i := 0; i < 5; i++ {
		go func(id int) {
			token, _ := limiter.Acquire(context.Background())
			defer token.Release()
			fmt.Printf("Task %d running with limit %d\n", id, limiter.GetLimit())
			time.Sleep(time.Second)
		}(i)
	}

	time.Sleep(1 * time.Second)
	
	// 动态调整限制为4
	limiter.SetLimit(4)
	fmt.Println("Limit increased to 4")

	time.Sleep(3 * time.Second)
}

高级功能：带权重的并发限制

func main() {
	// 创建带权重的限流器，总权重限制为10
	limiter := concurrency_limiter.NewWeightedLimiter(10)

	for i := 0; i < 5; i++ {
		taskID := i
		weight := i + 1 // 权重1-5
		go func() {
			token, err := limiter.AcquireWithWeight(context.Background(), weight)
			if err != nil {
				fmt.Printf("Task %d failed to acquire token\n", taskID)
				return
			}
			defer token.Release()

			fmt.Printf("Task %d (weight %d) started\n", taskID, weight)
			time.Sleep(2 * time.Second)
			fmt.Printf("Task %d completed\n", taskID)
		}()
	}

	time.Sleep(10 * time.Second)
}

性能考虑

1. 对于高并发场景，考虑使用 TryAcquire 非阻塞方法
2. 合理设置超时时间避免长时间阻塞
3. 动态调整限制时要考虑当前正在执行的任务

总结

concurrency-limiter 提供了灵活的并发控制机制，通过超时、优先级和动态调整等功能，可以满足各种复杂的并发控制需求。在实际应用中，可以根据具体场景选择合适的限制策略和参数配置。

注意：上述代码示例中的 concurrency_limiter 是一个假设的库名，实际使用时需要替换为真实的库名或实现。