Go服务端如何用io.Pipe接收文件并流式传输至处理服务

嘿，第一次提问完全不用紧张！我刚好能帮你解决这个流式转发的问题——其实核心就是利用Go的io.Pipe把上传服务接收到的请求体直接流式传给处理服务，不用等整个文件上传完再处理，刚好符合你“收到第一个字节就开始处理”的需求。

用io.Pipe创建一个内存管道：上传服务一边从客户端读取请求体（图片数据），一边把数据写入管道的Writer端；同时启动goroutine，把管道的Reader端作为请求体发送给处理服务。这样处理服务就能实时收到数据并启动处理流程。

这个服务负责接收客户端的图片上传请求，然后流式转发给处理服务：

package main

import (
	"io"
	"log"
	"net/http"
)

func uploadHandler(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "仅支持POST请求", http.StatusMethodNotAllowed)
		return
	}

	// 创建io.Pipe：reader用于传给处理服务，writer用于写入上传的图片数据
	pipeReader, pipeWriter := io.Pipe()
	defer pipeReader.Close()

	// 启动goroutine：将客户端上传的实时数据写入管道
	go func() {
		defer pipeWriter.Close()
		// 实时复制请求体数据到管道，一收到字节就会传给处理服务
		_, err := io.Copy(pipeWriter, r.Body)
		if err != nil {
			log.Printf("读取上传数据失败: %v", err)
		}
		defer r.Body.Close()
	}()

	// 构造转发到处理服务的请求
	req, err := http.NewRequest(http.MethodPost, "http://处理服务地址/process", pipeReader)
	if err != nil {
		http.Error(w, "创建转发请求失败", http.StatusInternalServerError)
		log.Printf("创建请求出错: %v", err)
		return
	}
	// 传递原请求的Content-Type等元数据（如果需要）
	req.Header.Set("Content-Type", r.Header.Get("Content-Type"))

	// 发送流式请求到处理服务
	client := &http.Client{}
	resp, err := client.Do(req)
	if err != nil {
		http.Error(w, "转发到处理服务失败", http.StatusInternalServerError)
		log.Printf("转发请求出错: %v", err)
		return
	}
	defer resp.Body.Close()

	// 将处理服务的响应返回给客户端
	w.WriteHeader(resp.StatusCode)
	_, err = io.Copy(w, resp.Body)
	if err != nil {
		log.Printf("返回响应出错: %v", err)
	}
}

func main() {
	http.HandleFunc("/upload", uploadHandler)
	log.Fatal(http.ListenAndServe(":8080", nil))
}

这个服务负责实时接收流式数据并处理图片：

package main

import (
	"io"
	"log"
	"net/http"
)

func processHandler(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "仅支持POST请求", http.StatusMethodNotAllowed)
		return
	}

	// 实时读取流式传输的图片数据，模拟马赛克处理流程
	buf := make([]byte, 1024)
	totalBytes := 0
	for {
		n, err := r.Body.Read(buf)
		if err != nil && err != io.EOF {
			log.Printf("读取处理数据失败: %v", err)
			http.Error(w, "图片处理失败", http.StatusInternalServerError)
			return
		}
		if n == 0 {
			break
		}
		totalBytes += n
		log.Printf("已接收 %d 字节，累计接收 %d 字节", n, totalBytes)
		// 在这里加入你的照片马赛克处理逻辑，直接对buf[:n]进行实时处理
	}

	// 处理完成后返回响应
	w.WriteHeader(http.StatusOK)
	w.Write([]byte("图片马赛克处理完成"))
}

func main() {
	http.HandleFunc("/process", processHandler)
	log.Fatal(http.ListenAndServe(":8081", nil))
}

• io.Pipe的作用：它把io.Reader和io.Writer绑定在一起，写入Writer的数据会立即被Reader读取，完全不需要中间缓存整个文件，完美实现流式传输。
• goroutine的必要性：io.Copy（读取上传数据写入管道）和client.Do（从管道读取数据发送给处理服务）会互相阻塞，所以需要把其中一个操作放到goroutine中并发执行。
• App Engine适配：如果是App Engine标准环境，建议使用urlfetch.Client(r.Context())代替标准http.Client（标准环境对网络请求有特殊限制）；灵活环境可以直接用标准Client。
• 测试方式：用Postman发送POST请求到http://localhost:8080/upload，选择「binary」模式上传图片，就能看到处理服务实时打印接收的字节数。

内容的提问来源于stack exchange，提问作者Valentin