foo

a.go

package main

import (
  "bytes"
  "crypto/sha256"
  "encoding/json"
  "fmt"
  "io"
  "io/ioutil"
  "sync"
)

func main() {

  // 存储后端，可以是任意 key/value 式存储，例如对象存储服务
  store := new(sync.Map)
  set := func(key string, value []byte) {
    if _, ok := store.Load(key); ok {
      return
    }
    store.Store(key, value)
  }
  get := func(key string) []byte {
    v, ok := store.Load(key)
    if !ok {
      panic("key not found")
    }
    return v.([]byte)
  }

  // []byte -> key 函数
  getKey := func(bs []byte) string {
    h := sha256.New()
    h.Write(bs)
    return fmt.Sprintf("%x", h.Sum(nil))
  }

  // 输入，可以是任意 io.Reader
  input := bytes.NewReader([]byte("foobarbaz"))

  // 元数据
  type Meta struct {
    Chunks []string // 各段的 key
  }
  var meta Meta

  // 最大 4 个处理线程
  sem := make(chan struct{}, 4)
  cond := sync.NewCond(new(sync.Mutex))
  next := 0
  ready := 0
  for {
    // 分段读取
    r := &io.LimitedReader{
      R: input,
      N: 2,
    }
    chunk, err := ioutil.ReadAll(r)

    if len(chunk) > 0 {
      sem <- struct{}{}
      i := next
      next++
      go func() {
        defer func() {
          <-sem
        }()

        cond.L.Lock()
        for i != ready {
          cond.Wait()
        }

        // 保存
        key := getKey(chunk)
        set(key, chunk)
        meta.Chunks = append(meta.Chunks, key)

        ready++
        cond.Broadcast()
        cond.L.Unlock()

      }()
    }

    if err == io.EOF {
      break
    } else if err != nil {
      panic(err)
    }
    if len(chunk) == 0 {
      break
    }
  }
  for i := 0; i < cap(sem); i++ {
    sem <- struct{}{}
  }

  // 保存元数据
  buf := new(bytes.Buffer)
  if err := json.NewEncoder(buf).Encode(meta); err != nil {
    panic(err)
  }
  encoded := buf.Bytes()
  metaKey := "meta-" + getKey(encoded)
  set(metaKey, encoded)

  // 用 metaKey 就能获得数据
  {
    var meta Meta
    if err := json.Unmarshal(get(metaKey), &meta); err != nil {
      panic(err)
    }
    buf := new(bytes.Buffer)
    for _, chunkKey := range meta.Chunks {
      buf.Write(get(chunkKey))
    }
    fmt.Printf("%s\n", buf.Bytes())
  }

}