func (engine *Engine) indexerAddDocumentWorker(shard int) {for {request := <-engine.indexerAddDocumentChannels[shard] //关键addInvertedIndex := engine.indexers[shard].AddDocument(request.document, request.dealDocInfoChan) // 向反向索引表(数组)中加入一个文档// saveif engine.initOptions.UsePersistentStorage {for k, v := range addInvertedIndex {engine.persistentStorageIndexDocumentChannels[shard] <- persistentStorageIndexDocumentRequest{typ:            "index",keyword:        k,keywordIndices: v,}}}atomic.AddUint64(&engine.numTokenIndexAdded,uint64(len(request.document.Keywords)))atomic.AddUint64(&engine.numDocumentsIndexed, 1)}
} 

package engineimport ("bytes""encoding/binary""encoding/gob""github.com/huichen/wukong/core""github.com/huichen/wukong/types""sync""sync/atomic"
)type persistentStorageIndexDocumentRequest struct {typ string //"info"or"index"// typ=="info"时，以下两个字段有效
    docId   uint64docInfo *types.DocInfo// typ=="index"时，以下两个字段有效keyword        stringkeywordIndices *types.KeywordIndices
}func (engine *Engine) persistentStorageIndexDocumentWorker(shard int) {for {request := <-engine.persistentStorageIndexDocumentChannels[shard]switch request.typ {case "info":// 得到keyb := make([]byte, 10)length := binary.PutUvarint(b, request.docId)// 得到valuevar buf bytes.Bufferenc := gob.NewEncoder(&buf)err := enc.Encode(request.docInfo)if err != nil {atomic.AddUint64(&engine.numDocumentsStored, 1)return}// 将key-value写入数据库engine.dbs[shard][getDB(request.typ)].Set(b[0:length], buf.Bytes())atomic.AddUint64(&engine.numDocumentsStored, 1)case "index":// 得到keyb := []byte(request.keyword)// 得到valuevar buf bytes.Bufferenc := gob.NewEncoder(&buf)err := enc.Encode(request.keywordIndices)if err != nil {return}// 将key-value写入数据库
            engine.dbs[shard][getDB(request.typ)].Set(b, buf.Bytes())}}
}func (engine *Engine) persistentStorageRemoveDocumentWorker(docId uint64, shard int) {// 得到keyb := make([]byte, 10)length := binary.PutUvarint(b, docId)// 从数据库删除该keyengine.dbs[shard][getDB("info")].Delete(b[0:length])
}func (engine *Engine) persistentStorageInitWorker(shard int) {var finish sync.WaitGroupfinish.Add(2)// 恢复docInfo
    go func() {defer finish.Add(-1)engine.dbs[shard][getDB("info")].ForEach(func(k, v []byte) error {key, value := k, v// 得到docIDdocId, _ := binary.Uvarint(key)// 得到databuf := bytes.NewReader(value)dec := gob.NewDecoder(buf)var data types.DocInfoerr := dec.Decode(&data)if err == nil {// 添加索引core.AddDocInfo(shard, docId, &data)}return nil})}()// 恢复invertedIndex
    go func() {defer finish.Add(-1)engine.dbs[shard][getDB("index")].ForEach(func(k, v []byte) error {key, value := k, v// 得到keywordkeyword := string(key)// 得到databuf := bytes.NewReader(value)dec := gob.NewDecoder(buf)var data types.KeywordIndiceserr := dec.Decode(&data)if err == nil {// 添加索引core.AddKeywordIndices(shard, keyword, &data)}return nil})}()finish.Wait()engine.persistentStorageInitChannel <- true
}