使用Golang并发发送TCP数据包的最佳实践

大家好！

感谢大家的评论！

我通过限制 make 命令中创建的通道数量解决了这个问题。 看起来在创建了大约 30,000 个 Go 协程后，程序就会失败。

祝大家周末愉快！

你的代码存在几个关键问题，导致在高并发场景下失败。主要问题是资源耗尽和goroutine管理不当。

问题分析

1. 套接字耗尽：每个goroutine创建独立的TCP连接，当超过系统限制时会出现"cannot assign requested address"错误
2. 内存泄漏：连接没有正确关闭
3. 并发控制缺失：一次性创建大量goroutine会导致资源竞争
4. 通道使用错误：每个goroutine向通道发送多个消息，但主程序只接收固定数量

改进方案

package main

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

type Config struct {
	Endpoint string
	Port     string
	Points   int
	Workers  int
}

func main() {
	config := getConfig()
	start := time.Now()

	fmt.Printf("Sending %d points to %s:%s using %d workers\n", 
		config.Points, config.Endpoint, config.Port, config.Workers)

	// 使用工作池模式控制并发
	results := make(chan string, config.Points)
	var wg sync.WaitGroup

	// 创建worker池
	for w := 0; w < config.Workers; w++ {
		wg.Add(1)
		go worker(w, config, results, &wg)
	}

	// 等待所有worker完成
	go func() {
		wg.Wait()
		close(results)
	}()

	// 收集结果
	count := 0
	for result := range results {
		count++
		if count%1000 == 0 {
			fmt.Printf("Processed %d/%d points\n", count, config.Points)
		}
		_ = result // 处理结果，这里简单丢弃
	}

	fmt.Printf("Total time: %v\n", time.Since(start))
}

func worker(id int, config Config, results chan<- string, wg *sync.WaitGroup) {
	defer wg.Done()

	// 每个worker复用连接
	conn, err := net.Dial("tcp", config.Endpoint+":"+config.Port)
	if err != nil {
		log.Printf("Worker %d: failed to connect: %v", id, err)
		return
	}
	defer conn.Close()

	// 设置连接超时
	conn.SetDeadline(time.Now().Add(30 * time.Second))

	// 每个worker处理一部分任务
	batchSize := config.Points / config.Workers
	start := id * batchSize
	end := start + batchSize
	if id == config.Workers-1 {
		end = config.Points // 最后一个worker处理剩余部分
	}

	for i := start; i < end; i++ {
		message := fmt.Sprintf("prod.monitoring.test 1 source=mysource %d\n", i)
		
		_, err := conn.Write([]byte(message))
		if err != nil {
			log.Printf("Worker %d: write failed: %v", id, err)
			// 尝试重新连接
			conn.Close()
			conn, err = net.Dial("tcp", config.Endpoint+":"+config.Port)
			if err != nil {
				log.Printf("Worker %d: reconnection failed: %v", id, err)
				return
			}
			conn.SetDeadline(time.Now().Add(30 * time.Second))
			continue
		}

		// 可选：读取响应（如果需要）
		// buf := make([]byte, 1024)
		// n, _ := conn.Read(buf)

		results <- fmt.Sprintf("Worker %d sent point %d", id, i)
	}
}

func getConfig() Config {
	reader := bufio.NewReader(os.Stdin)
	
	fmt.Print("Endpoint (default: server.server.com): ")
	endpoint, _ := reader.ReadString('\n')
	if endpoint == "\n" {
		endpoint = "server.server.com"
	} else {
		endpoint = endpoint[:len(endpoint)-1] // 去除换行符
	}

	fmt.Print("Port (default: 2878): ")
	port, _ := reader.ReadString('\n')
	if port == "\n" {
		port = "2878"
	} else {
		port = port[:len(port)-1]
	}

	fmt.Print("Total points to send: ")
	var points int
	fmt.Scanf("%d", &points)

	fmt.Print("Number of workers (default: 1000): ")
	var workers int
	_, err := fmt.Scanf("%d", &workers)
	if err != nil || workers <= 0 {
		workers = 1000
	}

	// 限制最大worker数量，避免资源耗尽
	if workers > 5000 {
		workers = 5000
		fmt.Printf("Warning: Limiting workers to %d to avoid resource exhaustion\n", workers)
	}

	return Config{
		Endpoint: endpoint,
		Port:     port,
		Points:   points,
		Workers:  workers,
	}
}

关键改进点

1. 工作池模式：限制并发goroutine数量，避免系统资源耗尽
2. 连接复用：每个worker复用TCP连接，减少套接字创建
3. 连接管理：正确关闭连接，设置超时时间
4. 错误处理：添加重连机制，避免单点失败影响整个测试
5. 资源限制：自动限制最大worker数量
6. 进度显示：定期显示处理进度

性能优化建议

如果需要更高性能，可以考虑以下额外优化：

// 使用连接池
type ConnectionPool struct {
	connections chan net.Conn
	factory     func() (net.Conn, error)
}

// 批量发送数据
func batchSend(conn net.Conn, messages []string) error {
	var buf []byte
	for _, msg := range messages {
		buf = append(buf, []byte(msg+"\n")...)
	}
	_, err := conn.Write(buf)
	return err
}

// 使用sync.Pool减少内存分配
var bufPool = sync.Pool{
	New: func() interface{} {
		return make([]byte, 1024)
	},
}

这个改进版本可以处理数百万个请求而不会导致系统资源耗尽。通过控制并发数量和合理复用资源，避免了原始代码中的套接字耗尽问题。