怎么利用 Semaphore 的可重入性包装实现在分布式环境下对并发请求的动态限额配额管理

来源：3D模型作者：弥生美月头衔：网络博主
导读：本期聚焦于小伙伴创作的《怎么利用 Semaphore 的可重入性包装实现在分布式环境下对并发请求的动态限额配额管理》，敬请观看详情，探索知识的价值。以下视频、文章将为您系统阐述其核心内容与价值。如果您觉得《怎么利用 Semaphore 的可重入性包装实现在分布式环境下对并发请求的动态限额配额管理》有用，将其分享出去将是对创作者最好的鼓励。

在分布式系统架构中，多个服务实例同时接收外部请求时，若没有统一的并发管控机制，很容易出现数据库压力过大、下游接口被击穿等问题。传统的单机 Semaphore 仅能在单个 JVM 进程内生效，无法跨节点共享配额状态，因此需要对其做可重入性包装，适配分布式场景的配额管理需求。

怎么利用 Semaphore 的可重入性包装实现在分布式环境下对并发请求的动态限额配额管理

核心原理说明

Semaphore 的核心能力是通过维护一组许可证来控制并发访问量，可重入性指的是同一个线程可以多次获取许可证而不会被阻塞。在分布式场景下，我们需要把许可证的存储从单机内存迁移到共享的分布式存储中，同时记录每个请求线程的许可证获取次数，实现跨节点的可重入逻辑。

可重入 Semaphore 的单机实现参考

先回顾下单机可重入 Semaphore 的实现逻辑，核心是用 ThreadLocal 记录当前线程的获取次数：

import java.util.concurrent.Semaphore;

public class ReentrantSemaphore {
    // 基础Semaphore，控制总许可证数量
    private final Semaphore semaphore;
    // 记录每个线程的许可证获取次数
    private final ThreadLocal<Integer> holdCount = ThreadLocal.withInitial(() -> 0);

    public ReentrantSemaphore(int permits) {
        this.semaphore = new Semaphore(permits);
    }

    public void acquire() throws InterruptedException {
        // 如果当前线程已经持有许可证，直接递增计数
        if (holdCount.get() > 0) {
            holdCount.set(holdCount.get() + 1);
            return;
        }
        // 首次获取，调用原生Semaphore的acquire方法
        semaphore.acquire();
        holdCount.set(1);
    }

    public void release() {
        int count = holdCount.get();
        if (count == 0) {
            throw new IllegalMonitorStateException("当前线程未持有许可证");
        }
        if (count == 1) {
            // 最后一次释放，调用原生Semaphore的release方法
            holdCount.remove();
            semaphore.release();
        } else {
            // 还有持有次数，递减计数
            holdCount.set(count - 1);
        }
    }

    // 获取当前总许可证数量
    public int availablePermits() {
        return semaphore.availablePermits();
    }
}

分布式场景的适配改造

单机版本的可重入 Semaphore 无法跨 JVM 共享状态，因此需要把许可证的计数、线程持有记录存储到分布式中间件中，这里以 Redis 为例实现改造。

核心改造点

  • 用 Redis 的 String 结构存储全局可用的许可证数量，保证所有节点共享同一份配额数据
  • 用 Redis 的 Hash 结构存储每个节点的线程持有许可证次数，key 为节点标识+线程ID，value 为持有次数
  • 许可证的获取和释放操作需要通过 Redis 的 Lua 脚本实现，保证原子性，避免并发问题
  • 支持动态修改许可证总数，直接更新 Redis 中存储的总配额值即可

分布式可重入 Semaphore 实现代码

首先定义 Lua 脚本，实现许可证的原子获取逻辑：

-- 获取许可证的Lua脚本
-- KEYS[1]: 全局许可证数量key
-- KEYS[2]: 线程持有记录hash key
-- ARGV[1]: 当前节点标识+线程ID
-- ARGV[2]: 初始许可证总数（用于初始化场景）
local permitKey = KEYS[1]
local holdKey = KEYS[2]
local threadId = ARGV[1]
local initPermits = tonumber(ARGV[2])

-- 初始化全局许可证数量（如果不存在的话）
if redis.call("exists", permitKey) == 0 then
    redis.call("set", permitKey, initPermits)
end

-- 查看当前线程是否已经持有许可证
local holdCount = redis.call("hget", holdKey, threadId)
if holdCount and tonumber(holdCount) > 0 then
    -- 已经持有，递增计数
    redis.call("hincrby", holdKey, threadId, 1)
    return 1
else
    -- 未持有，尝试获取许可证
    local currentPermits = tonumber(redis.call("get", permitKey))
    if currentPermits > 0 then
        -- 获取成功，递减全局许可证数量，记录持有次数
        redis.call("decr", permitKey)
        redis.call("hset", holdKey, threadId, 1)
        return 1
    else
        -- 获取失败
        return 0
    end
end

然后是许可证释放的 Lua 脚本：

-- 释放许可证的Lua脚本
-- KEYS[1]: 全局许可证数量key
-- KEYS[2]: 线程持有记录hash key
-- ARGV[1]: 当前节点标识+线程ID
local permitKey = KEYS[1]
local holdKey = KEYS[2]
local threadId = ARGV[1]

local holdCount = redis.call("hget", holdKey, threadId)
if not holdCount or tonumber(holdCount) == 0 then
    return -1
end

if tonumber(holdCount) == 1 then
    -- 最后一次释放，递增全局许可证数量，删除持有记录
    redis.call("incr", permitKey)
    redis.call("hdel", holdKey, threadId)
    return 1
else
    -- 还有持有次数，递减计数
    redis.call("hincrby", holdKey, threadId, -1)
    return 1
end

最后是 Java 封装的分布式可重入 Semaphore 类：

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import java.util.Collections;
import java.util.UUID;

public class DistributedReentrantSemaphore {
    // Redis中全局许可证数量的key
    private final String permitKey;
    // Redis中线程持有记录的hash key
    private final String holdKey;
    // 当前节点标识
    private final String nodeId;
    // Redis连接池
    private final JedisPool jedisPool;
    // 初始许可证数量
    private final int initPermits;
    // 获取许可证的Lua脚本
    private final String acquireScript;
    // 释放许可证的Lua脚本
    private final String releaseScript;

    public DistributedReentrantSemaphore(String permitKey, String holdKey, int initPermits, JedisPool jedisPool) {
        this.permitKey = permitKey;
        this.holdKey = holdKey;
        this.initPermits = initPermits;
        this.jedisPool = jedisPool;
        this.nodeId = UUID.randomUUID().toString();
        // 加载获取许可证的Lua脚本
        this.acquireScript = "local permitKey = KEYS[1]n" +
                "local holdKey = KEYS[2]n" +
                "local threadId = ARGV[1]n" +
                "local initPermits = tonumber(ARGV[2])n" +
                "if redis.call("exists", permitKey) == 0 thenn" +
                "    redis.call("set", permitKey, initPermits)n" +
                "endn" +
                "local holdCount = redis.call("hget", holdKey, threadId)n" +
                "if holdCount and tonumber(holdCount) > 0 thenn" +
                "    redis.call("hincrby", holdKey, threadId, 1)n" +
                "    return 1n" +
                "elsen" +
                "    local currentPermits = tonumber(redis.call("get", permitKey))n" +
                "    if currentPermits > 0 thenn" +
                "        redis.call("decr", permitKey)n" +
                "        redis.call("hset", holdKey, threadId, 1)n" +
                "        return 1n" +
                "    elsen" +
                "        return 0n" +
                "    endn" +
                "end";
        // 加载释放许可证的Lua脚本
        this.releaseScript = "local permitKey = KEYS[1]n" +
                "local holdKey = KEYS[2]n" +
                "local threadId = ARGV[1]n" +
                "local holdCount = redis.call("hget", holdKey, threadId)n" +
                "if not holdCount or tonumber(holdCount) == 0 thenn" +
                "    return -1n" +
                "endn" +
                "if tonumber(holdCount) == 1 thenn" +
                "    redis.call("incr", permitKey)n" +
                "    redis.call("hdel", holdKey, threadId)n" +
                "    return 1n" +
                "elsen" +
                "    redis.call("hincrby", holdKey, threadId, -1)n" +
                "    return 1n" +
                "end";
    }

    public void acquire() throws InterruptedException {
        String threadId = nodeId + ":" + Thread.currentThread().getId();
        try (Jedis jedis = jedisPool.getResource()) {
            // 执行获取许可证的Lua脚本
            Object result = jedis.eval(acquireScript, 2, permitKey, holdKey, threadId, String.valueOf(initPermits));
            Long res = (Long) result;
            if (res == 0) {
                // 获取失败，自旋等待
                Thread.sleep(100);
                acquire();
            }
        }
    }

    public void release() {
        String threadId = nodeId + ":" + Thread.currentThread().getId();
        try (Jedis jedis = jedisPool.getResource()) {
            Object result = jedis.eval(releaseScript, 2, permitKey, holdKey, threadId);
            Long res = (Long) result;
            if (res == -1) {
                throw new IllegalMonitorStateException("当前线程未持有许可证");
            }
        }
    }

    // 动态更新许可证总数
    public void updatePermits(int newPermits) {
        try (Jedis jedis = jedisPool.getResource()) {
            // 计算新增的许可证数量
            int currentPermits = Integer.parseInt(jedis.get(permitKey));
            int delta = newPermits - (currentPermits + getHeldCount());
            if (delta > 0) {
                // 新增许可证，直接增加全局计数
                jedis.incrBy(permitKey, delta);
            } else if (delta < 0) {
                // 减少许可证，需要判断当前可用数量是否足够
                int available = currentPermits;
                if (available >= -delta) {
                    jedis.decrBy(permitKey, -delta);
                } else {
                    // 可用数量不足，先设置为0，等后续释放后再生效
                    jedis.set(permitKey, "0");
                }
            }
        }
    }

    // 获取当前线程持有的许可证数量（辅助方法）
    private int getHeldCount() {
        try (Jedis jedis = jedisPool.getResource()) {
            String threadId = nodeId + ":" + Thread.currentThread().getId();
            String count = jedis.hget(holdKey, threadId);
            return count == null ? 0 : Integer.parseInt(count);
        }
    }
}

使用示例

在实际业务中，可以用这个分布式可重入 Semaphore 来管控接口的总并发量，示例如下：

import redis.clients.jedis.JedisPool;

public class RequestLimitExample {
    public static void main(String[] args) {
        // 初始化Redis连接池，连接本地Redis
        JedisPool jedisPool = new JedisPool("127.0.0.1", 6379);
        // 创建分布式可重入Semaphore，全局配额key为request_permits，持有记录key为request_hold，初始配额10
        DistributedReentrantSemaphore semaphore = new DistributedReentrantSemaphore(
                "request_permits",
                "request_hold",
                10,
                jedisPool
        );

        // 模拟处理请求
        new Thread(() -> {
            try {
                semaphore.acquire();
                System.out.println("线程1获取许可证成功，开始处理请求");
                // 模拟可重入场景，同一线程再次获取许可证
                semaphore.acquire();
                System.out.println("线程1再次获取许可证成功");
                // 处理业务逻辑
                Thread.sleep(2000);
                // 释放许可证
                semaphore.release();
                semaphore.release();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        // 动态修改配额，调整为20
        semaphore.updatePermits(20);
    }
}

注意事项

  • Redis 需要保证高可用，建议使用 Redis 集群或者哨兵模式，避免单点故障导致配额管理失效
  • Lua 脚本的执行效率很高，但如果并发量特别大，可以考虑对热点 key 做本地缓存优化
  • 线程持有记录的 Hash 结构需要设置过期时间，避免节点下线后无效记录一直占用存储空间
  • 动态修改配额时，减少配额的操作不会立即回收已经分配的许可证，需要等线程主动释放后才会生效

Semaphore可重入性分布式环境动态限额配额管理修改时间：2026-07-10 11:22:04

免责声明：​ 已尽一切努力确保本网站所含信息的准确性。网站内容多为原创整理与精心编撰，观点力求客观中立。本站旨在免费分享，内容仅供个人学习、研究或参考使用。若引用了第三方作品，版权归原作者所有。如内容涉及您的权益，请联系我们处理。

上一篇如何在MacOS中验证Java开发环境是否正常

下一篇怎么在Windows后台服务中运行Tomcat 注册系统服务的完整步骤

猜你喜欢

内容垂直聚焦
专注技术核心技术栏目，确保每篇文章深度聚焦于实用技能。从代码技巧到架构设计，为用户提供无干扰的纯技术知识沉淀，精准满足专业提升需求。
知识结构清晰
覆盖从开发到部署的全链路。AI、前端、编程、数据库、服务器、建站、系统层层递进，构建清晰学习路径，帮助用户系统化掌握开发与运维所需的核心技术。
深度技术解析
拒绝泛泛而谈，深入技术细节与实践难点。无论是数据库优化还是服务器配置，均结合真实场景与代码示例进行剖析，致力于提供可直接应用于工作的解决方案。
专业领域覆盖
精准对应开发生命周期。从前端界面到后端编程，从数据库操作到服务器运维，形成完整闭环，一站式满足全栈工程师和运维人员的技术需求。
即学即用高效
内容强调实操性，步骤清晰、代码完整。用户可根据教程直接复现和应用于自身项目，显著缩短从学习到实践的距离，快速解决开发中的具体问题。
持续更新保障
专注既定技术方向进行长期、稳定的内容输出。确保各栏目技术文章持续更新迭代，紧跟主流技术发展趋势，为用户提供经久不衰的学习价值。

负责任网站 - 免责声明 - 隐私政策 - 网站地图

举报邮箱：chomcom@qq.com

Copyright (C) www.ipipp.com All Rights Reserved.