只用了半个Redisson的Semaphore实现并发控制

淡淡的烟草味﹌ 2024-03-27 02:20 59阅读 0赞

做过企业微信开发的同学应该知道，企业微信有一个很讨厌的报错--接口并发超过限制（45033）。报错的原因呢就是因为有多个线程在同时调用企业微信的接口，为了不让接口调用一直报错，我的服务就也要有一个接口并发控制体系。

首先想到的就是用线程池来实现

![9a510b5d8d4f884cbde23c34ce0efbf5.png][]

刚开始看起来似乎没什么问题，但是企业微信的接口之间并发限制是互相隔离的，那就意味着，我需要创建几十个线程池。。。你以为这就完了？其实不同企业用户调用的时候又是互相隔离的，所以，我需要为每个企业。。。。

![e320ddb299961039213992828ecb9933.png][]

线程池肯定是用不了了，我猛然想起前天晚上做梦，大师似乎跟我讲了信号量可以用来控制并发，考虑到线上环境有多个节点，同时Redisson也提供了Semaphore，打算用一下试试。想想距离上次使用信号量已经过去了∞天（毕竟我从来都没有用过），我直接打开IDEA，先用java的试试水，public static void main一气呵成

public static void main(String[] args) {
        //创建一个信号量
        Semaphore semaphore = new Semaphore(1);
        //如果没获取到就抛出异常
        if (!semaphore.tryAcquire()) {
            throw new RuntimeException();
        }
        //try不能括到上面，获取不到的时候是不需要释放的
        try{
            //做一些业务
            System.out.println("do something");
        }finally {
            //释放掉
            semaphore.release();
        }
    }
    复制代码

又检查了一下逻辑，看起来似乎没什么问题，然后我突发奇想，如果我直接release()会怎么样

Semaphore semaphore = new Semaphore(1);
    semaphore.release();
    System.out.println(semaphore.availablePermits());
    复制代码

![8bcc1f19f0bcf60d930869ff50b60f70.png][]

。。。0.0

此时我的心态出现了亿点变化，我意识到，其实初始化的`new Semaphore(1)`只是设置了一个初始值，并不是一个最大值，直接调用`release()`，可获取信号量的值会直接+1，这时我赶紧用Redisson的api `redissonClient.getSemaphore("test");` (其他代码和上面的一样)试了一下，万一呢

果然效果和Java的是一模一样。

不过，问题似乎也不是很大？如果我保证业务在拿到信号后一定会在`finally{..}`中释放掉，并且`semaphore.trySetPermits(1);`这个方法只初始化一次，并且一定是单线程，那其实就没有问题。

![7debbc3ce1ea31a472946af1e31cc015.png][]

企业微信的这个并发限制，没有明文规定，那上线之后，大概是要在违法的边缘多试探试探的，所以就涉及到更新信号量的问题了

![25e526102bb7a32de542d897920e43ac.png][]

可以看到，如果在多线程的情况下，贸然更新信号量，由于模型没有最大值的概念，更新后的信号量总量比起期望值是只多不少的。

问题主要还是出在`release()`这个方法没有上限概念上，那我来弄一个有上限的release方法就好了。此时第一个思路就是继承`RedissonSemaphore`这个类重写一下`release()`，但是后来发现自己的子类无法用`redissonClient`初始化出来，方案就pass掉了。所谓工作时间长了，什么都敢干，我直接打开源码，我只要仿照着写个一样的是不就行了！

![29cc673819515541c4d424da8b487b7e.png][]

public RFuture<Void> releaseAsync(int permits) {
        ...
        RFuture<Void> future = commandExecutor.evalWriteAsync(getRawName(), StringCodec.INSTANCE, RedisCommands.EVAL_VOID,
                "local value = redis.call('incrby', KEYS[1], ARGV[1]); " +
                        "redis.call('publish', KEYS[2], value); ",
                Arrays.asList(getRawName(), getChannelName()), permits);
        ...
    }
    复制代码

好家伙，核心逻辑其实就一行lua脚本`redis.call('incrby', KEYS[1], ARGV[1]);`，我只要一个判断上限的就好了。虽然之前也没写过lua脚本，但是好在逻辑不复杂，只需要一个`if`判断就行了，redisson源码中有很多这样的判断，现学现卖就可以了。

另外，鉴于面向对象的思想，如果我自己工具类写一个`release()`方法，并让这个方法直接出现在业务中多少有一些不妥。所以新建了一个对象，封装一下获取和释放两个方法，之后项目中所有信号量的相关业务，都用这个对象，redisson的信号量概念就不会再出现在具体的业务中了，可以提高业务代码的可读性，也会信号量相关操作的修改控制在这个类中。

经过了一番折腾，总算是把功能实现了。部分代码如下，`tryAcquire()`方法还是使用redisson提供的，但是`release()`方法用了自己实现的～

public class SemaphoreModel {
        /**
         * 信号量本体
         */
        private RedissonSemaphore semaphore;
    
        private RedissonClient redissonClient;
    
        /**
         * 获取到信号的mark
         */
        private boolean mark = false;
    
        public void release() {
             //获取信号量的值
            String script = "local value = redis.call('get', '" + semaphore.getRawName() + "'); " +
                    //如果没有设置过（false意味着undefind）或者当前值小于5（我设置的最大值）就做release的逻辑
                    "if (value ~= false and tonumber(value) < 5) then " +
                    //信号量加一
                    "local val = redis.call('incrby', '" + semaphore.getRawName() + "', 1); " +
                    "redis.call('publish', '" + RedissonSemaphore.getChannelName(semaphore.getRawName()) + "', val); " +
                    "return 1; " +
                    "end; " +
                    "return 0;";
            redissonClient.getScript().eval(RScript.Mode.READ_WRITE, script, RScript.ReturnType.VALUE);
        }
    
        /**
         * 获取一个信号量
         */
        public boolean tryAcquire() {
            return semaphore.tryAcquire();
        }
    
    }
    复制代码

经过了这样的设计，就造出了一个有上限的信号量，问题也都解决了。

本文的内容到这里就结束了，如果有不对的地方，欢迎在评论区指出，觉得有用的话别忘了点个赞哦～

作者：笨天才  
链接：https://juejin.cn/post/7197782794675585080

[9a510b5d8d4f884cbde23c34ce0efbf5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/9f0caf8fa8b34c3c90f6a85639b4cf6d.png
[e320ddb299961039213992828ecb9933.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/d09c793cf7d8428ea231fa443380d459.png
[8bcc1f19f0bcf60d930869ff50b60f70.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/d8052f64634b464fb642e3b997dcf884.png
[7debbc3ce1ea31a472946af1e31cc015.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/c90058c4f70b4916a33953563e2b4e4a.png
[25e526102bb7a32de542d897920e43ac.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/09909932fb064f8fa014aa0417f08a15.png
[29cc673819515541c4d424da8b487b7e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/fb28c82100924c2dad3becfb06f2be85.png