go并发-读文件->管道->写文件->优雅退出

并发读3个文件-开3个线程-插入管道-开1个线程写入新文件后优雅退出

package main

import (
	"bufio"
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"path/filepath"
	"runtime"
	"strings"
	"sync"
)

var ch = make(chan string, 1000000) //  读文件插入的隧道
var write_main = make(chan struct{}) // 优雅退出main协程的空隧道
var wg_1 = sync.WaitGroup{} //优先等待读文件协程全部结束后 在执行下一步操作

//读目录下所有txt文件获取path+文件名
func FileName(path string) (string, []string) {
	filename := []string{}
	subDirs, err := ioutil.ReadDir(path)
	// fmt.Println(subDirs)
	if err != nil {
		return "nil", nil
	}
	for _, dir := range subDirs {
		if dir.IsDir() {
			FileName(filepath.Join(path, dir.Name()))
		} else {
			if strings.HasSuffix(dir.Name(), ".txt") {
				// fmt.Println(dir.Name())
				filename = append(filename, dir.Name())
			}
		}
	}
	return path, filename
}

3个协程并发读文件
func ReadFile(path string, filename []string) {
	wg_1.Add(len(filename))

	for _, v := range filename {

		go func(v string) {
			//读文件并发
			path_filename := path + "/" + v

			fin, err := os.Open(path_filename)
			if err != nil {
				fmt.Printf("打开文件失败:%v\n", err)
			}
			defer fin.Close()

			reder := bufio.NewReader(fin)
			// Hostslice := []string{}

			for {
				line, err := reder.ReadString('\n')
				fmt.Println(line, "line")
				if err != nil {
					if err == io.EOF {
						break
					} else {
						fmt.Printf("读文件失败:%v\n", err)
					}
				} else {
					line = strings.TrimRight(line, "\n")
					ch <- line
					fmt.Println(line, "插入管道的每一行")
				}
			}
			wg_1.Done()
		}(v)
	}

}
func WriteFile(ch chan string) {
	f, err := os.OpenFile("dir/new.txt", os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0777)
	if err != nil {
		fmt.Println("打开文件失败", err)
	}
	defer f.Close()
	for {
		v, ok := <-ch
		if !ok {
			break
		} else {
			writer := bufio.NewWriter(f)
			writer.WriteString(v + "\n")
			writer.Flush()
			fmt.Println(v, "write")
		}
	}
	<-write_main
}

func main() {
	path, filename := FileName("dir") 
	ReadFile(path, filename)
	fmt.Println("当前存在的协程数量", runtime.NumGoroutine())
	go WriteFile(ch)
	wg_1.Wait()              //等读完文件
	close(ch)                //读完关闭文件
	write_main <- struct{}{} //先插入一条进管道，然后write执行后取出管道
	fmt.Println("当前存在的协程数量", runtime.NumGoroutine())

}