Go TCP实现及状态监控

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
package main

import (
	"bufio"
	"context"
	"fmt"
	"github.com/go-redis/redis/v8"
	"net"
	"strconv"
	"sync"
	"time"
)

var wg sync.WaitGroup
var ctx = context.Background()
var rdb *redis.Client

func online(deviceId string,quit chan int) {
	defer wg.Done()
	count := 0
	for {
		select {
		case <-quit:
			fmt.Println("收到退出信号")
			err := rdb.Set(ctx, deviceId, false, 20*time.Second).Err()
			if err != nil {
				panic(err)
			}
			return
		default:
			// 在这里写更新TCP心跳函数
			fmt.Println("在线" + strconv.Itoa(count))
			count++

			err := rdb.Set(ctx, deviceId, true, 20*time.Second).Err()
			if err != nil {
				panic(err)
			}

			fmt.Println("没收到信号")
		}
		time.Sleep(1 * time.Second)
	}
}

// 处理函数
func process(conn net.Conn) {
	defer conn.Close() // 关闭连接
	fmt.Println("建立连接")

	// 退出通道
	quit := make(chan int)

	wg.Add(1)
	// 创建一个可以让函数方法只执行一次并发辅助对象
	var once sync.Once

	for {
		deviceId := time.Now().Format("150105")
		// 初始化协程，用于更新tcp链接状态
		once.Do(func() {
			go online(deviceId,quit)
		})
		reader := bufio.NewReader(conn)
		var buf [128]byte
		n, err := reader.Read(buf[:]) // 读取数据
		if err != nil {
			fmt.Println("客户端关闭")

			// 向通道发送关闭协程信号
			quit <- 1
			wg.Wait()
			fmt.Println("read from client failed, err:", err)
			return
		}

		recvStr := string(buf[:n])
		fmt.Println("收到client端发来的数据：", recvStr)
		conn.Write([]byte(recvStr)) // 发送数据
	}
}

func main() {
	rdb = redis.NewClient(&redis.Options{
		Addr:        "localhost:6379",
		Password:    "",  // no password set
		DB:          0,   // use default DB
		IdleTimeout: 300, // 默认Idle超时时间
		PoolSize:    100, // 连接池
	})
	// 验证是否连接到redis服务端
	res, err := rdb.Ping(ctx).Result()
	if err != nil {
		fmt.Printf("Connect Failed! Err: %vn", err)
	} else {
		fmt.Printf("Connect Successful! Ping => %vn", res)
	}

	listen, err := net.Listen("tcp", "127.0.0.1:20000")
	if err != nil {
		fmt.Println("listen failed, err:", err)
		return
	}
	for {
		conn, err := listen.Accept() // 建立连接
		if err != nil {
			fmt.Println("accept failed, err:", err)
			continue
		}
		go process(conn) // 启动一个goroutine处理连接
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
package main

import (
	"bufio"
	"fmt"
	"net"
	"os"
	"strings"
)

// 客户端
func main() {
	conn, err := net.Dial("tcp", "127.0.0.1:20000")
	if err != nil {
		fmt.Println("err :", err)
		return
	}
	defer conn.Close() // 关闭连接
	inputReader := bufio.NewReader(os.Stdin)
	for {
		input, _ := inputReader.ReadString('n') // 读取用户输入
		inputInfo := strings.Trim(input, "rn")
		if strings.ToUpper(inputInfo) == "Q" { // 如果输入q就退出
			return
		}
		_, err = conn.Write([]byte(inputInfo)) // 发送数据
		if err != nil {
			return
		}
		buf := [512]byte{}
		n, err := conn.Read(buf[:])
		if err != nil {
			fmt.Println("recv failed, err:", err)
			return
		}
		fmt.Println(string(buf[:n]))
	}
}