增加连接修复
This commit is contained in:
parent
141682065f
commit
7dc125cc22
|
@ -13,8 +13,11 @@ import (
|
|||
"time"
|
||||
)
|
||||
|
||||
var lock sync.Once
|
||||
var TcpPoolFactory = &TcpPool{Full: new(int32)}
|
||||
var (
|
||||
fullNum int32 = 1
|
||||
lock sync.Once
|
||||
TcpPoolFactory = &TcpPool{Full: &full}
|
||||
)
|
||||
|
||||
type TcpPool struct {
|
||||
client easypool.Pool
|
||||
|
|
|
@ -1,9 +1,7 @@
|
|||
package tcppool
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"quenue/app/utils"
|
||||
"quenue/config"
|
||||
|
@ -29,55 +27,22 @@ func (t *TcpHelper) Init(port string) *TcpHelper {
|
|||
var conn, err = net.Dial("tcp", "192.168.110.50:"+port)
|
||||
if err == nil {
|
||||
t.client = conn
|
||||
//t.watch(t.client)
|
||||
t.watch(t.client)
|
||||
}
|
||||
})
|
||||
return t
|
||||
}
|
||||
func (t *TcpHelper) handRead(conn net.Conn) {
|
||||
//defer func() {
|
||||
// if err := recover(); err != nil {
|
||||
// utils.Log(nil, "tcp read err", err)
|
||||
// }
|
||||
//}()
|
||||
for {
|
||||
if time.Now().Unix()-t.lastTime > 10 {
|
||||
//t.isDie = true
|
||||
//t.client.Close()
|
||||
//t.client = nil
|
||||
//return
|
||||
}
|
||||
fmt.Println("read")
|
||||
reader := bufio.NewReader(conn)
|
||||
var buffer [256]byte
|
||||
// 持续读取数据
|
||||
n, err := reader.Read(buffer[:])
|
||||
if err == io.EOF {
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
//if t.client != nil {
|
||||
// t.client.Close()
|
||||
// t.client = nil
|
||||
//}
|
||||
utils.Log(nil, "Error reading data:", err)
|
||||
continue
|
||||
|
||||
}
|
||||
recvStr := string(buffer[:n])
|
||||
if recvStr == "1" {
|
||||
atomic.AddInt32(t.Full, 1)
|
||||
} else if recvStr == "ping" {
|
||||
conn.Write([]byte("pong"))
|
||||
t.lastTime = time.Now().Unix()
|
||||
}
|
||||
func (t *TcpHelper) reconnect(port string) {
|
||||
var conn, err = net.Dial("tcp", "192.168.110.50:"+port)
|
||||
if err == nil {
|
||||
t.client = conn
|
||||
t.watch(t.client)
|
||||
}
|
||||
select {}
|
||||
}
|
||||
func (t *TcpHelper) SendMsg(msg []byte) error {
|
||||
_, err := t.Init(config.GetConf().OrderPort).client.Write(msg)
|
||||
|
||||
var buffer [256]byte
|
||||
var buffer [1]byte
|
||||
// 持续读取数据
|
||||
n, err := t.client.Read(buffer[:])
|
||||
if err == nil {
|
||||
|
@ -99,5 +64,19 @@ func (t *TcpHelper) Close(conn net.Conn) {
|
|||
t.client.Close()
|
||||
}
|
||||
func (t *TcpHelper) watch(conn net.Conn) {
|
||||
go t.handRead(conn)
|
||||
go func() {
|
||||
fmt.Println("watch")
|
||||
for {
|
||||
conn.SetWriteDeadline(time.Now().Add(time.Second * 3))
|
||||
_, err := conn.Write([]byte("1"))
|
||||
|
||||
if err != nil {
|
||||
utils.Log(nil, "连接关闭", err)
|
||||
t.client.Close()
|
||||
t.reconnect(config.GetConf().OrderPort)
|
||||
return
|
||||
}
|
||||
time.Sleep(time.Second)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
|
|
@ -64,9 +64,11 @@ func (kk KafkaV2Mq) Consume(name string, hand interface{}, ci int) {
|
|||
for i := 0; i < config.GetConf().Num; i++ {
|
||||
go func() {
|
||||
for {
|
||||
if atomic.LoadInt32(tcppool.TcpPoolFactory.Full) == 1 {
|
||||
utils.Log(nil, "tcp is die")
|
||||
time.Sleep(1 * time.Second)
|
||||
fmt.Println(atomic.LoadInt32(tcppool.TcpFactory.Full))
|
||||
if atomic.LoadInt32(tcppool.TcpFactory.Full) == 1 {
|
||||
utils.Log(nil, "对列阻塞")
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
continue
|
||||
}
|
||||
msg, err := consumer.ReadMessage(-1)
|
||||
fmt.Println("read msg", ci)
|
||||
|
|
Loading…
Reference in New Issue