flow.New().Source(func(chVipDataGlow chan [][]string) {
defer close(chVipDataGlow)
file, err := os.Open(fileName)
fileInfo, err := os.Stat(fileName)
if err != nil || fileInfo.Size() == 0 {
return
}
defer func() {
file.Close()
os.Remove(fileName)
}()
if err != nil {
log.Info(cc).Msgf("企业%d, 打开SFTP文件失败: %v 路径是:%s", enterpriseId, err, fileName)
return
}
buf := bufio.NewReader(file)
num := 2000
// var tmp = make([][]string, 0, num
var tmp [][]string
var i = 0
for {
b, err := buf.ReadString('\n')
if err != nil && err != io.EOF {
log.Info(cc).Msgf("数据读取报错了%+v\n", err)
break
}
i++
cell := strings.Split(b, ",")
if i == 1 && cell[0] == "数据更新时间" {
continue
}
if len(cell) <= 1 {
fileCountNumbers += 1
continue
}
tmp = append(tmp, cell)
if len(tmp) == num {
fileCountNumbers += num
chVipDataGlow <- tmp
tmp = [][]string{}
}
if err != nil || err == io.EOF {
log.Info(cc).Msgf("数据读完了或者报错了%+v", err)
break
}
}
if len(tmp) > 0 {
fileCountNumbers += len(tmp)
chVipDataGlow <- tmp
tmp = [][]string{}
}
}, 10).Map(func(params [][]string) [][]string {
defer func() {
if errs := recover(); errs != nil {
log.Error(cc).Msgf("数据库插入执行异常了%+v", errs)
errDefer = append(errDefer, params...)
return
}
}()
if len(params) > 0 {
sucn, ern, _, errData := c.readDataToTableOneByOne(cc, params, enterpriseId)
errNums += ern
successNums += sucn
return errData
}
return params
}).Map(func(errItems [][]string) {
fmt.println(errItems)
}).Run()
