Golang中如何分块处理HTTP响应数据

在Go中处理分块HTTP响应时，问题通常出现在读取响应体的方式上。你的代码使用了固定大小的缓冲区读取，但未正确处理分块编码的流式特性。Go的net/http客户端默认支持分块传输编码，但需要正确使用响应体的读取方式。

以下是修改后的代码示例，使用bufio.Scanner或持续读取来处理分块数据：

import (
    "bufio"
    "fmt"
    "io"
    "net/http"
    "net"
    "time"
    "sync"
)

func Stream() (StreamInterface, error) {
    roundTripper := &http.Transport{
        Proxy: http.ProxyFromEnvironment,
        DialContext: (&net.Dialer{
            Timeout:   60 * time.Second,
            KeepAlive: 60 * time.Second,
        }).DialContext,
    }
    httpClient := &http.Client{
        Transport: roundTripper,
        Timeout:   60 * time.Second,
    }

    httpRequest, err := http.NewRequest("GET", "http://localhost:8080/api/v1/stream", nil)
    if err != nil {
        return nil, err
    }
    httpRequest.Header.Set("Accept", "application/json")

    resp, err := httpClient.Do(httpRequest)
    if err != nil {
        return nil, err
    }

    if resp.StatusCode != http.StatusOK {
        resp.Body.Close()
        return nil, fmt.Errorf("wrong status code %d", resp.StatusCode)
    }

    dec := newDecoder(resp.Body)
    return NewStream(dec), nil
}

type decoder struct {
    r io.ReadCloser
}

func newDecoder(body io.ReadCloser) *decoder {
    return &decoder{
        r: body,
    }
}

func (d *decoder) close() error {
    return d.r.Close()
}

func (d *decoder) decode() error {
    scanner := bufio.NewScanner(d.r)
    for scanner.Scan() {
        data := scanner.Bytes()
        fmt.Printf("Received chunk: %s\n", string(data))
    }
    if err := scanner.Err(); err != nil {
        return err
    }
    return nil
}

type StreamInterface interface {
    Stop()
}

type streamImpl struct {
    mutex   sync.Mutex
    source  *decoder
    stopped bool
}

func NewStream(source *decoder) StreamInterface {
    stream := &streamImpl{
        source: source,
    }
    go stream.receive()
    return stream
}

func (s *streamImpl) Stop() {
    s.mutex.Lock()
    defer s.mutex.Unlock()
    s.stopped = true
}

func (s *streamImpl) isStopped() bool {
    s.mutex.Lock()
    defer s.mutex.Unlock()
    return s.stopped
}

func (s *streamImpl) receive() {
    defer s.source.close()
    for {
        if s.isStopped() {
            return
        }
        err := s.source.decode()
        if err != nil {
            if err != io.EOF {
                fmt.Printf("Decode error: %v\n", err)
            }
            return
        }
    }
}

关键修改点：

1. 使用bufio.Scanner逐行读取分块数据，假设服务器以换行符分隔JSON对象。
2. 移除固定缓冲区读取，避免阻塞直到所有数据到达。
3. 保持连接处理，确保在流结束时正确关闭。

如果服务器发送的是纯分块数据（不带换行符），可以使用以下替代decode方法：

func (d *decoder) decode() error {
    buffer := make([]byte, 1024)
    for {
        n, err := d.r.Read(buffer)
        if n > 0 {
            fmt.Printf("Received chunk: %s\n", string(buffer[:n]))
        }
        if err == io.EOF {
            return nil
        }
        if err != nil {
            return err
        }
    }
}

这允许在数据块到达时立即处理它们，模拟curl的行为。确保服务器正确实现分块传输，并且客户端超时设置足够长以避免中断。