如何将返回含json.RawMessage结果的REST API输出推送至Kafka？

Got it, let's walk through how to push your REST API's output to Kafka, especially handling that tricky json.RawMessage field in your Response struct. Here's a step-by-step solution tailored to your needs:

整体流程 #

First, we'll fetch the API response, prepare the data (either the full response or just the raw result), then send it to your Kafka topic. Let's break it down with code examples.

1. 定义Response结构体并获取API数据 #

First, make sure your struct is properly defined, then use Go's HTTP client to fetch and parse the API response:

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "net/http"

    "github.com/segmentio/kafka-go"
)

// 你提供的Response结构体
type Response struct {
    RequestID      string          `json:"requestId"`
    Success        bool            `json:"success"`
    NextPageToken  string          `json:"nextPageToken,omitempty"`
    MoreResult     bool            `json:"moreResult,omitempty"`
    Errors         []struct {
        Code    string `json:"code"`
        Message string `json:"message"`
    } `json:"errors,omitempty"`
    Result         json.RawMessage `json:"result,omitempty"`
}

// FetchAPIResponse 调用REST API并返回解析后的Response
func FetchAPIResponse(apiURL string) (*Response, error) {
    resp, err := http.Get(apiURL)
    if err != nil {
        return nil, fmt.Errorf("调用API失败: %w", err)
    }
    defer resp.Body.Close()

    var apiResp Response
    if err := json.NewDecoder(resp.Body).Decode(&apiResp); err != nil {
        return nil, fmt.Errorf("解析API响应失败: %w", err)
    }
    return &apiResp, nil
}

2. 准备Kafka消息内容 #

根据需求，你有两种选择：

选项1：发送完整的API响应

如果你需要保留RequestID、成功状态等辅助信息，直接序列化整个结构体：

func PrepareFullResponseMessage(apiResp *Response) ([]byte, error) {
    return json.Marshal(apiResp)
}

选项2：仅发送核心Result数据

由于Result是json.RawMessage类型，它本身已经是原始的JSON字节，无需二次序列化，这是最高效的选择：

func PrepareResultMessage(apiResp *Response) ([]byte, error) {
    if len(apiResp.Result) == 0 {
        return nil, fmt.Errorf("Result字段为空")
    }
    // RawMessage已经是合法的JSON，直接返回字节即可
    return apiResp.Result, nil
}

3. 将消息发送至Kafka #

使用Kafka客户端（这里用kafka-go，你也可以用sarama等其他库）发送准备好的消息：

func SendToKafka(brokerAddr, topic string, message []byte) error {
    writer := kafka.NewWriter(kafka.WriterConfig{
        Brokers: []string{brokerAddr},
        Topic:   topic,
        // 根据生产环境需求调整以下配置：
        // RequiredAcks: kafka.RequireAll,
        // BatchSize: 100,
    })
    defer writer.Close()

    return writer.WriteMessages(context.Background(),
        kafka.Message{
            Value: message,
            // 可选：如果需要按请求ID分区，可以添加Key
            // Key: []byte(apiResp.RequestID),
        },
    )
}

4. 整合所有流程 #

把所有步骤整合到主函数（或服务方法）中：

func main() {
    // 配置信息 - 根据你的环境修改
    apiURL := "https://your-api-endpoint.com/data"
    kafkaBroker := "localhost:9092"
    kafkaTopic := "api-results-topic"

    // 步骤1：获取API数据
    apiResp, err := FetchAPIResponse(apiURL)
    if err != nil {
        fmt.Printf("获取API数据失败: %v\n", err)
        return
    }

    // 步骤2：准备消息（二选一）
    // message, err := PrepareFullResponseMessage(apiResp)
    message, err := PrepareResultMessage(apiResp)
    if err != nil {
        fmt.Printf("准备消息失败: %v\n", err)
        return
    }

    // 步骤3：发送至Kafka
    if err := SendToKafka(kafkaBroker, kafkaTopic, message); err != nil {
        fmt.Printf("发送至Kafka失败: %v\n", err)
        return
    }

    fmt.Println("消息成功发送至Kafka!")
}

生产环境关键注意事项 #

• 错误处理：把示例中的fmt.Printf替换为结构化日志（比如zap或logrus），并为API调用、Kafka发送添加重试逻辑。
• json.RawMessage的优势：这个类型避免了不必要的序列化开销，如果不需要修改Result数据，直接使用它的原始字节是最优选择。
• Kafka配置：生产环境中，设置RequiredAcks确保消息持久性，配置批量发送提升吞吐量，如果集群要求安全连接则启用TLS。
• Schema管理：如果消费者需要解析Result数据，建议使用Schema Registry（比如Confluent Schema Registry）配合Avro，优雅处理Schema变更。

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