这是一个创建于 2454 天前的主题，其中的信息可能已经有所发展或是发生改变。
下面是主程代码，这是详细代码
func main() { //解析参数 filePath := flag.String("f", "", "文件路径") tplId := flag.String("t", "", "模版 ID") flag.Parse() //解析密钥 pk, err := ParsePrivateKey() check(err) //读取文件 start := time.Now() csvFile, err := os.Open(*filePath) check(err) defer csvFile.Close() csvReader := csv.NewReader(csvFile) arr, err := csvReader.ReadAll() fmt.Println(len(arr)) check(err) paramsChan := make(chan string, 200) //统计成功与失败数量 var mutex = &sync.Mutex{} successNum := 0 failNum := 0 var wg sync.WaitGroup go func() { for _, row := range arr { wg.Add(1) go func(row []string) { //通过添加显式参数，确保当 go 语句执行时，使用当前 row 值（参考 5.6.1 内部匿名函数中获取循环变量的问题） defer wg.Done() params, err := getQuery(row, *tplId, pk) if err != nil { fmt.Println(err) } paramsChan <- params }(row) } wg.Wait() close(paramsChan) //安全关闭通道 }() var wg2 sync.WaitGroup limit := make(chan bool, 100) for s := range paramsChan { wg2.Add(1) limit <- true go func(s string) { defer wg2.Done() res, err := sendMsg(s) if err != nil { fmt.Println(err) mutex.Lock() failNum++ mutex.Unlock() } if res { mutex.Lock() successNum++ mutex.Unlock() } else { mutex.Lock() failNum++ mutex.Unlock() } <-limit }(s) } wg2.Wait() fmt.Printf("发券成功:%d\n", successNum) fmt.Printf("发券失败:%d\n", failNum) fmt.Printf("%.2fs elapsed\n", time.Since(start).Seconds()) } 现在如果只整理请求参数，读取 10W 行的 csv 文件，大概耗时 110-120S 左右，耗费内存在 900M 左右。如果加上发送请求的代码，会因为内存消耗太大，直接被操作系统 KILL。
我用 PHP 开 4 个进程+guzzle 异步请求，处理完 10W 数据耗时在 110S 左右。
性能差这么多，这究竟是我代码写的太菜还是因为 PHP 是最好语言？（手动狗头）
第 1 条附言 2019 年 6 月 18 日
多谢各位老哥的指定，根据各位的建议改了代码，现在 10W 数据整理起来在 60S 左右，内存消耗在 10M。下面是更新代码：
```
func main() {
//解析参数
filePath := flag.String("f", "", "文件路径")
tplId := flag.String("t", "", "模版 ID")
flag.Parse()
//解析密钥
pk, err := ParsePrivateKey()
check(err)
//读取文件
start := time.Now()
paramsChan := make(chan string, runtime.NumCPU())
go readFile(*filePath, *tplId, pk, paramsChan)
//发送数据
var failNum int64
var successNum int64
var wg sync.WaitGroup
for s := range paramsChan {
wg.Add(1)
go func(s string) {
defer wg.Done()
res, err := sendMsg(s)
if res {
atomic.AddInt64(&successNum, 1)
} else {
if err != nil {
fmt.Println(err)
}
atomic.AddInt64(&failNum, 1)
}
}(s)
}
wg.Wait()
fmt.Printf("发券成功:%d\n", successNum)
fmt.Printf("发券失败:%d\n", failNum)
fmt.Printf("%.2fs elapsed\n", time.Since(start).Seconds())
}
func readFile(filePath string, tplId string, pk *rsa.PrivateKey, paramsChan chan string) {
csvFile, err := os.Open(filePath)
check(err)
defer csvFile.Close()
csvReader := csv.NewReader(csvFile)
limit := make(chan struct{}, runtime.NumCPU())
for {
row, err := csvReader.Read()
if err == io.EOF {
break
} else if err != nil {
fmt.Printf("读取 csv 错误: %s\n", err)
}
limit <- struct{}{}
go func() {
defer func() {
<-limit
}()
params, err := getQuery(row, tplId, pk)
if err != nil {
fmt.Println(err)
}
paramsChan <- params
}()
}
for i := 0; i < cap(limit); i++ {
limit <- struct{}{}
}
close(paramsChan)
}
```
```
func main() {
//解析参数
filePath := flag.String("f", "", "文件路径")
tplId := flag.String("t", "", "模版 ID")
flag.Parse()
//解析密钥
pk, err := ParsePrivateKey()
check(err)
//读取文件
start := time.Now()
paramsChan := make(chan string, runtime.NumCPU())
go readFile(*filePath, *tplId, pk, paramsChan)
//发送数据
var failNum int64
var successNum int64
var wg sync.WaitGroup
for s := range paramsChan {
wg.Add(1)
go func(s string) {
defer wg.Done()
res, err := sendMsg(s)
if res {
atomic.AddInt64(&successNum, 1)
} else {
if err != nil {
fmt.Println(err)
}
atomic.AddInt64(&failNum, 1)
}
}(s)
}
wg.Wait()
fmt.Printf("发券成功:%d\n", successNum)
fmt.Printf("发券失败:%d\n", failNum)
fmt.Printf("%.2fs elapsed\n", time.Since(start).Seconds())
}
func readFile(filePath string, tplId string, pk *rsa.PrivateKey, paramsChan chan string) {
csvFile, err := os.Open(filePath)
check(err)
defer csvFile.Close()
csvReader := csv.NewReader(csvFile)
limit := make(chan struct{}, runtime.NumCPU())
for {
row, err := csvReader.Read()
if err == io.EOF {
break
} else if err != nil {
fmt.Printf("读取 csv 错误: %s\n", err)
}
limit <- struct{}{}
go func() {
defer func() {
<-limit
}()
params, err := getQuery(row, tplId, pk)
if err != nil {
fmt.Println(err)
}
paramsChan <- params
}()
}
for i := 0; i < cap(limit); i++ {
limit <- struct{}{}
}
close(paramsChan)
}
```
 | 1 rrfeng 2019 年 6 月 17 日  1 无脑太菜。等下再看。 |
 | 2 littlewing 2019 年 6 月 17 日 php 是最好的语言 |
 | 4 DefoliationM 2019 年 6 月 17 日 via Android 1 你这前面加个 go 然后后面又 wait，你还不如直接把 go 和 wait 都去了 |
| 5 DefoliationM 2019 年 6 月 17 日 via Android 你一个函数里最后写一个 wait 就行了 一个里面定义两次，太菜了，不多说 |
 | 6 richzhu 2019 年 6 月 17 日 老哥，你的性能应该是卡在 ReadAll 处，不要用 ReadAll，改成按行读取试试呢，还有，你这里的等待组，和 goroutine 组合的用法有点够浪啊 |
 | 7 46fo 2019 年 6 月 17 日 1 试下 runtime.GOMAXPROCS(runtime.NumCPU() * 8) |
| 8 rrfeng 2019 年 6 月 17 日 1 @echo404 好好想一想哪里该用协程并发，哪里不该用。 我的话会这样写： 定义一个 channel 传消息 定义一个 channel 计数 go func(){ 计数器，不用锁了因为从 chan 读消息 } go func(){ for line := read_lind(file) { chan <- line } chan <- "end" } for msg := <- chan { go func() { send() } // 这里做并发控制，免得一次全部消息都打出去 } |
| 9 46fo 2019 年 6 月 17 日 好像还可以 atomic.AddUint64(failNum); atomic.AddUint64(successNum); |
 | 10 EthanDon 2019 年 6 月 17 日 你这个是串行啊。。。 |
 | 11 harryge 2019 年 6 月 17 日 因缺思厅，像 @richzhu 说的，你有输出的日志吗？是不是时间都耗在 readAll 上了？ 有点好奇 php 是怎么读取大文件，这块的性能受限于 IO 吧，和语言没啥关系。除非你 PHP 不是一次 readAll 的 |
 | 12 mengzhuo 2019 年 6 月 18 日 via iPhone 太菜了～ Chan 20 行左右就能实现并发控制，不需要你这些奇怪锁 我写过一 Go 小程序，每天处理 2T 左右的加密后的 SQL 数据，做些统计；性能瓶颈都是 io，跑满网卡，磁盘都是小事。 |
 | 13 heimeil 2019 年 6 月 18 日 via Android 4 你这有多少行就启动了多少 goroutine，一个 goroutine 的上下文占用差不多 8K+空间，10W 行大概就 800M 了，实际占用 900M 的话，基本都是创建 goroutine 的操作在消耗资源了。 你发券的话，外部请求明显比不上 range arr，只用一个 goroutine 读，再用一个 chan 发送给几个 goroutine 消费就行了，没必要开海量的 goroutine，开多了反而就出问题了。 |
 | 14 skiy 2019 年 6 月 18 日 哈哈。我用了这么久的 GO，都不敢贴代码。 |
 | 15 viger 2019 年 6 月 18 日 不想吐槽你的代码逻辑，只想吐槽一下你这代码风格。 因为超过 80 行的函数真心不想看。 佩服楼上几位居然能坚持看完的。 建议看完《代码大全》再来贴代码。 |
 | 16 elementpps1 2019 年 6 月 18 日 学习了 |
17 zarte 2019 年 6 月 18 日 你要吧 php 的拿出来对比吧 |
 | 19 echo404 OP @DefoliationM 多谢指点，昨天晚上太忙了，没来得急回复 |
| 24 echo404 OP @heimeil 多谢指点，因为我用 PHP 处理，整理请求在 75S 左右，请求耗时在 40S 左右，所以我一开始就觉得处理数据的需要用并发。 |
 | 26 tt67wq 2019 年 6 月 18 日 时间都花在文件 io 上了吧 |
 | 27 reus 2019 年 6 月 18 日 4 ```go package main import ( "bytes" "fmt" "runtime" "sync/atomic" ) func main() { sem := make(chan struct{}, runtime.NumCPU()) array := bytes.Repeat([]byte("a"), 1000_0000) var c int64 for _, b := range array { b := b sem <- struct{}{} go func() { defer func() { <-sem }() _ = b if n := atomic.AddInt64(&c, 1); n%10000 == 0 { fmt.Printf("%d\n", n) } }() } for i := 0; i < cap(sem); i++ { sem <- struct{}{} } } ``` 给你看一个并发模式，1 千万个任务，最多有 runtime.NumCPU() 个同时跑，而不是像你那样，不停开 1 千万个 goroutine |
 | 30 moliliang 2019 年 6 月 18 日 阿西吧。。 这代码怕是骗金币的吧。。 |
 | 31 useben 2019 年 6 月 18 日 go func() { for _, row := range arr { wg.Add(1) go func(row []string) { defer wg.Done() params, err := getQuery(row, *tplId, pk) if err != nil { fmt.Println(err) } paramsChan <- params }(row) } wg.Wait() close(paramsChan) }() 这是认真的吗。。。 每个 go 串起来了。。。 and 以后先检查下代码逻辑再提出疑问吧 |
 | 33 CEBBCAT 2019 年 6 月 18 日 via Android 1 能够在解决问题后把解法一并贴出的坛友越来越少了，赞楼主 贴代码可以用 gist |
 | 34 Cellei 2019 年 6 月 19 日 赞一个 |
Select Language