Go语言如何从Kafka REST Proxy返回消息中提取value字段映射到结构体

定义匹配JSON结构的结构体 #

首先声明两层结构体，分别对应返回的外层Kafka记录和你需要提取的value业务字段：

import "encoding/json"

// 自定义业务结构体，对应返回结果中value字段的结构
type AdoptionValue struct {
    AdoptionID string `json:"AdoptionID"`
    IPAddress  string `json:"IPAddress"`
    Port       string `json:"Port"`
    Status     string `json:"Status"`
}

// 外层结构体，对应Kafka REST代理返回的单条消息记录
type KafkaRecord struct {
    Topic     string        `json:"topic"`
    // 不需要的字段可以不用声明，JSON解码会自动忽略
    Value     AdoptionValue `json:"value"`
    Partition int           `json:"partition"`
    Offset    int64         `json:"offset"`
}

修改响应解析逻辑 #

不要用bufio.Scanner逐行扫描，直接用Go标准库的JSON解码器读取整个响应体完成映射：

// 原有请求构造逻辑保持不变
req, err := http.NewRequest(http.MethodGet, fmt.Sprintf(FETCH_CONSUMER, URL, GROUP, CONSUMER), nil)
if err != nil {
    panic(err)
}
req.Header.Add("Accept", CONTENT_TYPE)
respRecords, err := client.Do(req)
if err != nil {
    panic(err)
}
defer respRecords.Body.Close()
fmt.Printf("Response %s\n", respRecords.Status)

// 直接解码响应体到结构体数组
var records []KafkaRecord
if err = json.NewDecoder(respRecords.Body).Decode(&records); err != nil {
    panic(fmt.Errorf("JSON解码失败: %w", err))
}

// 遍历提取value字段映射后的结构体
for _, record := range records {
    businessData := record.Value
    // 此处可直接使用结构体的字段做业务逻辑处理
    fmt.Printf("业务数据：%+v\n", businessData)
}

注意事项 #

• 结构体所有字段首字母必须大写，否则JSON包无法访问私有字段，会出现解码后字段全为零值的问题
• 如果value字段可能为空，可以将KafkaRecord中Value的类型改为*AdoptionValue，避免空值解码报错
• 不需要用到的返回字段（比如示例中的key）无需在结构体中声明，解码时会自动忽略

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