多线程处理MySQL待处理记录时select for update事务锁失效导致重复获取问题咨询

Great question! Let's start by unpacking why you're hitting this unexpected behavior, then dive into cleaner, more robust solutions to eliminate the problem entirely.

Why This Happens #

Your current setup with SELECT ... FOR UPDATE and a transaction seems solid on paper, but the rare cases where Thread 2 grabs the same record (now marked INPROGRESS) likely stem from edge-case lock timing or subtle transaction isolation quirks. Even though Thread 2 waits for Thread 1's lock to release, in some scenarios, the database might re-run the query in a way that picks up the now-outdated record (though this shouldn't happen with proper current-read semantics). Either way, we can refactor the logic to avoid this uncertainty entirely.

Better Solutions #

1. Atomic Update + Fetch (Top Recommendation) #

The most reliable fix is to combine the "find eligible record" and "mark as INPROGRESS" steps into a single atomic SQL operation. This removes any window for concurrency issues because the database handles the lock and state change in one go.

For PostgreSQL (supports RETURNING):

Update your repository method to use an UPDATE that returns the modified record:

@Query(value = "UPDATE table SET status = 'INPROGRESS' WHERE id = (SELECT id FROM table WHERE status NOT IN ('COMPLETE', 'INPROGRESS') LIMIT 1) RETURNING *", nativeQuery=true)
Table lockAndFetchValidRecord();

For MySQL (no RETURNING support):

Use a nested subquery to update first, then fetch the locked record in the same transaction:

// First, atomically mark the record as INPROGRESS
@Modifying
@Query(value = "UPDATE table SET status = 'INPROGRESS' WHERE id = (SELECT id FROM (SELECT id FROM table WHERE status NOT IN ('COMPLETE', 'INPROGRESS') LIMIT 1) AS temp_subquery)", nativeQuery=true)
int markRecordAsInProgress();

// Then fetch the locked record
@Query(value = "SELECT * FROM table WHERE status = 'INPROGRESS' AND id = (SELECT id FROM (SELECT id FROM table WHERE status NOT IN ('COMPLETE', 'INPROGRESS') LIMIT 1) AS temp_subquery)", nativeQuery=true)
Table getLockedRecord();

In your service, wrap both calls in a single transaction to keep them atomic:

@Transactional(propagation = Propagation.REQUIRES_NEW)
public Table getOneValidRecord() {
    int updated = repository.markRecordAsInProgress();
    if (updated == 0) {
        return null; // No eligible records left
    }
    return repository.getLockedRecord();
}

This approach eliminates the "fetch then update" gap entirely—no more race conditions.

2. Hardening Your Existing Transaction Logic #

If you want to stick with your current pattern, you can add a safety check to validate the record's status right before updating it, just in case:

@Transactional(propagation = Propagation.REQUIRES_NEW)
public Table getOneValidRecord() {
    Table table = repository.getOneValidRecord();
    if (table == null) {
        return null;
    }
    // Double-check the status hasn't changed (edge-case safeguard)
    if (!"PENDING".equals(table.getStatus())) {
        // Recursively try again to find a valid record
        return getOneValidRecord();
    }
    table.setStatus("INPROGRESS");
    return repository.save(table);
}

This is a band-aid compared to the atomic approach, but it adds a safety net for rare edge cases.

3. Optimistic Locking (For Low-Concurrency Scenarios) #

If your system doesn't handle extremely high traffic, optimistic locking with a version column is another option. It avoids row locks entirely by checking if the record has changed since you fetched it:

First, add a version column to your table:

ALTER TABLE table ADD COLUMN version INT DEFAULT 0;

Then update your repository and service:

// Repository methods
@Query(value = "SELECT * FROM table WHERE status NOT IN ('COMPLETE', 'INPROGRESS') LIMIT 1", nativeQuery=true)
Table getOneValidRecord();

@Modifying
@Query(value = "UPDATE table SET status = 'INPROGRESS', version = version + 1 WHERE id = :id AND version = :version", nativeQuery=true)
int markAsInProgressWithVersionCheck(@Param("id") Long id, @Param("version") int version);

// Service method
@Transactional(propagation = Propagation.REQUIRES_NEW)
public Table getOneValidRecord() {
    Table table = repository.getOneValidRecord();
    if (table == null) {
        return null;
    }
    // Try to update only if the version hasn't changed
    int updatedRows = repository.markAsInProgressWithVersionCheck(table.getId(), table.getVersion());
    if (updatedRows == 0) {
        // Another thread grabbed this record—try again
        return getOneValidRecord();
    }
    table.setStatus("INPROGRESS");
    table.setVersion(table.getVersion() + 1);
    return table;
}

This works well for low-to-medium concurrency, but you might see more retries under heavy load.

Final Recommendation #

Go with the atomic update + fetch method—it's the most performant and reliable way to eliminate concurrency issues. It cuts out the middleman of separate fetch/update steps and leverages the database's built-in atomicity guarantees.

内容的提问来源于stack exchange，提问作者Hindol Dey