使用Redisson订阅数问题怎么解决

本文小编为大家详细介绍"使用Redisson订阅数问题怎么解决"，内容详细，步骤清晰，细节处理妥当，希望这篇"使用Redisson订阅数问题怎么解决"文章能帮助大家解决疑惑，下面跟着小编的思路慢慢深入，一起来学习新知识吧。

一、前提

最近在使用分布式锁redisson时遇到一个线上问题：发现是subscriptionsPerConnection or subscriptionConnectionPoolSize 的大小不够，需要提高配置才能解决。

二、源码分析

下面对其源码进行分析，才能找到到底是什么逻辑导致问题所在：

1、RedissonLock#lock() 方法

private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {        long threadId = Thread.currentThread().getId();        // 尝试获取，如果ttl == null，则表示获取锁成功        Long ttl = tryAcquire(leaseTime, unit, threadId);        // lock acquired        if (ttl == null) {            return;        }        // 订阅锁释放事件，并通过await方法阻塞等待锁释放，有效的解决了无效的锁申请浪费资源的问题        RFuture future = subscribe(threadId);        if (interruptibly) {            commandExecutor.syncSubscriptionInterrupted(future);        } else {            commandExecutor.syncSubscription(future);        }        // 后面代码忽略        try {            // 无限循环获取锁，直到获取锁成功            // ...        } finally {            // 取消订阅锁释放事件            unsubscribe(future, threadId);        }}

总结下主要逻辑：

• 获取当前线程的线程id；

• tryAquire尝试获取锁，并返回ttl

• 如果ttl为空，则结束流程；否则进入后续逻辑；

• this.subscribe(threadId)订阅当前线程，返回一个RFuture；

• 如果在指定时间没有监听到，则会产生如上异常。

• 订阅成功后， 通过while(true)循环，一直尝试获取锁

• fially代码块，会解除订阅

所以上述这情况问题应该出现在subscribe()方法中

2、详细看下subscribe()方法

protected RFuture subscribe(long threadId) {    // entryName 格式："id:name"；    // channelName 格式："redisson_lock__channel:name"；    return pubSub.subscribe(getEntryName(), getChannelName());}

RedissonLock#pubSub 是在RedissonLock构造函数中初始化的：

public RedissonLock(CommandAsyncExecutor commandExecutor, String name) {    // ....    this.pubSub = commandExecutor.getConnectionManager().getSubscribeService().getLockPubSub();}

而subscribeService在MasterSlaveConnectionManager的实现中又是通过如下方式构造的

public MasterSlaveConnectionManager(MasterSlaveServersConfig cfg, Config config, UUID id) {    this(config, id);    this.config = cfg;    // 初始化    initTimer(cfg);    initSingleEntry();}protected void initTimer(MasterSlaveServersConfig config) {    int[] timeouts = new int[]{config.getRetryInterval(), config.getTimeout()};    Arrays.sort(timeouts);    int minTimeout = timeouts[0];    if (minTimeout % 100 != 0) {        minTimeout = (minTimeout % 100) / 2;    } else if (minTimeout == 100) {        minTimeout = 50;    } else {        minTimeout = 100;    }    timer = new HashedWheelTimer(new DefaultThreadFactory("redisson-timer"), minTimeout, TimeUnit.MILLISECONDS, 1024, false);    connectionWatcher = new IdleConnectionWatcher(this, config);    // 初始化:其中this就是MasterSlaveConnectionManager实例，config则为MasterSlaveServersConfig实例：    subscribeService = new PublishSubscribeService(this, config);}

PublishSubscribeService构造函数

private final SemaphorePubSub semaphorePubSub = new SemaphorePubSub(this);public PublishSubscribeService(ConnectionManager connectionManager, MasterSlaveServersConfig config) {    super();    this.connectionManager = connectionManager;    this.config = config;    for (int i = 0; i < locks.length; i++) {        // 这里初始化了一组信号量，每个信号量的初始值为1        locks[i] = new AsyncSemaphore(1);    }}

3、回到subscribe()方法主要逻辑还是交给了 LockPubSub#subscribe()里面

private final ConcurrentMap entries = new ConcurrentHashMap<>();public RFuture subscribe(String entryName, String channelName) {      // 从PublishSubscribeService获取对应的信号量。 相同的channelName获取的是同一个信号量     // public AsyncSemaphore getSemaphore(ChannelName channelName) {    //    return locks[Math.abs(channelName.hashCode() % locks.length)];    // }    AsyncSemaphore semaphore = service.getSemaphore(new ChannelName(channelName));    AtomicReference listenerHolder = new AtomicReference();        RPromise newPromise = new RedissonPromise() {        @Override        public boolean cancel(boolean mayInterruptIfRunning) {            return semaphore.remove(listenerHolder.get());        }    };    Runnable listener = new Runnable() {        @Override        public void run() {            //  如果存在RedissonLockEntry， 则直接利用已有的监听            E entry = entries.get(entryName);            if (entry != null) {                entry.acquire();                semaphore.release();                entry.getPromise().onComplete(new TransferListener(newPromise));                return;            }            E value = createEntry(newPromise);            value.acquire();            E oldValue = entries.putIfAbsent(entryName, value);            if (oldValue != null) {                oldValue.acquire();                semaphore.release();                oldValue.getPromise().onComplete(new TransferListener(newPromise));                return;            }            // 创建监听，            RedisPubSubListener