yunfantec
/
mq
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

增加重连

This commit is contained in:
qiyunfanbo126.com 2025-01-23 16:32:02 +08:00
parent 6da589a937
commit 46adc9c617
2 changed files with 3 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
app/http/tcppool/single.go
View File

@ -127,7 +127,7 @@ func (t *TcpHelper) resend() {
var nowTime = time.Now().Unix() var nowTime = time.Now().Unix()
for _, v := range rs { for _, v := range rs {
json.Unmarshal([]byte(v), &data) json.Unmarshal([]byte(v), &data)
if (data["send_time"].(float64) - float64(nowTime)) > 60 { if (float64(nowTime) - data["send_time"].(float64)) > 60 {
t.SendMsg([]byte(v)) t.SendMsg([]byte(v))
} }
} }

2
app/utils/mq/kafka_v2.go
View File

@ -103,6 +103,8 @@ func (kk KafkaV2Mq) Consume(name string, hand interface{}, ci int) {
} else { } else {
//utils.Log(nil, "对列放开") //utils.Log(nil, "对列放开")
} }
time.Sleep(3 * time.Minute)
msg, err := consumer.ReadMessage(1 * time.Second) msg, err := consumer.ReadMessage(1 * time.Second)
if err == nil { if err == nil {
utils.Log(nil, "offset", msg.TopicPartition.Offset) utils.Log(nil, "offset", msg.TopicPartition.Offset)