快捷導(dǎo)航

Redisson分布式信號量RSemaphore的使用超詳細講解

更新時間：2023年02月11日 11:19:18 作者：每天都要進步一點點

這篇文章主要介紹了Redisson分布式信號量RSemaphore的使用，基于Redis的Redisson的分布式信號量RSemaphore采用了與java.util.concurrent.Semaphore相似的接口和用法

一、RSemaphore的使用

@Test
public void testRSemaphore() {
    Config config = new Config();
    config.useSingleServer().setAddress("redis://127.0.0.1:6379");
    RedissonClient redissonClient = Redisson.create(config);
    RSemaphore rSemaphore = redissonClient.getSemaphore("semaphore");
    // 設(shè)置5個許可，模擬五個停車位
    rSemaphore.trySetPermits(5);
    // 創(chuàng)建10個線程，模擬10輛車過來停車
    for (int i = 1; i <= 10; i++) {
        new Thread(() -> {
            try {
                rSemaphore.acquire();
                System.out.println(Thread.currentThread().getName() + "進入停車場...");
                TimeUnit.MILLISECONDS.sleep(new Random().nextInt(100));
                System.out.println(Thread.currentThread().getName() + "離開停車場...");
                rSemaphore.release();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }, "A" + i).start();
    }
    try {
        TimeUnit.MINUTES.sleep(1);
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
}

二、RSemaphore設(shè)置許可數(shù)量

初始化RSemaphore，需要調(diào)用trySetPermits()設(shè)置許可數(shù)量：

/**
 * 嘗試設(shè)置許可數(shù)量，設(shè)置成功，返回true，否則返回false
 */
boolean trySetPermits(int permits);

trySetPermits()內(nèi)部調(diào)用了trySetPermitsAsync()：

// 異步設(shè)置許可
@Override
public RFuture<Boolean> trySetPermitsAsync(int permits) {
    RFuture<Boolean> future = commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
            // 判斷分布式信號量的key是否存在，如果不存在，才設(shè)置
            "local value = redis.call('get', KEYS[1]); " +
                    "if (value == false) then "
                    // set "semaphore" permits
                    // 使用String數(shù)據(jù)結(jié)構(gòu)設(shè)置信號量的許可數(shù)
                    + "redis.call('set', KEYS[1], ARGV[1]); "
                    // 發(fā)布一條消息到redisson_sc:{semaphore}通道
                    + "redis.call('publish', KEYS[2], ARGV[1]); "
                    // 設(shè)置成功，返回1
                    + "return 1;"
                    + "end;"
                    // 否則返回0
                    + "return 0;",
            Arrays.asList(getRawName(), getChannelName()), permits);
    if (log.isDebugEnabled()) {
        future.thenAccept(r -> {
            if (r) {
                log.debug("permits set, permits: {}, name: {}", permits, getName());
            } else {
                log.debug("unable to set permits, permits: {}, name: {}", permits, getName());
            }
        });
    }
    return future;
}

可以看到，設(shè)置許可數(shù)量底層使用LUA腳本，實際上就是使用redis的String數(shù)據(jù)結(jié)構(gòu)，保存了我們指定的許可數(shù)量。如下圖：

參數(shù)說明:

KEYS[1]: 我們指定的分布式信號量key，例如redissonClient.getSemaphore("semaphore")中的"semaphore")
KEYS[2]: 釋放鎖的channel名稱，redisson_sc:{分布式信號量key}，在本例中，就是redisson_sc:{semaphore}
ARGV[1]: 設(shè)置的許可數(shù)量

總結(jié)設(shè)置許可執(zhí)行流程為：

get semaphore，獲取到semaphore信號量的當(dāng)前的值
第一次數(shù)據(jù)為0，然后使用set semaphore 3，將這個信號量同時能夠允許獲取鎖的客戶端的數(shù)量設(shè)置為3。（注意到，如果之前設(shè)置過了信號量，將無法再次設(shè)置，直接返回0。想要更改信號量總數(shù)可以使用addPermits方法）
然后redis發(fā)布一些消息，返回1

三、RSemaphore的加鎖流程

許可數(shù)量設(shè)置好之后，我們就可以調(diào)用acquire()方法獲取了，如果未傳入許可數(shù)量，默認獲取一個許可。

public void acquire() throws InterruptedException {
    acquire(1);
}
public void acquire(int permits) throws InterruptedException {
    // 嘗試獲取鎖成功，直接返回
    if (tryAcquire(permits)) {
        return;
    }
    // 對于沒有獲取鎖的那些線程，訂閱redisson_sc:{分布式信號量key}通道的消息
    CompletableFuture<RedissonLockEntry> future = subscribe();
    semaphorePubSub.timeout(future);
    RedissonLockEntry entry = commandExecutor.getInterrupted(future);
    try {
        // 不斷循環(huán)嘗試獲取許可
        while (true) {
            if (tryAcquire(permits)) {
                return;
            }
            entry.getLatch().acquire();
        }
    } finally {
        // 取消訂閱
        unsubscribe(entry);
    }
//        get(acquireAsync(permits));
}

可以看到，獲取許可的核心邏輯在tryAcquire()方法中，如果tryAcquire()返回true說明獲取許可成功，直接返回；如果返回false，說明當(dāng)前沒有許可可以使用，則對于沒有獲取鎖的那些線程，訂閱redisson_sc:{分布式信號量key}通道的消息，并通過死循環(huán)不斷嘗試獲取鎖。

我們看一下tryAcquire()方法的邏輯，內(nèi)部調(diào)用了tryAcquireAsync()方法：

// 異步獲取許可
@Override
public RFuture<Boolean> tryAcquireAsync(int permits) {
    if (permits < 0) {
        throw new IllegalArgumentException("Permits amount can't be negative");
    }
    if (permits == 0) {
        return new CompletableFutureWrapper<>(true);
    }
    return commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
              // 獲取當(dāng)前剩余的許可數(shù)量
              "local value = redis.call('get', KEYS[1]); " +
              // 許可不為空，并且許可數(shù)量 大于等于 當(dāng)前線程申請的許可數(shù)量        
              "if (value ~= false and tonumber(value) >= tonumber(ARGV[1])) then " +
                  // 通過decrby減少剩余可用許可    
                  "local val = redis.call('decrby', KEYS[1], ARGV[1]); " +
                  // 返回1    
                  "return 1; " +
              "end; " +
              // 其它情況，返回0        
              "return 0;",
              Collections.<Object>singletonList(getRawName()), permits);
}

從源碼可以看到，獲取許可就是操作redis中的數(shù)據(jù)，首先獲取到redis中剩余的許可數(shù)量，只有當(dāng)剩余的許可數(shù)量大于線程申請的許可數(shù)量時，才獲取成功，返回1；否則獲取失敗，返回0；

總結(jié)加鎖執(zhí)行流程為：

get semaphore，獲取到一個當(dāng)前的值，比如說是3，3 > 1
decrby semaphore 1，將信號量允許獲取鎖的客戶端的數(shù)量遞減1，變成2
decrby semaphore 1
decrby semaphore 1
執(zhí)行3次加鎖后，semaphore值為0
此時如果再來進行加鎖則直接返回0，然后進入死循環(huán)去獲取鎖

四、RSemaphore的解鎖流程

通過前面對RSemaphore獲取鎖的分析，我們很容易能猜到，釋放鎖，無非就是歸還許可數(shù)量到redis中。我們查看具體的源碼：

public RFuture<Void> releaseAsync(int permits) {
    if (permits < 0) {
        throw new IllegalArgumentException("Permits amount can't be negative");
    }
    if (permits == 0) {
        return new CompletableFutureWrapper<>((Void) null);
    }
    RFuture<Void> future = commandExecutor.evalWriteAsync(getRawName(), StringCodec.INSTANCE, RedisCommands.EVAL_VOID,
            // 通過incrby增加許可數(shù)量
            "local value = redis.call('incrby', KEYS[1], ARGV[1]); " +
                    // 發(fā)布一條消息到redisson_sc:{semaphore}中
                    "redis.call('publish', KEYS[2], value); ",
            Arrays.asList(getRawName(), getChannelName()), permits);
    if (log.isDebugEnabled()) {
        future.thenAccept(o -> {
            log.debug("released, permits: {}, name: {}", permits, getName());
        });
    }
    return future;
}

到此這篇關(guān)于Redisson分布式信號量RSemaphore的使用超詳細講解的文章就介紹到這了,更多相關(guān)Redisson RSemaphore內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: