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
}

работник. 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 -программа читать результаты. Операция запроса завершится (при условии, что есть 1000 строк или меньше), и стандартные операции go могут начать свою параллельную работу.

Если это исправит ситуацию, тогда подумайте о введении, скажем, ограничения SQL в ваших запросах (например, LIMIT 1000), чтобы убедиться, что вы больше не попадете в тупик (если реально более 1000 записей).

Создание запросов в стиле «нумерации страниц» для захвата следующих, скажем, 1000 строк, используя Маркеры RowID et c. чтобы обеспечить полный охват результатов - и все это, избегая блокировки любых ваших операций update.