Go 控制协程(goroutine)的并发数量

package main

import (
	"fmt"
	"sync"
)

func worker(id int, sem chan struct{}) {
	sem <- struct{}{} // 占用一个信号量
	defer func() {
		<-sem // 方法运行结束释放信号量
	}()

	// 执行工作任务
	fmt.Printf("Worker %d: Working...\n", id)
}

func main() {
	concurrency := 3
	sem := make(chan struct{}, concurrency)
	var wg sync.WaitGroup

	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			worker(id, sem)
		}(i)
	}

	wg.Wait()
	close(sem)
}

package main

import (
	"fmt"
	"sync"
)

func worker(id int, jobs <-chan int, results chan<- int) {
	//jobs等待主要协程往jobs放数据
	for j := range jobs {
		fmt.Printf("协程池 %d: 协程池正在工作 %d\n", id, j)
		results <- j
	}
}

func main() {
	const numJobs = 5    //协程要做的工作数量
	const numWorkers = 3 //协程池数量

	jobs := make(chan int, numJobs)
	results := make(chan int, numJobs)

	var wg sync.WaitGroup

	// 启动协程池
	for i := 1; i <= numWorkers; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			worker(id, jobs, results)
		}(i)
	}

	// 提交任务
	for j := 1; j <= numJobs; j++ {
		jobs <- j
	}

	close(jobs)

	// 等待所有工作完成
	go func() {
		wg.Wait()
		close(results)
	}()

	// 处理结果
	for result := range results {
		fmt.Println("Result:", result)
	}
}

package main

import (
	"context"
	"fmt"
	"golang.org/x/sync/semaphore"
	"sync"
	"time"
)

func main() {
	/*
		1、在并发量一定的情况下，通过改变允许并发请求数可以更快处理请求任务(在CPU够用的前提下)
		2、sem := semaphore.NewWeighted(n)，参数n就是权重量
		3、当一个协程需要获取的单位的权重越多，运行就会慢(比如权重总量n=5个，一个协程分配了2个，跟一个协程分配1个效率是不一样的)
		4、信号量没有足够的可用权重的情况发生在所有已分配的权重单位都已经被占用，即信号量的当前权重计数达到了它的总容量。在这种情况下，任何尝试通过Acquire方法获取更多权重的调用都将无法立即完成，从而导致调用者（通常是goroutine）阻塞，直到其他调用者释放一些权重单位。
	*/
	/*
		1、权权重较大的任务在资源竞争时有更高的优先级，更容易获得执行的机会
		2、如果当前资源足够满足高权重任务的需求，这些任务将立即执行；若资源不足，则按照权重高低顺序排队等待
		3、一旦任务开始执行，其完成的速度主要取决于任务自身的逻辑复杂度、所需资源以及系统的当前负载等因素，与任务在信号量中的权重无关
		3、高权重的任务并不会中断已经在执行的低权重任务，而是等待这些任务自行释放资源。一旦资源释放，等待队列中的高权重任务将优先被唤醒
		4、Acquire 方法会检查当前信号量的可用资源量是否满足请求的权重，如果满足，则立即减少信号量的资源计数并返回，允许任务继续执行。如果不满足，任务将阻塞等待，直到有足够的资源被释放
	*/
	// 记录开始时间
	startTime := time.Now()
	// 假设外部API允许的最大并发请求为5(信号量的总容量是5个权重单位)
	const (
		maxConcurrentRequests = 5
	)
	sem := semaphore.NewWeighted(maxConcurrentRequests)

	var wg sync.WaitGroup

	// 模拟对外部API的10个并发请求
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(requestId int) {
			defer wg.Done()
			// 假设我们想要获取2个单位的权重
			if err := sem.Acquire(context.Background(), 2); err != nil {
				fmt.Printf("请求 %d 无法获取信号量: %v\n", requestId, err)
				return
			}
			defer sem.Release(2) // 请求完成后释放信号量

			// 模拟对API的请求处理
			fmt.Printf("请求 %d 开始...\n", requestId)
			time.Sleep(2 * time.Second) // 模拟网络延迟
			fmt.Printf("请求 %d 完成。\n", requestId)
		}(i)
	}
	wg.Wait()
	// 记录结束时间
	endTime := time.Now()
	// 计算并打印总耗时
	fmt.Printf("程序总耗时: %v\n", endTime.Sub(startTime))
}

package main

import (
	"fmt"
	"golang.org/x/net/context"
	"golang.org/x/sync/semaphore"
	"runtime"
	"sync"
	"time"
)

// VideoTranscodeJob 视频转码任务
type VideoTranscodeJob struct {
	resolution string
	weight     int64
}

func main() {
	cpuCount := runtime.NumCPU()
	fmt.Printf("当前CPU个数%v\n", cpuCount)
	/*
		1、权权重较大的任务在资源竞争时有更高的优先级，更容易获得执行的机会
		2、如果当前资源足够满足高权重任务的需求，这些任务将立即执行；若资源不足，则按照权重高低顺序排队等待
		3、一旦任务开始执行，其完成的速度主要取决于任务自身的逻辑复杂度、所需资源以及系统的当前负载等因素，与任务在信号量中的权重无关
		3、高权重的任务并不会中断已经在执行的低权重任务，而是等待这些任务自行释放资源。一旦资源释放，等待队列中的高权重任务将优先被唤醒
		4、Acquire 方法会检查当前信号量的可用资源量是否满足请求的权重，如果满足，则立即减少信号量的资源计数并返回，允许任务继续执行。如果不满足，任务将阻塞等待，直到有足够的资源被释放
	*/
	// 初始化两个信号量，一个用于标清，一个用于高清，假设总共有8个CPU核心可用
	normalSem := semaphore.NewWeighted(6)  // 标清任务，分配6个单位权重，因为它们消耗资源较少
	highDefSem := semaphore.NewWeighted(2) // 高清任务，分配2个单位权重，因为它们更消耗资源

	var wg sync.WaitGroup

	//假设有20个需要转码的视频
	videoJobs := []VideoTranscodeJob{
		{"HD", 2}, {"HD", 2}, {"SD", 1}, {"SD", 1}, {"HD", 2},
		{"SD", 1}, {"SD", 1}, {"HD", 2}, {"SD", 1}, {"HD", 2},
		{"SD", 4}, {"SD", 4}, {"HD", 2}, {"SD", 1}, {"HD", 2},
		{"SD", 1}, {"SD", 4}, {"HD", 2}, {"SD", 6}, {"HD", 2},
	}

	for _, job := range videoJobs {
		wg.Add(1)
		go func(job VideoTranscodeJob) {
			defer wg.Done()

			var sem *semaphore.Weighted
			switch job.resolution {
			case "SD":
				sem = normalSem //分配权重大，当前为6，任务在获取执行机会上有优势，但并不直接意味着执行速度快
			case "HD":
				sem = highDefSem
			default:
				panic("无效的分辨率")
			}

			if err := sem.Acquire(context.Background(), job.weight); err != nil {
				fmt.Printf("名为 %s 视频无法获取信号量: %v\n", job.resolution, err)
				return
			}
			defer sem.Release(job.weight) //释放权重对应的信号量

			// 模拟转码任务执行
			fmt.Printf("转码 %s 视频 (权重: %d)...\n", job.resolution, job.weight)
			//通过利用VideoTranscodeJob的weight值来模拟转码时间的长短，HD用时长则设置2比SD的1大，*时间就自然长，运行就时间长
			time.Sleep(time.Duration(job.weight*100) * time.Millisecond) // 模拟不同分辨率视频转码所需时间
			fmt.Printf("------------------------%s 视频转码完成。。。\n", job.resolution)
		}(job)
	}
	wg.Wait()
}

package main

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

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

type weightedTask struct {
	id     int
	weight int64
}

func main() {
	const (
		maxTotalWeight = 20 // 最大总权重
	)
	sem := semaphore.NewWeighted(maxTotalWeight)

	var wg sync.WaitGroup
	tasksCh := make(chan weightedTask, 10)

	// 发送任务
	for i := 1; i <= 10; i++ {
		tasksCh <- weightedTask{id: i, weight: int64(i)} // 假设任务ID即为其权重
	}

	close(tasksCh)

	// 启动任务处理器
	for task := range tasksCh {
		wg.Add(1)
		go func(task weightedTask) {
			defer wg.Done()

			if err := sem.Acquire(context.Background(), int64(task.id)); err != nil {
				fmt.Printf("任务 %d 无法获取信号量: %v\n", task.id, err)
				return
			}
			defer sem.Release(int64(task.id)) //释放

			// 模拟任务执行
			fmt.Printf("任务 %d (权重: %d) 正在运行...\n", task.id, task.weight)
			time.Sleep(time.Duration(task.weight*100) * time.Millisecond) // 示例中简单用时间模拟权重影响
			fmt.Printf("任务 %d 完成.\n", task.id)
		}(task)
	}
	wg.Wait()
}