在 Go 中过滤 json 流中的非 json 内容

Question

我正在使用 Go 中的 json 结构输入流。我在我的标准输入上接收到来自另一个应用程序的输入流，但我无法更改通信协议。

我遇到的问题是每个 json 结构都由一个非 json 字符串行终止：“end”（不带引号）。

我正在使用 Golang 编码器/json 包来解码我从标准输入接收到的 json。问题是解码器在我第二次使用 msg 调用它时产生错误：“invalid character 'e' looking for beginning of value”。

问题当然是“end”字符串不是 json 编码的。我想知道如何让 Go 的 json 解码器跳过这个字符串？

一些示例输入：

{"command": "ack", "id": "1231231"}
end
{"command": "fail", "id": "1231231"}
end
{
    "command": "log",
    // the message to log
    "msg": "hello world!"
}
end

我尝试过的事情：

• 我声明：endStr := make([]byte, 10)
• 我尝试使用 fmt.Fscanf(os.Stdin, "%s", endStr) 来读取字符串，但没有读取任何数据。
• 我尝试使用 os.Stdin.Read(endStr)，但它也没有返回任何数据。
• 读取第一个 json 结构后，dec.Buffered() 返回一个包含“end”字符串的 io.Reader，但我不知道如何告诉解码器跳过这个。

任何帮助将不胜感激。

Answer 1

所以我能想到的最佳解决方案是：

1. 抛弃 json 解码器，
2. 从标准输入读取一个字节切片，
3. 修剪切片以排除 ("\nend\n") 字符串
4. 将修剪后的切片传递给 json Unmarshaller

我必须写的代码：

// Create a buffer to hold the stream data
data := make([]byte, 5000)

// Read data from stdin in a loop
for {
    _, err = os.Stdin.Read(data)
    if err != nil {
        panic(err)
    }

    index := bytes.Index(data, []byte("\n"))
    data = data[:index]

    var myStruct MyStruct
    err = json.Unmarshal(data, &myStruct)
    if err != nil {
        panic(err)
    }

    //(Do something with myStruct)
}

Answer 2

这很麻烦，但它会解决问题：

package main

import "fmt"
import "encoding/json"
import "bytes"
import "io"
import "bufio"

var input = `{"command": "ack", "id": "1231231"}
end
{"command": "fail", "id": "1231231"}
end
`

func main() {
    // make an io.Reader out of our input constant
    var input = bytes.NewBuffer([]byte(input))
    // we're going to need a buffered reader so we can Peek
    var buf = bufio.NewReader(input)

    // This is the result of the decode.  Use whatever makes sense for you
    var res map[string]interface{}
    var err error
    var dec *json.Decoder
    // We're going to loop until we get an error (hopefully it will be io.EOF
    for err == nil {
        if dec != nil {
            // This is the tricky bit:  json.Decoder has its own buffer.
            // it will read more than the data it needs.  In my simple test, 
            // it buffers all of the data.  What we're doing here is constructing
            // a new bufio.Reader using the remaining bytes in the json decoder's buffer
            // and whatever hasn't been read from our original buffer.
            buf = bufio.NewReader(io.MultiReader(dec.Buffered(), buf))
        }
        // Now let's try to drop an 'end' statement from the buffer
        dropEnd(buf)
        // We need a new json.Decoder each time since the old one contains unusable
        // data in its internal buffer.
        dec = json.NewDecoder(buf)
        // do the decode
        if err = dec.Decode(&res); err == nil {
            fmt.Println("Read:", res)
        }
    }
    if err != io.EOF {
        fmt.Println("Unexpected error:", err)
    }
}

func dropEnd(buf *bufio.Reader) {
    var check = make([]byte, 4)
    // If the next 4 bytes (either "\nend" or "end\n") contain "end", drop read them off the buffer
    if check, _ = buf.Peek(4); bytes.Contains(check, []byte("end")) {
        buf.Read(check)
    }
}

你可以在这里玩这个代码：http://play.golang.org/p/7NER_fTzXI

Answer 3

您可以将os.Stdin 包装在bufio.Reader 中，来自bufio 包。然后在解码之前使用buf.Peek(num) 向前看。

您还可以使用自定义 Scanner 来分隔 JSON 块。

与静态缓冲区相比，使用 bufio 的好处是它可以在流上工作。

Answer 4

如果您可以将 JSON 对象限制为单行，则只需逐行中断输入并忽略未解组的内容。这是一小段代码。

package main

import (
    "encoding/json"
    "fmt"
    "strings"
)

var input = `{"command": "ack", "id": "1231231"}
end
{"command": "fail", "id": "1231231"}
end
`

type Obj struct {
    Command string
    Msg     string
    Id      string
}

func DoSomethingCool(o Obj) {
    // Do something cool here
    fmt.Println(o)
}

func main() {
    inputs := strings.Split(input, "\n")
    for _, v := range inputs {
        var obj Obj
        if err := json.Unmarshal([]byte(v), &obj); err == nil {
            DoSomethingCool(obj) // Got a valid JSON
        }
    }
}