112 lines
2.1 KiB
Go
112 lines
2.1 KiB
Go
|
package bytesSendUtil
|
|||
|
|
|
|||
|
|
import (
|
|||
|
|
"fmt"
|
|||
|
|
"time"
|
|||
|
|
|
|||
|
|
"goutil/debugUtil"
|
|||
|
|
"goutil/logUtil"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
// 负责发送数据的协程
|
|||
|
|
func sendLoop(s sender, closeSignal chan struct{}) {
|
|||
|
|
defer func() {
|
|||
|
|
if r := recover(); r != nil {
|
|||
|
|
logUtil.LogUnknownError(r)
|
|||
|
|
}
|
|||
|
|
}()
|
|||
|
|
|
|||
|
|
for {
|
|||
|
|
select {
|
|||
|
|
case <-s.Done():
|
|||
|
|
closeSignal <- struct{}{}
|
|||
|
|
return
|
|||
|
|
case v := <-s.Data():
|
|||
|
|
if err := s.Send(v); err != nil {
|
|||
|
|
// 发送失败存入缓存
|
|||
|
|
s.Cache() <- v
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// 定时重发失败的数据
|
|||
|
|
func resendLoop(s sender, folder string, closeSignal chan struct{}) {
|
|||
|
|
defer func() {
|
|||
|
|
if r := recover(); r != nil {
|
|||
|
|
logUtil.LogUnknownError(r)
|
|||
|
|
}
|
|||
|
|
}()
|
|||
|
|
|
|||
|
|
// debug模式每秒重试1次
|
|||
|
|
var delay time.Duration
|
|||
|
|
if debugUtil.IsDebug() {
|
|||
|
|
delay = time.Second
|
|||
|
|
} else {
|
|||
|
|
delay = time.Minute * 5
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// 定时重发失败数据
|
|||
|
|
for {
|
|||
|
|
select {
|
|||
|
|
case <-s.Done():
|
|||
|
|
closeSignal <- struct{}{}
|
|||
|
|
return
|
|||
|
|
case <-time.After(delay):
|
|||
|
|
sendCacheData(s, folder)
|
|||
|
|
loadData(s.(EnsureSender), folder)
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// 从sender获取失败数据重发
|
|||
|
|
func sendCacheData(s sender, folder string) {
|
|||
|
|
failed := make([]dataItem, 0)
|
|||
|
|
length := len(s.Cache())
|
|||
|
|
|
|||
|
|
defer func() {
|
|||
|
|
// 用于记录多次失败后放弃发送的数据
|
|||
|
|
giveUpItems := make(chan dataItem, len(failed))
|
|||
|
|
|
|||
|
|
for _, v := range failed {
|
|||
|
|
if v.Count() >= 3 {
|
|||
|
|
// 失败次数太多的数据准备存放到磁盘中
|
|||
|
|
giveUpItems <- v
|
|||
|
|
} else {
|
|||
|
|
s.Cache() <- v
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
giveUpLen := len(giveUpItems)
|
|||
|
|
if giveUpLen > 0 {
|
|||
|
|
// 将多次失败的数据保存到文件中
|
|||
|
|
if folder[len(folder)-1] == '/' {
|
|||
|
|
folder = folder[:len(folder)-1]
|
|||
|
|
}
|
|||
|
|
saveData(giveUpItems, folder+"_giveup")
|
|||
|
|
|
|||
|
|
if giveUpLen >= 5 {
|
|||
|
|
log := fmt.Sprintf("ensureSendUtil: 有%d条数据多次发送失败", giveUpLen)
|
|||
|
|
logUtil.NormalLog(log, logUtil.Error)
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// 输出信息
|
|||
|
|
log := fmt.Sprintf("ensureSendUtil: 重发%d条数据，失败%d条，存盘%d条\n", length, len(failed), giveUpLen)
|
|||
|
|
logUtil.NormalLog(log, logUtil.Info)
|
|||
|
|
}()
|
|||
|
|
|
|||
|
|
for {
|
|||
|
|
select {
|
|||
|
|
case v := <-s.Cache():
|
|||
|
|
// 重发数据
|
|||
|
|
if e := s.Send(v); e != nil {
|
|||
|
|
// 记录失败的数据
|
|||
|
|
failed = append(failed, v)
|
|||
|
|
}
|
|||
|
|
default:
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|