您现在的位置是：网站首页> 编程资料编程资料

关于使用Redisson订阅数问题_Redis_

2023-05-27 573人已围观

简介 关于使用Redisson订阅数问题_Redis_

一、前提

最近在使用分布式锁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 listener = createListener(channelName, value);             // 订阅监听             service.subscribe(LongCodec.INSTANCE, channelName, semaphore, listener);         }     };     // 最终会执行listener.run方法     semaphore.acquire(listener);     listenerHolder.set(listener);     return newPromise; }
AsyncSemaphore#acquire()方法
public void acquire(Runnable listener) {     acquire(listener, 1); } public void acquire(Runnable listener, int permits) {     boolean run = false;     synchronized (this) {         // counter初始化值为1         if (counter < permits) {             // 如果不是第一次执行，则将listener加入到listeners集合中             listeners.add(new Entry(listener, permits));             return;         } else {             counter -= permits;             run = true;         }     }     // 第一次执行acquire， 才会执行listener.run()方法     if (run) {         listener.run();     } }
梳理上述逻辑：
1、从PublishSubscribeService获取对应的信号量， 相同的channelName获取的是同一个信号量
2、如果是第一次请求，则会立马执行listener.run()方法， 否则需要等上个线程获取到该信号量执行完方能执行；
3、如果已经存在RedissonLockEntry， 则利用已经订阅就行
4、如果不存在RedissonLockEntry， 则会创建新的RedissonLockEntry，然后进行。
从上面代码看，主要逻辑是交给了PublishSubscribeService#subscribe方法
4、PublishSubscribeService#subscribe逻辑如下：
private final ConcurrentMap name2PubSubConnection = new ConcurrentHashMap<>(); private final Queue freePubSubConnections = new ConcurrentLinkedQueue<>(); public RFuture subscribe(Codec codec, String channelName, AsyncSemaphore semaphore, RedisPubSubListener... listeners) {     RPromise promise = new RedissonPromise();     // 主要逻辑入口， 这里要主要channelName每次都是新对象, 但内部覆写hashCode+equals。     subscribe(codec, new ChannelName(channelName), promise, PubSubType.SUBSCRIBE, semaphore, listeners);     return promise; } private void subscribe(Codec codec, ChannelName channelName,  RPromise promise, PubSubType type, AsyncSemaphore lock, RedisPubSubListener... listeners) {     PubSubConnectionEntry connEntry = name2PubSubConnection.get(channelName);     if (connEntry != null) {         // 从已有Connection中取，如果存在直接把listeners加入到PubSubConnectionEntry中         addListeners(channelName, promise, type, lock, connEntry, listeners);         return;     }     // 没有时，才是最重要的逻辑     freePubSubLock.acquire(new Runnable() {         @Override         public void run() {             if (promise.isDone()) {                 lock.release();                 freePubSubLock.release();                 return;             }             // 从队列中取头部元素             PubSubConnectionEntry freeEntry = freePubSubConnections.peek();             if (freeEntry == null) {                 // 第一次肯定是没有的需要建立                 connect(codec, channelName, promise, type, lock, listeners);                 return;             }             // 如果存在则尝试获取，如果remainFreeAmount小于0则抛出异常终止了。             int remainFreeAmount = freeEntry.tryAcquire();             if (remainFreeAmount == -1) {                 throw new IllegalStateException();             }             PubSubConnectionEntry oldEntry = name2PubSubConnection.putIfAbsent(channelName, freeEntry);             if (oldEntry != null) {                 freeEntry.release();                 freePubSubLock.release();                 addListeners(channelName, promise, type, lock, oldEntry, listeners);                 return;             }             // 如果remainFreeAmount=0， 则从队列中移除             if (remainFreeAmount == 0) {                 freePubSubConnections.poll();             }             freePubSubLock.release();             // 增加监听             RFuture subscribeFuture = addListeners(channelName, promise, type, lock, freeEntry, listeners);             ChannelFuture future;             if (PubSubType.PSUBSCRIBE == type) {                 future = 
                
                

                

                                            
上一篇：Redis 使用 List 实现消息队列的优缺点_Redis_

                                                                
下一篇：redis的list数据类型相关命令介绍及使用_Redis_

                                    

            

        
        

            
相关内容

            
        

    
    

        

            
点击排行

            
        

        

            
本栏推荐

            
        


        

            
猜你喜欢