MySQL 查询有时会死锁

Question

我正在开发一个对 MySQL 进行查询的程序，然后为每一行更改该行的某些内容，然后更新该行。

问题是，有时在执行更新时我会遇到死锁，我不确定是因为在我更新时查询没有释放锁，还是其他原因。

我在做什么的例子：

const (
    selectQuery = `select user_id, original_transaction_id, max(payment_id) as max_payment_id from Payment_Receipt 
                        where auto_renew_status = 1 group by user_id, original_transaction_id having count(*) > 1`
    updateQuery = `update Payment_Receipt set auto_renew_status = 0, changed_by = "payment_receipt_condenser", 
changed_time = ? where user_id = ? and original_transaction_id = ? and payment_id != ? and auto_renew_status = 1`
)

mysql.go:

func New(db *sql.DB, driver string) (database.Database, error) {
    sqlDB := sqlx.NewDb(db, driver)

    if err := db.Ping(); err != nil {
        return nil, errors.Wrap(err, "connecting to database")
    }

    selectStmt, err := sqlDB.Preparex(selectQuery)
    if err != nil {
        return nil, errors.Wrap(err, "preparing select query")
    }

    updateStmt, err := sqlDB.Preparex(updateQuery)
    if err != nil {
        return nil, errors.Wrap(err, "preparing update query")
    }

    return &mysql{
        db:         sqlDB,
        selectStmt: selectStmt,
        updateStmt: updateStmt,
    }, nil
}

func (m *mysql) Query() (<- chan *database.Row, error) {
    rowsChan := make(chan *database.Row)

    rows, err := m.selectStmt.Queryx()
    if err != nil {
        return nil, errors.Wrap(err, "making query")
    }

    go func() {
        defer rows.Close()
        defer close(rowsChan)

        for rows.Next() {
            row := &database.Row{}

            if err := rows.StructScan(row); err != nil {
                log.WithError(err).WithField("user_id", row.UserID.Int32).Error("scanning row")
            }

            // change some of the data here
            // and put into channel for worker to consume

            rowsChan <- row
        }
    }()

    return rowsChan, nil
}

func (m *mysql) Update(row *database.Row) error {
    tx, err := m.db.Beginx()
    if err != nil {
        return errors.Wrap(err, "beginning transaction")
    }

    if _, err := tx.Stmtx(m.updateStmt).Exec(row.ChangedTime); err != nil {
        return errors.Wrap(err, "executing update")
    }

    if err := tx.Commit(); err != nil {
        return errors.Wrap(err, "committing transaction")
    }

    return nil
}

worker.go

func (w *worker) Run(wg *sync.WaitGroup) {
    rowsChan, err := w.db.Query()
    if err != nil {
        log.WithError(err).Fatal("failed making query")
    }

    for i := 0; i < w.config.Count(); i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for row := range rowsChan {
                if err := w.db.Update(row); err != nil {
                    log.WithError(err).WithField("user_id", row.UserID.Int32).Error("updating row")
                }
            }
        }()
    }
}

Answer 1

您可以从Query() 缓冲结果（行）通道：

func (m *mysql) Query() (<- chan *database.Row, error) {

    rowsChan := make(chan *database.Row, 1000) // <- band-aid fix

    // ...
}

这将确保行收集器函数可以写入多个结果，而无需等待您的工作程序 go-routine 读取结果。查询操作将完成（假设有 1000 行或更少），更新 go-routine 操作可以开始它们的并行工作。

如果这能解决问题，那么考虑对您的查询设置一个 SQL 限制（例如 LIMIT 1000），以确保您不会再次遇到死锁（如果 1000 多条记录确实有可能）。

制作“分页”样式查询以获取接下来的 1000 行，使用 RowID 标记等以确保完全覆盖结果 - 同时避免锁定您的任何 update 操作。