Golang中处理未知数量数据行的信号量实现

我不理解你问题中的这一部分：

anthonys: 以防某些通道运行时间较长

通道只是一种将项目安全传递给多个协程的方式，那么你说它“运行时间较长”是什么意思呢？

也许你的意思是，如果你尝试从通道中获取项目一段时间，但一直没有获取到任何东西？ 如果是这种情况，那么你需要一个带有两个 case 的 select 语句：一个用于拉取新项目，另一个用于超时处理。因此，工作函数将如下所示：

func worker(input chan int) {
	for {
    select {
		case item, more := <-input :
      if more {
        item++
        fmt.Println(item)
      } else {
        break
      }
    case <-time.After(time.Second * 10):
      break
    }
	}
}

这个 select 语句将等待从通道中获取一个项目或者超时结束。如果超时结束，它将跳出循环。

你的描述有点模糊，以下是我的理解：

1. 你有一组数据需要并行处理
2. 你想并行处理它们，但最多同时处理10项？还是总共就处理10项？
3. 我假设是第一种情况（最多同时并行处理10项），但当我阅读代码时，它对于你的描述来说似乎过于复杂了。

下面的代码片段演示了如何获取一组数据项并进行并行处理。请告诉我我对你问题的理解哪里不对，我们可以一起改进它。

package main

import (
	"fmt"
	"sync"
)

func main() {
	items := []int{1, 2, 3, 4, 5, 6, 7}
	maxWorkers := 10
	inputChan := make(chan int, maxWorkers)
	wg := sync.WaitGroup{}

	for i := 0; i < maxWorkers; i++ {
		wg.Add(1)
		go func() {
			worker(inputChan)
			wg.Done()
		}()
	}

	go func() {
		for item := range items {
			inputChan <- item
		}
		close(inputChan)
	}()

	wg.Wait()
}

func worker(input chan int) {
	for {
		item, more := <-input
		if more {
			item++
			fmt.Println(item)
		} else {
			break
		}
	}
}

这是一个典型的并发控制场景，使用缓冲通道作为信号量是正确的思路。你的代码可以简化，以下是更清晰的实现：

package main

import (
	"fmt"
	"sync"
	"time"
)

const (
	maxWorkers = 10
	totalTasks = 100 // 模拟未知数量的数据行
)

func processRecord(id int) {
	time.Sleep(100 * time.Millisecond) // 模拟处理时间
	fmt.Printf("Processed record %d\n", id)
}

func main() {
	var wg sync.WaitGroup
	semaphore := make(chan struct{}, maxWorkers) // 信号量通道

	// 模拟读取未知数量的数据行
	for i := 0; i < totalTasks; i++ {
		wg.Add(1)
		semaphore <- struct{}{} // 获取信号量

		go func(id int) {
			defer wg.Done()
			defer func() { <-semaphore }() // 释放信号量

			processRecord(id)
		}(i)
	}

	wg.Wait()
	fmt.Println("All records processed")
}

对于你的具体场景（逐行读取CSV），可以这样实现：

package main

import (
	"encoding/csv"
	"fmt"
	"io"
	"os"
	"sync"
)

const maxConcurrent = 10

func worker(id int, record []string, wg *sync.WaitGroup, sem chan struct{}) {
	defer wg.Done()
	defer func() { <-sem }()

	fmt.Printf("Worker %d processing: %v\n", id, record)
	// 处理记录的逻辑
}

func main() {
	file, _ := os.Open("data.csv")
	defer file.Close()

	reader := csv.NewReader(file)
	var wg sync.WaitGroup
	sem := make(chan struct{}, maxConcurrent)

	recordNum := 0
	for {
		record, err := reader.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			continue // 或处理错误
		}

		wg.Add(1)
		sem <- struct{}{} // 等待有空闲槽位
		recordNum++

		go worker(recordNum, record, &wg, sem)
	}

	wg.Wait()
	fmt.Println("Processing complete")
}

对于你的阻塞读取场景，可以使用带超时的select：

func processStream(sign chan struct{}, c *websocket.Conn) {
	sem := make(chan struct{}, 10)
	var wg sync.WaitGroup

	for {
		select {
		case <-sign:
			wg.Wait()
			return
		default:
			sem <- struct{}{}
			wg.Add(1)

			go func() {
				defer wg.Done()
				defer func() { <-sem }()

				msg, err := c.ReadMessage()
				if err != nil {
					return
				}
				processMessage(msg)
			}()
		}
	}
}

关键点：

1. 缓冲通道作为信号量（sem := make(chan struct{}, N)）
2. sem <- struct{}{} 会阻塞直到有空闲槽位
3. <-sem 在goroutine完成时释放槽位
4. sync.WaitGroup 等待所有goroutine完成

这种模式避免了忙等待（time.Sleep循环），更高效且不会挂起。