可能要用心学高并发核心编程,限流原理与实战,分布式令牌桶限流

简介:   实战:分布式令牌桶限流  本节介绍的分布式令牌桶限流通过Lua+Java结合完成,首先在Lua脚本中完成限流的计算,然后在Java代码中进行组织和调用。  分布式令牌桶限流Lua脚本  分布式令牌桶限流Lua脚本的核心逻辑和Java令牌桶的执行逻辑类似,只是限流计算相关的统计和时间数据存放于Redis中。  这里将限流的脚本命名为rate_limiter.lua,该脚本既使用Redis存储令牌桶信息,自身又执行于Redis中,所以笔者将该脚本放置于base-redis基础模块中,它的代码如下:  ---此脚本的环境:redis内部,不是运行在Nginx内部

  实战:分布式令牌桶限流

  本节介绍的分布式令牌桶限流通过Lua+Java结合完成,首先在Lua脚本中完成限流的计算,然后在Java代码中进行组织和调用。

  分布式令牌桶限流Lua脚本

  分布式令牌桶限流Lua脚本的核心逻辑和Java令牌桶的执行逻辑类似,只是限流计算相关的统计和时间数据存放于Redis中。

  这里将限流的脚本命名为rate_limiter.lua,该脚本既使用Redis存储令牌桶信息,自身又执行于Redis中,所以笔者将该脚本放置于base-redis基础模块中,它的代码如下:

  ---此脚本的环境:redis内部,不是运行在Nginx内部

  ---方法:申请令牌

  ----1:failed

  ---1:success

  ---@param key:key限流关键字

  ---@param apply:申请的令牌数量

  local function acquire(key, apply)

  local times=redis.call('TIME');

  --times[1] 秒数 --times[2] 微秒数

  local curr_mill_second=times[1] *1000000 + times[2];

  curr_mill_second=curr_mill_second / 1000;

  local cacheInfo=redis.pcall("HMGET", key, "last_mill_second", "curr_permits", "max_permits", "rate")

  ---局部变量:上次申请的时间

  local last_mill_second=cacheInfo[1];

  ---局部变量:之前的令牌数

  local curr_permits=tonumber(cacheInfo[2]);

  ---局部变量:桶的容量

  local max_permits=tonumber(cacheInfo[3]);

  ---局部变量:令牌的发放速率

  local rate=cacheInfo[4];

  ---局部变量:本次的令牌数

  local local_curr_permits=max_permits;

  if (type(last_mill_second) ~='boolean' and last_mill_second ~=nil) then

  --计算时间段内的令牌数

  local reverse_permits=math.floor(((curr_mill_second - last_mill_second) / 1000) *rate);

  --令牌总数

  local expect_curr_permits=reverse_permits + curr_permits;

  --可以申请的令牌总数

  local_curr_permits=math.min(expect_curr_permits, max_permits);

  else

  --第一次获取令牌

  redis.pcall("HSET", key, "last_mill_second", curr_mill_second)

  end

  local result=-1;

  --有足够的令牌可以申请

  if (local_curr_permits - apply >=0) then

  --保存剩余的令牌

  redis.pcall("HSET", key, "curr_permits", local_curr_permits - apply);

  --保存时间,下次令牌获取时使用

  redis.pcall("HSET", key, "last_mill_second", curr_mill_second)

  --返回令牌获取成功

  result=1;

  else

  --保存令牌总数

  redis.pcall("HSET", key, "curr_permits", local_curr_permits);

  --返回令牌获取失败

  result=-1;

  end

  return result

  end

  ---方法:初始化限流器

  ---1 success

  ---@param key key

  ---@param max_permits 桶的容量

  ---@param rate 令牌的发放速率

  local function init(key, max_permits, rate)

  local rate_limit_info=redis.pcall("HMGET", key, "last_mill_second", "curr_permits", "max_permits", "rate")

  local org_max_permits=tonumber(rate_limit_info[3])

  local org_rate=rate_limit_info[4]

  if (org_max_permits==nil) or (rate ~=org_rate or max_permits ~=org_max_permits) then

  redis.pcall("HMSET", key, "max_permits", max_permits, "rate", rate, "curr_permits", max_permits)

  end

  return 1;

  end

  ---方法:删除限流Key

  local function delete(key)

  redis.pcall("DEL", key) return 1;

  end

  local key=KEYS[1]

  local method=ARGV[1]

  if method=='acquire' then

  return acquire(key, ARGV[2], ARGV[3])

  elseif method=='init' then

  return init(key, ARGV[2], ARGV[3])

  elseif method=='delete' then

  return delete(key)

  else

  --ignore

  end

  该脚本有3个方法,其中两个方法比较重要,分别说明如下:

  (1)限流器初始化方法init(key,max_permits,rate),此方法在限流开始时被调用。

  (2)限流检测的方法acquire(key,apply),此方法在请求到来时被调用。

  Java分布式令牌桶限流

  rate_limiter.lua脚本既可以在Java中调用,又可以在Nginx中调用。本小节先介绍其在Java中的使用,第10章再介绍其在Nginx中的使用。

  Java分布式令牌桶限流器的实现就是通过Java代码向Redis加载rate_limiter.lua脚本,然后封装其令牌桶初始化方法init(...)和限流监测方法acquire(...),以供外部调用。它的代码如下:

  package com.crazymaker.springcloud.standard.ratelimit;

  ...

  /**

  *实现:令牌桶限流服务

  *create by尼恩 @ 富贵论坛

  **/

  @Slf4j

  public class RedisRateLimitImpl implements RateLimitService, InitializingBean

  {

  /**

  *限流器的redis key前缀

  */

  private static final String RATE_LIMITER_KEY_PREFIX="rate_limiter:";

  //private ScheduledExecutorService executorService=Executors.newScheduledThreadPool(1);

  private RedisRateLimitProperties redisRateLimitProperties;

  private RedisTemplate redisTemplate;

  //lua脚本的实例

  private static RedisScript rateLimiterScript=null;

  //lua脚本的类路径

  private static String rateLimitLua="script/rate_limiter.lua";

  static

  {

  //从类路径文件中加载令牌桶lua脚本

  String script=IOUtil.loadJarFile(RedisRateLimitImpl.class.getClassLoader(), rateLimitLua);

  if (StringUtils.isEmpty(script))

  {

  log.error("lua script load failed:" + rateLimitLua);

  } else

  {

  //创建Lua脚本实例

  rateLimiterScript=new DefaultRedisScript<>(script, Long.class);

  }

  }

  public RedisRateLimitImpl(

  RedisRateLimitProperties redisRateLimitProperties,

  RedisTemplate redisTemplate)

  {

  thisisRateLimitProperties=redisRateLimitProperties;

  thisisTemplate=redisTemplate;

  }

  private Map<String, LimiterInfo> limiterInfoMap=new HashMap<>();

  /**

  *限流器的信息

  */

  @Builder

  @Data

  public static class LimiterInfo

  {

  /**

  *限流器的key,如秒杀的id

  */ private String key;

  /**

  *限流器的类型,如seckill

  */

  private String type="default";

  /**

  *限流器的最大桶容量

  */

  private Integer maxPermits;

  /**

  *限流器的速率

  */

  private Integer rate;

  /**

  *限流器的redis key

  */

  public String fullKey()

  {

  return RATE_LIMITER_KEY_PREFIX + type + ":" + key;

  }

  /**

  *限流器在map中的缓存key

  */

  public String cashKey()

  {

  return type + ":" + key;

  }

  }

  /**

  *限流检测:是否超过redis令牌桶限速器的限制

  *

  *@param cacheKey计数器的key

  *@return true or false

  */

  @Override

  public Boolean tryAcquire(String cacheKey)

  {

  if (cacheKey==null)

  {

  return true;

  }

  if (cacheKey.indexOf(":") <=0)

  {

  cacheKey="default:" + cacheKey;

  }

  LimiterInfo limiterInfo=limiterInfoMap.get(cacheKey);

  if (limiterInfo==null)

  {

  return true;

  }

  Long acquire=(Long) redisTemplate.execute(rateLimiterScript,

  ImmutableList.of(limiterInfo.fullKey()),

  "acquire",

  "1");

  if (acquire==1)

  {

  return false;

  }

  return true;

  }

  /**

  *重载方法:限流器初始化

  *

  *@param limiterInfo限流的类型

  */

  public void initLimitKey(LimiterInfo limiterInfo)

  {

  if (null==rateLimiterScript)

  {

  return;

  }

  String maxPermits=limiterInfo.getMaxPermits().toString();

  String rate=limiterInfo.getRate().toString();

  //执行redis脚本

  Long result=(Long) redisTemplate.execute(rateLimiterScript,

  ImmutableList.of(limiterInfo.fullKey()),

  "init",

  maxPermits,

  rate); limiterInfoMap.put(limiterInfo.cashKey(), limiterInfo);

  }

  /**

  *限流器初始化

  *

  *@param type类型

  *@param key id

  *@param maxPermits上限

  *@param rate 速度

  */

  public void initLimitKey(String type, String key,

  Integer maxPermits, Integer rate)

  {

  LimiterInfo limiterInfo=LimiterInfo.builder()

  .type(type)

  .key(key)

  .maxPermits(maxPermits)

  .rate(rate)

  .build();

  initLimitKey(limiterInfo);

  }

  /**

  *获取redis lua脚本的sha1编码,并缓存到redis

  */

  public String cacheSha1()

  {

  String sha1=rateLimiterScript.getSha1();

  redisTemplate.opsForValue().set("lua:sha1:rate_limiter", sha1);

  return sha1;

  }

  }Java分布式令牌桶限流的自验证

  自验证的工作:首先初始化分布式令牌桶限流器,然后使用两条

  线程不断进行限流的检测。自验证的代码如下:

  package com.crazymaker.springcloud.ratelimit;

  ...

  @Slf4j

  @RunWith(SpringRunner.class)

  //指定启动类

  @SpringBootTest(classes={DemoCloudApplication.class})

  /**

  *redis分布式令牌桶测试类

  */

  public class RedisRateLimitTest

  {

  @Resource(name="redisRateLimitImpl")

  RedisRateLimitImpl limitService;

  //线程池,用于多线程模拟测试

  private ExecutorService pool=Executors.newFixedThreadPool(10);

  @Test

  public void testRedisRateLimit()

  {

  //初始化分布式令牌桶限流器

  limitService.initLimitKey(

  "seckill", //redis key中的类型

  "10000", //redis key中的业务key,比如商品id

  2, //桶容量

  2); //每秒令牌数

  AtomicInteger count=new AtomicInteger();

  long start=System.currentTimeMillis();

  //线程数

  final int threads=2;

  //每条线程的执行轮数

  final int turns=20;

  //同步器

  CountDownLatch countDownLatch=new CountDownLatch(threads);

  for (int i=0; i < threads; i++)

  {

  pool.submit(() ->

  {

  try

  {

  //每个用户访问turns次

  for (int j=0; j < turns; j++)

  {

  boolean limited=limitService.tryAcquire

  ("seckill:10000");

  if (limited)

  {

  count.getAndIncrement();

  }

  Thread.sleep(200);

  }

  } catch (Exception e)

  { e.printStackTrace();

  }

  countDownLatch.countDown();

  });

  }

  try

  {

  countDownLatch.await();

  } catch (InterruptedException e)

  {

  e.printStackTrace();

  }

  float time=(System.currentTimeMillis() - start) / 1000F;

  //输出统计结果

  log("限制的次数为:" + count.get() + " 时长为:" + time);

  log("限制的次数为:" + count.get() +

  ",通过的次数为:" + (threads *turns - count.get()));

  log("限制的大专比例为:" +

  (float) count.get() / (float) (threads *turns));

  log("运行的时长为:" + time);

  try

  {

  Thread.sleep(Integer.MAX_VALUE);

  } catch (InterruptedException e)

  {

  e.printStackTrace();

  }

  }

  }

  两条线程各运行20次,每一次运行休眠200毫秒,总计耗时4秒,

  运行40次,部分输出结果如下:

  [main] INFO c.c.s.risRateLimitTest - 限制的次数为:32 时长为:4.015

  [main] INFO c.c.s.risRateLimitTest - 限制的次数为:32,通过的次数为:8

  [main] INFO c.c.s.risRateLimitTest - 限制的比例为:0.8

  [main] INFO c.c.s.risRateLimitTest - 运行的时长为:4.015

  大家可以自行调整参数,运行以上自验证程序并观察实验结果,体验一下分布式令牌桶限流的效果。

  本文给大家讲解的内容是高并发核心编程,限流原理与实战,实战:分布式令牌桶限流下篇文章给大家讲解的是高并发核心编程,Spring Cloud+Nginx秒杀实战;觉得文章不错的朋友可以转发此文关注小编;感谢大家的支持!

目录
相关文章
|
6月前
|
人工智能 Kubernetes 数据可视化
Kubernetes下的分布式采集系统设计与实战:趋势监测失效引发的架构进化
本文回顾了一次关键词监测任务在容器集群中失效的全过程,分析了中转IP复用、调度节奏和异常处理等隐性风险,并提出通过解耦架构、动态IP分发和行为模拟优化采集策略,最终实现稳定高效的数据抓取与分析。
113 2
Kubernetes下的分布式采集系统设计与实战:趋势监测失效引发的架构进化
|
6月前
|
数据采集 缓存 NoSQL
分布式新闻数据采集系统的同步效率优化实战
本文介绍了一个针对高频新闻站点的分布式爬虫系统优化方案。通过引入异步任务机制、本地缓存池、Redis pipeline 批量写入及身份池策略,系统采集效率提升近两倍,数据同步延迟显著降低,实现了分钟级热点追踪能力,为实时舆情监控与分析提供了高效、稳定的数据支持。
262 1
分布式新闻数据采集系统的同步效率优化实战
|
7月前
|
缓存 NoSQL 算法
高并发秒杀系统实战(Redis+Lua分布式锁防超卖与库存扣减优化)
秒杀系统面临瞬时高并发、资源竞争和数据一致性挑战。传统方案如数据库锁或应用层锁存在性能瓶颈或分布式问题,而基于Redis的分布式锁与Lua脚本原子操作成为高效解决方案。通过Redis的`SETNX`实现分布式锁,结合Lua脚本完成库存扣减,确保操作原子性并大幅提升性能(QPS从120提升至8,200)。此外,分段库存策略、多级限流及服务降级机制进一步优化系统稳定性。最佳实践包括分层防控、黄金扣减法则与容灾设计,强调根据业务特性灵活组合技术手段以应对高并发场景。
2046 7
|
8月前
|
监控 Java 调度
SpringBoot中@Scheduled和Quartz的区别是什么?分布式定时任务框架选型实战
本文对比分析了SpringBoot中的`@Scheduled`与Quartz定时任务框架。`@Scheduled`轻量易用,适合单机简单场景,但存在多实例重复执行、无持久化等缺陷;Quartz功能强大,支持分布式调度、任务持久化、动态调整和失败重试,适用于复杂企业级需求。文章通过特性对比、代码示例及常见问题解答,帮助开发者理解两者差异,合理选择方案。记住口诀:单机简单用注解,多节点上Quartz;若是任务要可靠,持久化配置不能少。
769 4
|
消息中间件 Java Linux
2024年最全BATJ真题突击:Java基础+JVM+分布式高并发+网络编程+Linux(1),2024年最新意外的惊喜
2024年最全BATJ真题突击:Java基础+JVM+分布式高并发+网络编程+Linux(1),2024年最新意外的惊喜
|
4月前
|
数据采集 存储 弹性计算
高并发Java爬虫的瓶颈分析与动态线程优化方案
高并发Java爬虫的瓶颈分析与动态线程优化方案
|
缓存 NoSQL Java
Java高并发实战:利用线程池和Redis实现高效数据入库
Java高并发实战:利用线程池和Redis实现高效数据入库
1057 0
|
存储 NoSQL Java
探索Java分布式锁:在高并发环境下的同步访问实现与优化
【6月更文挑战第30天】Java分布式锁在高并发下确保数据一致性,通过Redis的SETNX、ZooKeeper的临时节点、数据库操作等方式实现。优化策略包括锁超时重试、续期、公平性及性能提升,关键在于平衡同步与效率,适应大规模分布式系统的需求。
494 1
|
监控 算法 Java
企业应用面临高并发等挑战,优化Java后台系统性能至关重要
随着互联网技术的发展,企业应用面临高并发等挑战,优化Java后台系统性能至关重要。本文提供三大技巧:1)优化JVM,如选用合适版本(如OpenJDK 11)、调整参数(如使用G1垃圾收集器)及监控性能;2)优化代码与算法,减少对象创建、合理使用集合及采用高效算法(如快速排序);3)数据库优化,包括索引、查询及分页策略改进,全面提升系统效能。
219 0