限制最大准备语句计数

问题

我编写了一个应用程序,将bigquery中的数据同步到mysql数据库中。我尝试每3小时批量插入大约10-20K行(每批最多10个项目)。出于某种原因,当它试图将这些行升迁到mysql中时,会收到以下错误:

不能创建超过max_-prepared_-stmt_-count个语句:

错误1461:无法创建超过max-prepared-stmt-count个语句 (当前值:2000)

我的“相关代码”

// ProcessProjectSkuCost receives the given sku cost entries and sends them in batches to upsertProjectSkuCosts()
func ProcessProjectSkuCost(done <-chan bigquery.SkuCost) {
    var skuCosts []bigquery.SkuCost
    var rowsAffected int64
    for skuCostRow := range done {
        skuCosts = append(skuCosts, skuCostRow)

        if len(skuCosts) == 10 {
            rowsAffected += upsertProjectSkuCosts(skuCosts)
            skuCosts = []bigquery.SkuCost{}
        }
    }
    if len(skuCosts) > 0 {
        rowsAffected += upsertProjectSkuCosts(skuCosts)
    }
    log.Infof("Completed upserting project sku costs. Affected rows: '%d'", rowsAffected)
}

// upsertProjectSkuCosts inserts or updates ProjectSkuCosts into SQL in batches
func upsertProjectSkuCosts(skuCosts []bigquery.SkuCost) int64 {
    // properties are table fields
    tableFields := []string{"project_name", "sku_id", "sku_description", "usage_start_time", "usage_end_time",
        "cost", "currency", "usage_amount", "usage_unit", "usage_amount_in_pricing_units", "usage_pricing_unit",
        "invoice_month"}
    tableFieldString := fmt.Sprintf("(%s)", strings.Join(tableFields, ","))

    // placeholderstring for all to be inserted values
    placeholderString := createPlaceholderString(tableFields)
    valuePlaceholderString := ""
    values := []interface{}{}
    for _, row := range skuCosts {
        valuePlaceholderString += fmt.Sprintf("(%s),", placeholderString)
        values = append(values, row.ProjectName, row.SkuID, row.SkuDescription, row.UsageStartTime,
            row.UsageEndTime, row.Cost, row.Currency, row.UsageAmount, row.UsageUnit,
            row.UsageAmountInPricingUnits, row.UsagePricingUnit, row.InvoiceMonth)
    }
    valuePlaceholderString = strings.TrimSuffix(valuePlaceholderString, ",")

    // put together SQL string
    sqlString := fmt.Sprintf(`INSERT INTO
        project_sku_cost %s VALUES %s ON DUPLICATE KEY UPDATE invoice_month=invoice_month`, tableFieldString, valuePlaceholderString)
    sqlString = strings.TrimSpace(sqlString)

    stmt, err := db.Prepare(sqlString)
    if err != nil {
        log.Warn("Error while preparing SQL statement to upsert project sku costs. ", err)
        return 0
    }

    // execute query
    res, err := stmt.Exec(values...)
    if err != nil {
        log.Warn("Error while executing statement to upsert project sku costs. ", err)
        return 0
    }

    rowsAffected, err := res.RowsAffected()
    if err != nil {
        log.Warn("Error while trying to access affected rows ", err)
        return 0
    }

    return rowsAffected
}

// createPlaceholderString creates a string which will be used for prepare statement (output looks like "(?,?,?)")
func createPlaceholderString(tableFields []string) string {
    placeHolderString := ""
    for range tableFields {
        placeHolderString += "?,"
    }
    placeHolderString = strings.TrimSuffix(placeHolderString, ",")

    return placeHolderString
}

我的问题:

为什么我要打 max_prepared_stmt_count 当我立即执行准备好的语句时(参见函数 upsertProjectSkuCosts )?

我只能想象,在准备和执行所有这些语句的同时,正是某种并发性创建了大量的准备好的语句。另一方面,我不明白为什么在 ProcessProjectSkuCost 是一个大小为20的缓冲通道。