高可用之限流-03-Semaphore 信号量做限流

news2024/10/11 20:31:29

限流系列

开源组件 rate-limit: 限流

高可用之限流-01-入门介绍

高可用之限流-02-如何设计限流框架

高可用之限流-03-Semaphore 信号量做限流

高可用之限流-04-fixed window 固定窗口

高可用之限流-05-slide window 滑动窗口

高可用之限流-06-slide window 滑动窗口 sentinel 源码

高可用之限流-07-token bucket 令牌桶算法

高可用之限流 08-leaky bucket漏桶算法

高可用之限流 09-guava RateLimiter 入门使用简介 & 源码分析

主流的限流方式

目前主要有以下几种限流方式：

信号量
计数器
滑动窗口
漏桶算法
令牌桶算法
分布式限流

信号量

信号量实际上就是限制系统的并发量，来达到限流的目的。

常见的用法是：创建Semaphore，指定permit的数量。

在方法开始时，调用 Semaphore.acquire() 或者 Semaphore.tryAcquire() 来获取permit，并在方法返回前，调用Semaphore.release()来返还permit。

核心代码实现

public class LimitSemaphore extends LimitAdaptor {

    /**
     * 日志
     *
     * @since 0.0.5
     */
    private static final Log LOG = LogFactory.getLog(LimitSemaphore.class);

    /**
     * 信号量
     *
     * @since 0.0.5
     */
    private final Semaphore semaphore;

    /**
     * 构造器
     *
     * @param context 上下文
     * @since 0.0.5
     */
    public LimitSemaphore(final ILimitContext context) {
        this.semaphore = new Semaphore(context.count());
    }

    @Override
    public synchronized void acquire() {
        try {
            LOG.debug("[Limit] start acquire");
            this.semaphore.acquire(1);
            LOG.debug("[Limit] end acquire");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            LOG.error("[Limit] semaphore meet ex: ", e);
        }
    }

    @Override
    public void release() {
        LOG.debug("[Limit] start release");
        this.semaphore.release(1);
        LOG.debug("[Limit] end release");
    }

}

测试

我们限定每次只有一个线程可以执行核心方法，如下：

public class LimitSemaphoreTest {

    private static final Log LOG = LogFactory.getLog(LimitSemaphoreTest.class);

    private static final ILimit LIMIT = LimitBs.newInstance(LimitSemaphore.class)
            .count(1)
            .build();

    static class LimitRunnable implements Runnable {
        @Override
        public void run() {
            for(int i = 0; i < 2; i++) {
                try {
                    LIMIT.acquire();
                    LOG.info("{}-{}", Thread.currentThread().getName(), i);
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    LIMIT.release();
                }
            }
        }
    }

    public static void main(String[] args) {
        new Thread(new LimitRunnable()).start();
        new Thread(new LimitRunnable()).start();
    }

}

日志输出

13:35:37.501 [Thread-1] INFO  com.github.houbb.rate.limit.test.semaphore.LimitSemaphoreTest - Thread-1-0
13:35:38.501 [Thread-2] INFO  com.github.houbb.rate.limit.test.semaphore.LimitSemaphoreTest - Thread-2-0
13:35:39.502 [Thread-1] INFO  com.github.houbb.rate.limit.test.semaphore.LimitSemaphoreTest - Thread-1-1
13:35:40.503 [Thread-2] INFO  com.github.houbb.rate.limit.test.semaphore.LimitSemaphoreTest - Thread-2-1

可以看到每次只有一个线程可以执行方法。