Redis Subscribe timeout 報(bào)錯(cuò)的問題解決

更新時(shí)間：2025年08月15日 11:04:59 作者：一樂小哥

最近系統(tǒng)偶爾報(bào)出org.redisson.client.RedisTimeoutException: Subscribe timeout: (7500ms)的錯(cuò)誤,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

?? 介紹

Redisson版本 2.8.2

最近公司系統(tǒng)偶爾報(bào)出org.redisson.client.RedisTimeoutException: Subscribe timeout: (7500ms)的錯(cuò)誤,觀察堆棧信息看到報(bào)錯(cuò)是一段使用Redisson的redis鎖的地方,去除業(yè)務(wù)邏輯代碼基本如下

public void mockLock(String phoneNum) {
log.info("{} - prepare lock", threadName);
RLock lock = redissonClient.getLock("redis_cache_test" + phoneNum);
try {
    lock.lock();
    log.info("{} - get lock", threadName);
    //睡眠10s
    Thread.sleep(10000);
} catch (Exception e) {
    log.info("{} - exception", threadName,e);
} finally {
    log.info("{} - unlock lock", threadName);
    lock.unlock();
}

導(dǎo)致報(bào)錯(cuò)的代碼是lock.lock()的實(shí)現(xiàn)

@Override
public void syncSubscription(RFuture<?> future) {
    MasterSlaveServersConfig config = connectionManager.getConfig();
    try {
        int timeout = config.getTimeout() + config.getRetryInterval()*config.getRetryAttempts();
        if (!future.await(timeout)) {
            throw new RedisTimeoutException("Subscribe timeout: (" + timeout + "ms)");
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    future.syncUninterruptibly();
}

溯因

syncSubscription中的future是RedissonLock.subscribe(long threadId)方法

protected RFuture<RedissonLockEntry> subscribe(long threadId) {
    return PUBSUB.subscribe(getEntryName(), getChannelName(), commandExecutor.getConnectionManager());
}

這里可以看出大概是在PUBSUB中獲取一個(gè)訂閱,再往下看源碼

public RFuture<E> subscribe(final String entryName, final String channelName, final ConnectionManager connectionManager) {
    //監(jiān)聽持有
    final AtomicReference<Runnable> listenerHolder = new AtomicReference<Runnable>();
    //獲取鎖訂閱隊(duì)列
    final AsyncSemaphore semaphore = connectionManager.getSemaphore(channelName);
    //訂閱拒絕實(shí)現(xiàn)
    final RPromise<E> newPromise = new PromiseDelegator<E>(connectionManager.<E>newPromise()) {
        @Override
        public boolean cancel(boolean mayInterruptIfRunning) {
            return semaphore.remove(listenerHolder.get());
        }
    };

    Runnable listener = new Runnable() {

        @Override
        public void run() {
        //判斷是否已經(jīng)存在相同的entry
            E entry = entries.get(entryName);
            if (entry != null) {
                entry.aquire();
                semaphore.release();
                entry.getPromise().addListener(new TransferListener<E>(newPromise));
                return;
            }
            //沒有則新建
            E value = createEntry(newPromise);
            value.aquire();
            
            E oldValue = entries.putIfAbsent(entryName, value);
            if (oldValue != null) {
                oldValue.aquire();
                semaphore.release();
                oldValue.getPromise().addListener(new TransferListener<E>(newPromise));
                return;
            }
            //監(jiān)聽對應(yīng)的entry
            RedisPubSubListener<Object> listener = createListener(channelName, value);
            //訂閱事件
            connectionManager.subscribe(LongCodec.INSTANCE, channelName, listener, semaphore);
        }
    };
    //用semaphore管理監(jiān)聽隊(duì)列,因?yàn)榭赡艽嬖诙鄠€(gè)線程等待一個(gè)鎖
    semaphore.acquire(listener);
    //保證訂閱拒絕邏輯
    listenerHolder.set(listener);
    
    return newPromise;
}

這里可以看到這個(gè)方法其實(shí)只是定義了一個(gè)名叫listener的Runnable, semaphore.acquire(listener);則保證了同一個(gè)channel僅會有一個(gè)線程去監(jiān)聽,其他的繼續(xù)等待,而訂閱邏輯還在connectionManager.subscribe里面

private void subscribe(final Codec codec, final String channelName, final RedisPubSubListener<?> listener, 
        final RPromise<PubSubConnectionEntry> promise, final PubSubType type, final AsyncSemaphore lock) {
    final PubSubConnectionEntry connEntry = name2PubSubConnection.get(channelName);
    if (connEntry != null) {
        connEntry.addListener(channelName, listener);
        connEntry.getSubscribeFuture(channelName, type).addListener(new FutureListener<Void>() {
            @Override
            public void operationComplete(Future<Void> future) throws Exception {
                lock.release();
                promise.trySuccess(connEntry);
            }
        });
        return;
    }

    freePubSubLock.acquire(new Runnable() {

        @Override
        public void run() {
            if (promise.isDone()) {
                return;
            }
            //如果沒有獲取到公共的連接直接返回
            final PubSubConnectionEntry freeEntry = freePubSubConnections.peek();
            if (freeEntry == null) {
                connect(codec, channelName, listener, promise, type, lock);
                return;
            }
            //entry有個(gè)計(jì)數(shù)器subscriptionsPerConnection
            如果為-1報(bào)錯(cuò)因?yàn)橄旅嬗?的判斷
            int remainFreeAmount = freeEntry.tryAcquire();
            if (remainFreeAmount == -1) {
                throw new IllegalStateException();
            }
            
            final PubSubConnectionEntry oldEntry = name2PubSubConnection.putIfAbsent(channelName, freeEntry);
            if (oldEntry != null) {
                freeEntry.release();
                freePubSubLock.release();
                
                oldEntry.addListener(channelName, listener);
                oldEntry.getSubscribeFuture(channelName, type).addListener(new FutureListener<Void>() {
                    @Override
                    public void operationComplete(Future<Void> future) throws Exception {
                        lock.release();
                        promise.trySuccess(oldEntry);
                    }
                });
                return;
            }
            //subscriptionsPerConnection為0時(shí)從公共連接池中吐出
            if (remainFreeAmount == 0) {
                freePubSubConnections.poll();
            }
            freePubSubLock.release();
            
            freeEntry.addListener(channelName, listener);
            freeEntry.getSubscribeFuture(channelName, type).addListener(new FutureListener<Void>() {
                @Override
                public void operationComplete(Future<Void> future) throws Exception {
                    lock.release();
                    promise.trySuccess(freeEntry);
                }
            });
            
            if (PubSubType.PSUBSCRIBE == type) {
                freeEntry.psubscribe(codec, channelName);
            } else {
                freeEntry.subscribe(codec, channelName);
            }
        }
        
    });
}

這里在沒有連接的情況下會進(jìn)到connect(codec, channelName, listener, promise, type, lock);中去

private void connect(final Codec codec, final String channelName, final RedisPubSubListener<?> listener,
        final RPromise<PubSubConnectionEntry> promise, final PubSubType type, final AsyncSemaphore lock) {
    final int slot = calcSlot(channelName);
    //根據(jù)subscriptionConnectionPoolSize獲取下一個(gè)鏈接
    RFuture<RedisPubSubConnection> connFuture = nextPubSubConnection(slot);
    connFuture.addListener(new FutureListener<RedisPubSubConnection>() {

        @Override
        public void operationComplete(Future<RedisPubSubConnection> future) throws Exception {
            if (!future.isSuccess()) {
                freePubSubLock.release();
                lock.release();
                promise.tryFailure(future.cause());
                return;
            }

            RedisPubSubConnection conn = future.getNow();
            
            final PubSubConnectionEntry entry = new PubSubConnectionEntry(conn, config.getSubscriptionsPerConnection());
            entry.tryAcquire();
            
            final PubSubConnectionEntry oldEntry = name2PubSubConnection.putIfAbsent(channelName, entry);
            if (oldEntry != null) {
                releaseSubscribeConnection(slot, entry);
                
                freePubSubLock.release();
                
                oldEntry.addListener(channelName, listener);
                oldEntry.getSubscribeFuture(channelName, type).addListener(new FutureListener<Void>() {
                    @Override
                    public void operationComplete(Future<Void> future) throws Exception {
                        lock.release();
                        promise.trySuccess(oldEntry);
                    }
                });
                return;
            }
            
            freePubSubConnections.add(entry);
            freePubSubLock.release();
            
            entry.addListener(channelName, listener);
            entry.getSubscribeFuture(channelName, type).addListener(new FutureListener<Void>() {
                @Override
                public void operationComplete(Future<Void> future) throws Exception {
                    lock.release();
                    promise.trySuccess(entry);
                }
            });
            
            if (PubSubType.PSUBSCRIBE == type) {
                entry.psubscribe(codec, channelName);
            } else {
                entry.subscribe(codec, channelName);
            }
            
        }
    });
}

這里的RFuture<RedisPubSubConnection> connFuture = nextPubSubConnection(slot);最終會調(diào)用ClientConnectionsEntry#acquireSubscribeConnection方法的
freeSubscribeConnectionsCounter.acquire(runnable) 至此我們找到原因當(dāng)同時(shí)等待鎖訂閱消息達(dá)到subscriptionConnectionPoolSize*subscriptionsPerConnection個(gè)時(shí),再多一個(gè)訂閱消息,連接一直無法獲取導(dǎo)致MasterSlaveConnectionManager中的freePubSubLock沒有釋放。另外由于在超時(shí)場景下MasterSlaveConnectionManager向連接池獲取連接后是直接緩存下來，不把分發(fā)訂閱鏈接釋返回給連接池的，因此導(dǎo)致freeSubscribeConnectionsCounter一直等待，出現(xiàn)死鎖情況。

最終表現(xiàn)就是org.redisson.client.RedisTimeoutException: Subscribe timeout: (7500ms)

復(fù)現(xiàn)

Redis配置

public RedissonClient redissonClient(RedisConfig redisConfig) {

    Config config = new Config();
    config.useSingleServer()
        .setAddress(redisConfig.getHost() + ":" + redisConfig.getPort())
        .setPassword(redisConfig.getPassword())
        .setDatabase(redisConfig.getDatabase())
        .setConnectTimeout(redisConfig.getConnectionTimeout())
        .setTimeout(redisConfig.getTimeout())
        //把兩個(gè)配置項(xiàng)設(shè)置為1
        .setSubscriptionConnectionPoolSize(1)
        .setSubscriptionsPerConnection(1);
    return Redisson.create(config);
}

測試方法

void contextLoads() throws InterruptedException {
    Runnable runnable = () -> {
        redissonLock.tryRedissonLock();
    };
    new Thread(runnable, "線程1").start();
    new Thread(runnable, "線程12").start();
    new Thread(runnable, "線程23").start();
    new Thread(runnable, "線程21").start();
    
    Thread.sleep(200000);
}

結(jié)果

org.redisson.client.RedisTimeoutException: Subscribe timeout: (5500ms)
	at org.redisson.command.CommandAsyncService.syncSubscription(CommandAsyncService.java:126) ~[redisson-2.8.2.jar:na]
	at org.redisson.RedissonLock.lockInterruptibly(RedissonLock.java:121) ~[redisson-2.8.2.jar:na]
	at org.redisson.RedissonLock.lockInterruptibly(RedissonLock.java:108) ~[redisson-2.8.2.jar:na]
	at org.redisson.RedissonLock.lock(RedissonLock.java:90) ~[redisson-2.8.2.jar:na]
	at com.rick.redislock.lock.RedissonLock.registerPersonalMember(RedissonLock.java:30) ~[classes/:na]
	at com.rick.redislock.RedisLockApplicationTests.lambda$contextLoads$0(RedisLockApplicationTests.java:15) [test-classes/:na]
	at java.lang.Thread.run(Thread.java:748) ~[na:1.8.0_144]

符合預(yù)期

到此這篇關(guān)于Redis Subscribe timeout 報(bào)錯(cuò)的問題解決的文章就介紹到這了,更多相關(guān)Redis Subscribe timeout 報(bào)錯(cuò)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: