json - 并发解析 JSON - panic of runtime error(解码相关)

Question

我最近在玩 go，遇到运行时错误，我无法解释。这些是我的工作职能。

    type User struct {
        Browsers []string `json:"browsers"`
        Name     string   `json:"name"`
        Email    string   `json:"email"`
    }

    func asyncUserProcJson(wg *sync.WaitGroup, users *[]User, ch chan []byte) {
        for buf := range ch {
            var mu sync.Mutex

            var user User
            mu.Lock()
            err := json.Unmarshal(buf, &user)
            mu.Unlock()
            if err != nil {
                fmt.Println("json:", err)
                wg.Done()
                continue
            }

            *users = append(*users, user)
            wg.Done()

        }
    }

    func userProcJson(buf []byte) (User, error) {
        var user User
        err := json.Unmarshal(buf, &user)
        if err != nil {
            return User{}, err
        }
        return user, nil
    }

如果我执行一个常见的非并发方法，它会按预期工作。但是，如果尝试使用 channel 将字节传递给 goroutine... 它会失败。

type AsyncUserProc func(*sync.WaitGroup, *[]User, chan []byte)
type UserProc func(buf []byte) (User, error)

type SearchParams struct {
    out              io.Writer
    asyncUserProc    AsyncUserProc 
    userProc         UserProc 
}

func (sp SearchParams) AsyncSearch() []User {
    file, err := os.Open(filePath)
    if err != nil {
        log.Fatalln(err)
    }

    var Users = make([]User, 0, 1024)
    var ch = make(chan []byte)
    var wg sync.WaitGroup

    go sp.asyncUserProcess(&wg, &Users, ch)

    scanner := bufio.NewScanner(file)
    for scanner.Scan() {
        wg.Add(1)
        ch <- scanner.Bytes()
    }

    if err := scanner.Err(); err != nil {
        fmt.Fprintln(os.Stderr, "reading standard input:", err)
    }
    close(ch)
    wg.Wait()

    return Users
}

func (sp SearchParams) Search() []User {
    file, err := os.Open(filePath)
    if err != nil {
        log.Fatalln(err)
    }

    // json processor
    var Users = make([]User, 0, 1024)

    scanner := bufio.NewScanner(file)
    for scanner.Scan() {

        u, err := sp.userProcess(scanner.Bytes())
        if err != nil {
            log.Panicln(err)
            continue
        }
        Users = append(Users, u)
    }

    if err := scanner.Err(); err != nil {
        fmt.Fprintln(os.Stderr, "reading standard input:", err)
    }

    return Users
}

下一个是工作流程:

1. filePath 包含一个 JSON block (每个在新行上)
2. 打开阅读。

3. 创建行扫描仪

   1. (异步搜索)

      • 将线路传递给 channel 。
      • 返回范围内行的值(阻塞操作)
      • 传递给 json.Unmarshal
      • 麻烦

   2. (搜索)

      • 将行直接传递给 userProc 函数
      • 享受结果

我遇到了很多(不同的)错误。

• 很多 json 解码错误。
• 索引超出范围
• JSON 解码器不同步 - 脚下的数据发生变化？

作为上次错误的描述:

// phasePanicMsg is used as a panic message when we end up with something that
// shouldn't happen. It can indicate a bug in the JSON decoder, or that
// something is editing the data slice while the decoder executes.

那么这里有一个问题:bytes slice是怎么修改的？ 我以为是阻塞操作。我在语言机制中缺少什么？

错误示例(每次运行不同)

json: invalid character 'i' looking for beginning of value
json: invalid character ':' after top-level value
json: invalid character 'r' looking for beginning of value
panic: runtime error: index out of range
----
json: invalid character '.' after top-level value
json: invalid character 'K' looking for beginning of value
panic: JSON decoder out of sync - data changing underfoot?

Answer 1

Package bufio

import "bufio"

func (*Scanner) Bytes

func (s *Scanner) Bytes() []byte

Bytes returns the most recent token generated by a call to Scan. The underlying array may point to data that will be overwritten by a subsequent call to Scan. It does no allocation.

---

底层数组可能指向将被后续调用 Scan 覆盖的数据。