Golang实现TCP服务器实时处理不定字节数据流

在处理TCP流数据时，特别是当数据包大小不固定时，确实需要一种更灵活的方法。你的当前实现使用固定大小的缓冲区，这可能导致数据截断或处理不完整消息的问题。以下是改进方案：

package main

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

const (
	CONN_HOST = "localhost"
	CONN_PORT = "9000"
	CONN_TYPE = "tcp"
	CONN_URL  = CONN_HOST + ":" + CONN_PORT
)

func main() {
	l, err := net.Listen(CONN_TYPE, CONN_URL)
	if err != nil {
		fmt.Println("Error listening:", err.Error())
		os.Exit(1)
	}
	defer l.Close()

	fmt.Println("Listening on " + CONN_URL)
	for {
		conn, err := l.Accept()
		if err != nil {
			fmt.Println("Error accepting connection:", err.Error())
			continue
		}
		go handleRequest(conn)
	}
}

func handleRequest(conn net.Conn) {
	defer conn.Close()
	
	// 设置读取超时以避免永久阻塞
	conn.SetReadDeadline(time.Now().Add(30 * time.Second))
	
	reader := bufio.NewReader(conn)
	
	for {
		// 读取直到遇到分隔符（这里使用换行符作为示例）
		data, err := reader.ReadBytes('\n')
		if err != nil {
			if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
				fmt.Println("Read timeout")
			} else {
				fmt.Println("Error reading:", err.Error())
			}
			break
		}
		
		// 重置读取超时
		conn.SetReadDeadline(time.Now().Add(30 * time.Second))
		
		// 处理数据（这里只是打印，实际应该发送到Kafka）
		if len(data) > 0 {
			message := string(data[:len(data)-1]) // 移除分隔符
			fmt.Printf("Received message: %s\n", message)
			fmt.Printf("Message length: %d bytes\n", len(data))
			
			// 这里添加Kafka生产逻辑
			// sendToKafka(message)
		}
	}
}

如果数据没有明确的分隔符，可以使用以下基于长度前缀的方法：

func handleRequestWithLengthPrefix(conn net.Conn) {
	defer conn.Close()
	
	conn.SetReadDeadline(time.Now().Add(30 * time.Second))
	
	for {
		// 先读取4字节的长度前缀
		lengthBuf := make([]byte, 4)
		_, err := conn.Read(lengthBuf)
		if err != nil {
			fmt.Println("Error reading length prefix:", err.Error())
			break
		}
		
		// 解析消息长度（大端序）
		messageLength := binary.BigEndian.Uint32(lengthBuf)
		
		// 读取实际消息数据
		messageBuf := make([]byte, messageLength)
		bytesRead := 0
		
		for bytesRead < int(messageLength) {
			n, err := conn.Read(messageBuf[bytesRead:])
			if err != nil {
				fmt.Println("Error reading message data:", err.Error())
				return
			}
			bytesRead += n
		}
		
		// 重置读取超时
		conn.SetReadDeadline(time.Now().Add(30 * time.Second))
		
		// 处理完整消息
		message := string(messageBuf)
		fmt.Printf("Received message: %s\n", message)
		fmt.Printf("Message length: %d bytes\n", messageLength)
		
		// 这里添加Kafka生产逻辑
		// sendToKafka(message)
	}
}

对于完全流式处理，可以使用缓冲区累积数据：

func handleRequestStreaming(conn net.Conn) {
	defer conn.Close()
	
	conn.SetReadDeadline(time.Now().Add(30 * time.Second))
	
	buffer := make([]byte, 0, 4096) // 初始容量4KB
	tempBuf := make([]byte, 256)    // 每次读取的临时缓冲区
	
	for {
		n, err := conn.Read(tempBuf)
		if err != nil {
			if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
				fmt.Println("Read timeout")
			} else {
				fmt.Println("Error reading:", err.Error())
			}
			break
		}
		
		// 将新数据追加到缓冲区
		buffer = append(buffer, tempBuf[:n]...)
		
		// 处理缓冲区中的完整消息（假设以换行符分隔）
		for {
			newlineIndex := -1
			for i, b := range buffer {
				if b == '\n' {
					newlineIndex = i
					break
				}
			}
			
			if newlineIndex == -1 {
				break // 没有完整消息，继续读取
			}
			
			// 提取完整消息
			message := buffer[:newlineIndex]
			buffer = buffer[newlineIndex+1:]
			
			// 处理消息
			fmt.Printf("Received message: %s\n", string(message))
			fmt.Printf("Message length: %d bytes\n", len(message))
			
			// 这里添加Kafka生产逻辑
			// sendToKafka(string(message))
		}
		
		// 重置读取超时
		conn.SetReadDeadline(time.Now().Add(30 * time.Second))
	}
}

关于TCP连接保持：对于实时数据流，保持连接开放是标准做法，避免了频繁建立和断开连接的开销。你的方法在架构上是正确的，只需要优化数据读取逻辑以适应不定长的数据流。