netProgram/tcp_server.go

package netProgram

import (
	"bytes"
	"context"
	"encoding/gob"
	"encoding/json"
	"errors"
	"io"
	"log"
	"math/rand"
	"net"
	"sync"
	"time"
)

const tcp = "tcp"

// 服务端
func TcpServer() {
	// A. 基于某个地址建立监听
	// 服务端地址
	//address := "127.0.0.1:5678"
	// 端口省略，会使用随机端口
	//address := "127.0.0.1:"
	// IP省略，使用服务器全部的可用IP

	// tcp协议类型
	//address := "127.0.0.1:5678" // IPv4
	//address := "[::1]:5678" // IPv6
	address := ":5678" // Any IP or version
	listener, err := net.Listen(tcp, address)
	if err != nil {
		log.Fatalln(err)
	}
	// 关闭监听
	defer listener.Close()
	log.Printf("%s server is listening on %s\n", tcp, listener.Addr())

	// B. 接受连接请求
	// 循环接受
	for {
		// 阻塞接受
		conn, err := listener.Accept()
		if err != nil {
			log.Println(err)
		}

		// 处理连接，读写
		// 日志连接的远程地址（client addr）
		log.Printf("accept from %s\n", conn.RemoteAddr())
	}
}

// 基本读写操作
func TcpServerRW() {
	// A. 基于某个地址建立监听
	// 服务端地址
	address := ":5678" // Any IP or version
	listener, err := net.Listen(tcp, address)
	if err != nil {
		log.Fatalln(err)
	}
	// 关闭监听
	defer listener.Close()
	log.Printf("%s server is listening on %s\n", tcp, listener.Addr())

	// B. 接受连接请求
	// 循环接受
	for {
		// 阻塞接受
		conn, err := listener.Accept()
		if err != nil {
			log.Println(err)
		}

		// 处理连接，读写
		go HandleConn(conn)
	}
}

// 服务端
func TcpBacklogServer() {
	// A. 基于某个地址建立监听
	// 服务端地址
	address := ":5678" // Any IP or version
	listener, err := net.Listen(tcp, address)
	if err != nil {
		log.Fatalln(err)
	}
	// 关闭监听
	defer listener.Close()
	log.Printf("%s server is listening on %s\n", tcp, listener.Addr())

	// B. 接受连接请求
	// 循环接受
	for {
		// 阻塞接受
		conn, err := listener.Accept()
		if err != nil {
			log.Println(err)
		}

		// 处理连接，读写
		go func(conn net.Conn) {
			// 日志连接的远程地址（client addr）
			log.Printf("accept from %s\n", conn.RemoteAddr())
			time.Sleep(time.Second)
		}(conn)
	}
}

// 处理每个连接
func HandleConn(conn net.Conn) {
	// 日志连接的远程地址（client addr）
	log.Printf("accept from %s\n", conn.RemoteAddr())

	// A.保证连接关闭
	defer conn.Close()

	// B.向客户端发送数据，SerWrite
	wn, err := conn.Write([]byte("send some data from server" + "\n"))
	if err != nil {
		log.Println(err)
	}
	log.Printf("server write len is %d\n", wn)

	// C.从客户端接收数据，SerRead
	buf := make([]byte, 1024)
	rn, err := conn.Read(buf)
	if err != nil {
		log.Println(err)
	}
	log.Println("received from client data is:", string(buf[:rn]))
}

// 基本读写操作
func TcpW() {
	// A. 基于某个地址建立监听
	// 服务端地址
	address := ":5678" // Any IP or version
	listener, err := net.Listen(tcp, address)
	if err != nil {
		log.Fatalln(err)
	}
	// 关闭监听
	defer listener.Close()
	log.Printf("%s server is listening on %s\n", tcp, listener.Addr())

	// B. 接受连接请求
	// 循环接受
	for {
		// 阻塞接受
		conn, err := listener.Accept()
		if err != nil {
			log.Println(err)
		}

		// 处理连接，读写
		go HandleConnW(conn)
	}
}

// 处理每个连接
func HandleConnW(conn net.Conn) {
	// 日志连接的远程地址（client addr）
	log.Printf("accept from %s\n", conn.RemoteAddr())

	// A.保证连接关闭
	defer conn.Close()

	// 1. 严谨的判断是否写入成功
	//data := []byte("send some data from server" + "\n")
	//wn, err := conn.SerWrite(data)
	//if err != nil {
	//	log.Println(err)
	//}
	//// 若要严谨的判断是否写入成功,需要:
	//if err == nil && wn == len(data) {
	//	log.Println("write success")
	//}
	//log.Printf("server write len is %d\n", wn)

	// 2. 写操作会被阻塞
	//for i := 0; i < 300000; i++ {
	//	data := []byte("send some data from server" + "\n")
	//	wn, err := conn.SerWrite(data)
	//	if err != nil {
	//		log.Fatalln(err)
	//	}
	//	log.Printf("%d, server write len is %d\n", i, wn)
	//}

	// 不执行任何写操作
	//time.Sleep(5 * time.Second)

	// 写入一次
	data := []byte("send some data from server" + "\n")
	wn, err := conn.Write(data)
	if err != nil {
		log.Println(err)
	}
	// 若要严谨的判断是否写入成功,需要:
	if err == nil && wn == len(data) {
		log.Println("write success")
	}
	log.Printf("server write len is %d\n", wn)
}

// 并发的读和写操作，全双工
func TcpServerRWConcurrency() {
	// A. 基于某个地址建立监听
	// 服务端地址
	address := ":5678" // Any IP or version
	listener, err := net.Listen(tcp, address)
	if err != nil {
		log.Fatalln(err)
	}
	// 关闭监听
	defer listener.Close()
	log.Printf("%s server is listening on %s\n", tcp, listener.Addr())

	// B. 接受连接请求
	// 循环接受
	for {
		// 阻塞接受
		conn, err := listener.Accept()
		if err != nil {
			log.Println(err)
		}

		// 处理连接，读写
		go HandleConnConcurrency(conn)
	}
}

// 处理每个连接
func HandleConnConcurrency(conn net.Conn) {
	// 日志连接的远程地址（client addr）
	log.Printf("accept from %s\n", conn.RemoteAddr())
	// A.保证连接关闭
	defer conn.Close()

	wg := sync.WaitGroup{}
	// 并发的写
	wg.Add(1)
	go SerWrite(conn, &wg, "abcd")
	wg.Add(1)
	go SerWrite(conn, &wg, "efgh")
	wg.Add(1)
	go SerWrite(conn, &wg, "ijkl")

	// 并发的读
	wg.Add(1)
	go SerRead(conn, &wg)

	wg.Wait()
}

func SerWrite(conn net.Conn, wg *sync.WaitGroup, data string) {
	defer wg.Done()
	// B.向客户端发送数据，SerWrite
	for {
		wn, err := conn.Write([]byte(data + "\n"))
		if err != nil {
			log.Println(err)
		}
		log.Printf("server write len is %d\n", wn)
		time.Sleep(1 * time.Second)
	}
}

func SerRead(conn net.Conn, wg *sync.WaitGroup) {
	defer wg.Done()
	for {
		// C.从客户端接收数据，SerRead
		buf := make([]byte, 1024)
		rn, err := conn.Read(buf)
		if err != nil {
			log.Println(err)
		}
		log.Println("received from client data is:", string(buf[:rn]))
	}
}

// 格式化传输
func TcpServerFormat() {
	// A. 基于某个地址建立监听
	// 服务端地址
	address := ":5678" // Any IP or version
	listener, err := net.Listen(tcp, address)
	if err != nil {
		log.Fatalln(err)
	}
	// 关闭监听
	defer listener.Close()
	log.Printf("%s server is listening on %s\n", tcp, listener.Addr())

	// B. 接受连接请求
	// 循环接受
	for {
		// 阻塞接受
		conn, err := listener.Accept()
		if err != nil {
			log.Println(err)
		}

		// 处理连接，读写
		go HandleConnFormat(conn)
	}
}

func HandleConnFormat(conn net.Conn) {
	// 日志连接的远程地址（client addr）
	log.Printf("accept from %s\n", conn.RemoteAddr())
	// A.保证连接关闭
	defer conn.Close()

	wg := sync.WaitGroup{}
	wg.Add(1)
	// 发送端，
	go SerWriteFormat(conn, &wg)
	wg.Wait()
}

func SerWriteFormat(conn net.Conn, wg *sync.WaitGroup) {
	defer wg.Done()
	for {
		// 向客户端发送数据
		// 数据编码后发送

		// 创建需要传递的数据
		// 自定义的消息结构类型
		type Message struct {
			ID      uint   `json:"id,omitempty"`
			Code    string `json:"code,omitempty"`
			Content string `json:"content,omitempty"`
		}
		message := Message{
			ID:      uint(rand.Int()),
			Code:    "SERVER-STANDARD",
			Content: "message from server",
		}
		// 编码后数据的展示
		var buf bytes.Buffer
		encoderData := json.NewEncoder(&buf)
		//encoderData := gob.NewEncoder(&buf)
		if err := encoderData.Encode(message); err != nil {
			log.Println(err)
			continue
		}
		log.Println(buf.String())

		// 1, JSON, 文本编码
		//// 创建编码器
		//encoder := json.NewEncoder(conn)
		//// 利用编码器进行编码
		//// encode 成功后，会写入到conn，已经完成了conn.Write()
		//if err := encoder.Encode(message); err != nil {
		//	log.Println(err)
		//	continue
		//}
		//log.Println("message was send")

		// 2, GOB, 二进制编码
		// 创建编码器
		encoder := gob.NewEncoder(conn)
		// 利用编码器进行编码
		// encode 成功后，会写入到conn，已经完成了conn.Write()
		if err := encoder.Encode(message); err != nil {
			log.Println(err)
			continue
		}
		log.Println("message was send")

		time.Sleep(1 * time.Second)
	}
}

// 短连接编程示例
func TcpServerSort() {
	// A. 基于某个地址建立监听
	// 服务端地址
	address := ":5678" // Any IP or version
	listener, err := net.Listen(tcp, address)
	if err != nil {
		log.Fatalln(err)
	}
	// 关闭监听
	defer listener.Close()
	log.Printf("%s server is listening on %s\n", tcp, listener.Addr())

	// B. 接受连接请求
	// 循环接受
	for {
		// 阻塞接受
		conn, err := listener.Accept()
		if err != nil {
			log.Println(err)
		}

		// 处理连接，读写
		go HandleConnSort(conn)
	}
}

func HandleConnSort(conn net.Conn) {
	// 日志连接的远程地址（client addr）
	log.Printf("accept from %s\n", conn.RemoteAddr())
	// A.保证连接关闭
	defer conn.Close()

	wg := sync.WaitGroup{}
	wg.Add(1)
	// 发送端，
	go SerWriteSort(conn, &wg)
	wg.Wait()
}

func SerWriteSort(conn net.Conn, wg *sync.WaitGroup) {
	defer wg.Done()

	// 创建需要传递的数据
	// 自定义的消息结构类型
	type Message struct {
		ID      uint   `json:"id,omitempty"`
		Code    string `json:"code,omitempty"`
		Content string `json:"content,omitempty"`
	}
	message := Message{
		ID:      uint(rand.Int()),
		Code:    "SERVER-STANDARD",
		Content: "message from server",
	}

	// GOB, 二进制编码
	// 创建编码器
	encoder := gob.NewEncoder(conn)
	// 利用编码器进行编码
	// encode 成功后，会写入到conn，已经完成了conn.Write()
	if err := encoder.Encode(message); err != nil {
		log.Println(err)
		return
	}
	log.Println("message was send")
	log.Println("link will be close")
	return
}

// 心跳检测
func TcpServerHB() {
	// A. 基于某个地址建立监听
	// 服务端地址
	address := ":5678" // Any IP or version
	listener, err := net.Listen(tcp, address)
	if err != nil {
		log.Fatalln(err)
	}
	// 关闭监听
	defer listener.Close()
	log.Printf("%s server is listening on %s\n", tcp, listener.Addr())

	// B. 接受连接请求
	// 循环接受
	for {
		// 阻塞接受
		conn, err := listener.Accept()
		if err != nil {
			log.Println(err)
		}

		// 处理连接，读写
		go HandleConnHB(conn)
	}
}

func HandleConnHB(conn net.Conn) {
	// 日志连接的远程地址（client addr）
	log.Printf("accept from %s\n", conn.RemoteAddr())
	// A.保证连接关闭
	defer func() {
		conn.Close()
		log.Println("connection be closed")
	}()

	wg := sync.WaitGroup{}

	// 独立的goroutine，在连接建立后，周期发送ping
	wg.Add(1)
	// 发送ping
	go SerPing(conn, &wg)
	wg.Wait()
}

func SerPing(conn net.Conn, wg *sync.WaitGroup) {
	defer wg.Done()

	// 启动接收pong
	ctx, cancel := context.WithCancel(context.Background())
	go SerReadPong(conn, ctx)

	// ping失败的次数
	const maxPingNum = 3
	pingErrCounter := 0

	//周期性的发送
	//利用 time.Ticker
	ticker := time.NewTicker(2 * time.Second)
	for t := range ticker.C {
		pingMsg := MessageHB{
			ID:   uint(rand.Int()),
			Code: "PING-SERVER",
			Time: t,
		}

		// GOB, 二进制编码
		encoder := gob.NewEncoder(conn)
		// encode 成功后，会写入到conn，已经完成了conn.Write()
		if err := encoder.Encode(pingMsg); err != nil {
			log.Println(err)
			// 连接有问题的情况
			// 累加错误计数器
			pingErrCounter++
			// 判断是否到达上限
			if pingErrCounter == maxPingNum {
				// 心跳失败
				// 终止pong的处理
				cancel()
				return
			}
		}
		log.Printf("ping send to %s, ping id is %d\n", conn.RemoteAddr(), pingMsg.ID)
	}
}
func SerReadPong(conn net.Conn, ctx context.Context) {

	for {
		// 处理Ping结束
		select {
		case <-ctx.Done():
			return
		default:
			message := MessageHB{}
			// GOB解码
			decoder := gob.NewDecoder(conn)
			// 解码操作，从conn中读取内容，成功会将解码后的结果，赋值到message变量
			err := decoder.Decode(&message)
			// 错误 io.EOF 时，表示连接被给关闭
			if err != nil && errors.Is(err, io.EOF) {
				log.Println(err)
				break
			}
			// 判断是为为 pong 类型消息
			if message.Code == "PONG-CLIENT" {
				log.Printf("receive pong from %s, %s\n", conn.RemoteAddr(), message.Content)
			}
		}
	}
}

// 测试连接池服务端
func TcpServerPool() {
	// A. 基于某个地址建立监听
	// 服务端地址
	address := ":5678" // Any IP or version
	listener, err := net.Listen(tcp, address)
	if err != nil {
		log.Fatalln(err)
	}
	// 关闭监听
	defer listener.Close()
	log.Printf("%s server is listening on %s\n", tcp, listener.Addr())

	// B. 接受连接请求
	// 循环接受
	for {
		// 阻塞接受
		conn, err := listener.Accept()
		if err != nil {
			log.Println(err)
		}

		// 处理连接，读写
		go HandleConnPool(conn)
	}
}

func HandleConnPool(conn net.Conn) {
	// 日志连接的远程地址（client addr）
	log.Printf("accept from %s\n", conn.RemoteAddr())
	// A.保证连接关闭
	defer func() {
		conn.Close()
		log.Println("connection be closed")
	}()

	select {}
}