我正在尝试使用 mgo 库进行批量更新插入。我正在阅读 documentation关于批量更新插入，因为这是我第一次使用 MongoDB，看起来我必须提供成对的文档才能更新。

在我的函数中，我正在执行查找所有查询，然后使用查询结果作为 bulk.Upsert() 操作的对的现有部分。我不确定这是否是正确的方法，但我必须一次对 ~65k 文档进行更新。

这里是类型结构，以及从 channel 读取以执行上述 MongoDB 操作的工作池函数。

// types from my project's `lib` package.
type Auctions struct {
    Auc          int `json:"auc" bson:"_id"`
    Item         int `json:"item" bson:"item"`
    Owner        string `json:"owner" bson:"owner"`
    OwnerRealm   string `json:"ownerRealm" bson:"ownerRealm"`
    Bid          int `json:"bid" bson:"bid"`
    Buyout       int `json:"buyout" bson:"buyout"`
    Quantity     int `json:"quantity" bson:"quantity"`
    TimeLeft     string `json:"timeLeft" bson:"timeLeft"`
    Rand         int `json:"rand" bson:"rand"`
    Seed         int `json:"seed" bson:"seed"`
    Context      int `json:"context" bson:"context"`
    BonusLists   []struct {
        BonusListID int `json:"bonusListId" bson:"bonusListId"`
    } `json:"bonusLists,omitempty" bson:"bonusLists,omitempty"`
    Modifiers    []struct {
        Type  int `json:"type" bson:"type"`
        Value int `json:"value" bson:"value"`
    } `json:"modifiers,omitempty" bson:"modifiers,omitempty"`
    PetSpeciesID int `json:"petSpeciesId,omitempty" bson:"petSpeciesId,omitempty"`
    PetBreedID   int `json:"petBreedId,omitempty" bson:"petBreedId,omitempty"`
    PetLevel     int `json:"petLevel,omitempty" bson:"petLevel,omitempty"`
    PetQualityID int `json:"petQualityId,omitempty" bson:"petQualityId,omitempty"`
}

type AuctionResponse struct {
    Realms   []struct {
        Name string `json:"name"`
        Slug string `json:"slug"`
    } `json:"realms"`
    Auctions []Auctions `json:"auctions"`
}

func (b *Blizzard) RealmAuctionGrabber(realms chan string, db *mgo.Database, wg *sync.WaitGroup) {
    defer wg.Done()
    for i := range realms {
        NewReq, err := http.NewRequest("GET", fmt.Sprintf("%s/auction/data/%s", b.Url, i), nil)
        if err != nil {
            fmt.Printf("Cannot create new request for realm %s: %s", i, err)
        }

        // Update the request with the default parameters and grab the files links.
        Request := PopulateDefaultParams(NewReq)
        log.Debugf("downloading %s auction locations.", i)
        Response, err := b.Client.Do(Request)
        if err != nil {
            fmt.Printf("Error request realm auction data: %s\n", err)
        }
        defer Response.Body.Close()

        Body, err := ioutil.ReadAll(Response.Body)
        if err != nil {
            fmt.Printf("Error parsing request body: %s\n", err)
        }

        var AuctionResp lib.AuctionAPI
        err = json.Unmarshal(Body, &AuctionResp)
        if err != nil {
            fmt.Printf("Error marshalling auction repsonse body: %s\n", err)
        }

        for _, x := range AuctionResp.Files {
            NewDataReq, err := http.NewRequest("GET", x.URL, nil)
            if err != nil {
                log.Error(err)
            }

            AuctionResponse, err := b.Client.Do(NewDataReq)
            if err != nil {
                fmt.Printf("Error request realm auction data: %s\n", err)
            }
            defer Response.Body.Close()

            AuctionBody, err := ioutil.ReadAll(AuctionResponse.Body)
            if err != nil {
                fmt.Printf("Error parsing request body: %s\n", err)
            }

            var AuctionData lib.AuctionResponse
            err = json.Unmarshal(AuctionBody, &AuctionData)

            // grab all the current records, then perform an Upsert!
            var existing []lib.Auctions
            col := db.C(i)
            err = col.Find(nil).All(&existing)
            if err != nil {
                log.Error(err)
            }

            log.Infof("performing bulk upsert for %s", i)

            auctionData, err := bson.Marshal(AuctionData.Auctions)
            if err != nil {
                log.Error("error marshalling bson: %s", err)
            }

            existingData, _ := bson.Marshal(existing)
            bulk := db.C(i).Bulk()
            bulk.Upsert(existingData, auctionData)
            _, err = bulk.Run()
            if err != nil {
                log.Error("error performing upsert! error: ", err)
            }
        }
    }
}

当我调用 bulk.Upsert(existingData,auctionData) 时，一切正常。但是，当我调用 bulk.Run() 时，我收到以下记录的错误消息:

{"level":"error","msg":"error performing upsert! error: wrong type for 'q' field, expected object, found q: BinData(0, 0500000000)","time":"2016-07-09T16:53:45-07:00"}

我假设这与我在 Auction 结构中进行 BSON 标记的方式有关，但我不确定，因为这是我第一次使用 MongoDB。现在数据库中没有集合，

错误消息是否与 BSON 标记有关，我该如何解决？

最佳答案

不知道这是否仍然是真实的。这是代码:

package main

import (
    "gopkg.in/mgo.v2"
    "log"
    "io/ioutil"
    "encoding/json"
    "gopkg.in/mgo.v2/bson"
)

type Auctions struct {
    Auc   int `json:"auc" bson:"_id"`
    Owner string `json:"owner" bson:"owner"`
}

type AuctionResponse struct {
    Auctions []Auctions `json:"auctions"`
}

func main() {
    session, err := mgo.Dial("mongodb://127.0.0.1:27017/aucs")

    if err != nil {
        panic(err)
    }
    defer session.Close()
    session.SetMode(mgo.Monotonic, true)

    db := session.DB("aucs")

    var AuctionData AuctionResponse
    AuctionBody, _ := ioutil.ReadFile("auctions.json")
    err = json.Unmarshal(AuctionBody, &AuctionData)

    log.Println("performing bulk upsert for %s", "realm")

    bulk := db.C("realm").Bulk()
    for _, auc := range AuctionData.Auctions {
        bulk.Upsert(bson.M{"_id": auc.Auc}, auc)
    }
    _, err = bulk.Run()
    if err != nil {
        log.Panic("error performing upsert! error: ", err)
    }
}

我做了一些更改以检查我在真实数据上是否正确，但我希望不难理解正在发生的事情。现在让我稍微解释一下 Mongo 的部分。

1. 您不需要获取现有 文档。在更新或插入现有文档之前请求所有现有文档会很奇怪，尤其是在大型数据库中。
2. mgo 的upsert 函数需要两个参数:
   • selector 文档(我更愿意将其视为 SQL 中的 WHERE 条件)
   • document 你实际拥有的，如果选择器查询失败，它将被插入到数据库中
3. 在批量upsert 的情况下，有无限数量的selector - document 对可用于推送一个请求，例如bulk.Upsert(sel1, doc1, sel2 , doc2, sel3, doc3) 但在您的情况下，没有必要使用此功能。

4. upsert 仅影响单个文档，因此即使您的 selector 适合多个文档，也只会首先更新。换句话说，upsert 更类似于 MySQL 的 INSERT ... ON DUPLICATE KEY UPDATE 而不是 UPDATE ... WHERE。在您的情况下(具有唯一的拍卖 ID)单个 upsert 看起来像 MySQL 查询:

   INSERT INTO realm (auc, owner) VALUES ('1', 'Hogger') ON DUPLICATE KEY UPDATE owner = 'Hogger';
   

5. 使用 bulk，我们可以在循环中进行多个更新插入，然后一次运行它们。

关于mongodb - Mgo 字段类型错误，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/38287439/