如何避免两个Azure Service Bus消费者同时更新同一Redis值

方案对比与选型 #

直接对比两种方案的适配性：

• SET NX+PX 分布式锁：完全匹配跨Pod实例的互斥需求，Redis 7.2.3原生支持，Lettuce有成熟API可实现，通过设置过期时间能避免死锁，容错性更高，是当前场景的最优选择。
• WATCH+EXEC事务：基于乐观锁机制，当WATCH的键被其他客户端修改时事务会直接失败，需要业务层手动重试，高并发下重试频率会飙升，且跨Pod网络延迟易导致误判，不适合你的ASB消费者并发更新场景。

另外还可以考虑Redis 7.0+支持的SET命令扩展参数，或是Redisson框架（封装了更完善的分布式锁实现，含自动续期等特性）。

代码实现示例（Spring Data Redis Lettuce） #

自定义分布式锁工具类 #

import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;
import java.util.UUID;

@Component
public class RedisDistributedLock {
    private final StringRedisTemplate stringRedisTemplate;
    // 根据业务更新耗时设置锁过期时间，防止死锁
    private static final long LOCK_EXPIRE_TIME = 5;
    private static final TimeUnit LOCK_TIME_UNIT = TimeUnit.SECONDS;

    public RedisDistributedLock(StringRedisTemplate stringRedisTemplate) {
        this.stringRedisTemplate = stringRedisTemplate;
    }

    // 获取锁，返回唯一标识用于后续释放锁（避免误删其他客户端的锁）
    public String tryLock(String lockKey) {
        String lockValue = UUID.randomUUID().toString();
        Boolean success = stringRedisTemplate.opsForValue()
                .setIfAbsent(lockKey, lockValue, LOCK_EXPIRE_TIME, LOCK_TIME_UNIT);
        return Boolean.TRUE.equals(success) ? lockValue : null;
    }

    // 原子性释放锁，仅持有锁的客户端能执行删除操作
    public boolean releaseLock(String lockKey, String lockValue) {
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        Long result = stringRedisTemplate.execute(
                (RedisCallback<Long>) connection -> connection.eval(script.getBytes(),
                        ReturnType.INTEGER, 1, lockKey.getBytes(), lockValue.getBytes())
        );
        return result != null && result > 0;
    }

    // 带重试的锁获取逻辑，适配ASB消费者等待场景
    public String acquireLockWithRetry(String lockKey, int retryTimes, long retryIntervalMs) throws InterruptedException {
        String lockValue;
        int attempt = 0;
        do {
            lockValue = tryLock(lockKey);
            if (lockValue != null) {
                return lockValue;
            }
            attempt++;
            Thread.sleep(retryIntervalMs);
        } while (attempt < retryTimes);
        return null;
    }
}

ASB消费者中集成锁逻辑 #

import com.azure.spring.cloud.service.servicebus.consumer.ServiceBusQueueAsyncClient;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

@Component
public class ASBMessageConsumer {
    private final RedisDistributedLock redisLock;
    private final StringRedisTemplate stringRedisTemplate;
    // Redis目标键（与ASB消息关联的待更新键）
    private static final String REDIS_TARGET_KEY = "asb:shared:data";
    // 对应目标键的锁键
    private static final String LOCK_KEY = "lock:" + REDIS_TARGET_KEY;

    public ASBMessageConsumer(RedisDistributedLock redisLock, StringRedisTemplate stringRedisTemplate) {
        this.redisLock = redisLock;
        this.stringRedisTemplate = stringRedisTemplate;
    }

    public void registerConsumer(ServiceBusQueueAsyncClient client) {
        client.receiveMessages()
                .flatMap(message -> {
                    String lockValue;
                    try {
                        // 尝试获取锁，重试3次，间隔500ms
                        lockValue = redisLock.acquireLockWithRetry(LOCK_KEY, 3, 500);
                        if (lockValue == null) {
                            // 获取锁失败，可选择放弃处理或把消息重新入队
                            return client.abandon(message).then(Mono.empty());
                        }
                        // 从ASB消息中提取数据，更新Redis目标键
                        String newData = extractDataFromMessage(message);
                        stringRedisTemplate.opsForValue().set(REDIS_TARGET_KEY, newData);
                        // 确认消息处理完成，最终释放锁
                        return client.complete(message)
                                .doFinally(signalType -> redisLock.releaseLock(LOCK_KEY, lockValue));
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        return client.abandon(message).then(Mono.error(e));
                    }
                })
                .subscribe();
    }

    private String extractDataFromMessage(com.azure.messaging.servicebus.ServiceBusReceivedMessage message) {
        // 自定义消息解析逻辑，根据实际消息格式调整
        return new String(message.getBody());
    }
}

其他可选方案 #

• Redisson分布式锁：Redisson封装了可重入锁、公平锁等高级特性，支持自动锁续期（避免业务更新超时导致锁提前释放），Spring Boot中可直接集成，减少自定义锁的代码复杂度。
• Redis 7.2+的SET扩展：使用SET lockKey lockValue NX PX 5000 GET命令，获取锁的同时返回之前的锁值，可用于更复杂的锁校验场景，但自定义实现仍需保证原子性。

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