bucket4j限流示例

女爷i 2021-10-18 10:18 364阅读 0赞

最近处理测试某个业务的性能，发现当kafka消息量特别大的时候需要限制kafka消息消费速度，因为接受消息的处理流程比较多，当消息量特别大的时候，如果kafka消息了消息（kafka是自动完成commit）提交给后台处理，一旦后台线程中断，就会导致有消息遗漏处理。  
这里补充一句，为什么没有设置kafka手动提交，因为每个消息的处理流程有差异，时间不一，整体上只要接受到kafka消息，完成基本处理，就提交到线程池中了（有的消息要提交给多个线程池）， 无法等没有线程池都处理完毕了才手动向kafka提交确认。

# bucket4j #

bucket4j是基于令牌桶算法的Java限流库， 主页在https://github.com/vladimir-bukhtoyarov/bucket4j。 它主要用在3种场景：  
a，限制比较重工作的速率。  
b，将限流作为定时器，例如有些场景限制你对服务提供方的调用速度，因此使用限流器作为定时器，定时按照约定速率调用服务提供方。  
c，限制对API访问速率。

# 示例简介 #

为了说明问题，我们开发一个简单的SpringBoot工程，就一个Rest API， 该接口会根据租户id进行限流。限流速度：每30秒2次。 因为逻辑比较简单就不再说明，直接看代码。

注意：这个示例只是演示如何使用bucket4j，实际工程中，对API调用的限流我们一般在API网关处完成。而不是在具体的API处。，本例子只是为了说明如何限流， 因为我们除了对API限流，也会多其他业务处理限流，例如我遇到的处理消息情况。

RateLimits 生成

public class RateLimits { 
        private final LocalBucket bucket;
    
        public RateLimits(String limitsConfiguration) { 
            LocalBucketBuilder builder = Bucket4j.builder();
            boolean initialized = false;
            for (String limitSrc : limitsConfiguration.split(",")) { 
                long capacity = Long.parseLong(limitSrc.split(":")[0]);
                long duration = Long.parseLong(limitSrc.split(":")[1]);
                builder.addLimit(Bandwidth.simple(capacity, Duration.ofSeconds(duration)));
                initialized = true;
            }
            if (initialized) { 
                bucket = builder.build();
            } else { 
                throw new IllegalArgumentException("Failed to parse rate limits configuration: " + limitsConfiguration);
            }
        }
    
        public boolean tryConsume() { 
            return bucket.tryConsume(1);
        }
    
        public long getAvailableTokens() { 
            return bucket.getAvailableTokens();
        }
    }

服务层

@Service
    @Slf4j
    public class RateLimitSvcImpl implements RateLimitSvc { 
        @Value("${rest.limits.tenant.enabled:false}")
        private boolean perTenantLimitsEnabled;
        @Value("${rest.limits.tenant.configuration:}")
        private String perTenantLimitsConfiguration;
    
        private ConcurrentMap<String, RateLimits> perTenantLimits = new ConcurrentHashMap<>();
    
        @Override
        public boolean execRateLimit(String tenantId) { 
            if (perTenantLimitsEnabled) { 
                RateLimits rateLimits = perTenantLimits.computeIfAbsent(tenantId, id -> new RateLimits(perTenantLimitsConfiguration));
                if (!rateLimits.tryConsume()) { 
                    log.info("tryConsume false, tenantId={}, leftToken={}", tenantId, rateLimits.getAvailableTokens());
                    return false;
                } else { 
                    log.info("tryConsume true, tenantId={}, leftToken={}", tenantId, rateLimits.getAvailableTokens());
                    return true;
                }
    
            return true;
        }
    }

具体API

@RestController
    @RequestMapping
    public class MyController { 
    
        //实际上我们一般不会再controller中直接进行rateLimit， 而是在网关处，根据租户id，用户id，应用id，或者ip进行限流。
        //本程序只是RateLimit的例子，不建议直接在生产代码中使用
        @Autowired
        private RateLimitSvc rateLimitSvc;
    
    
        @ApiOperation(value = "按用户id查询")
        @ApiImplicitParams({ 
                @ApiImplicitParam(name = "tenantId", value = "tenantId", defaultValue = "001", required = true, dataType = "string", paramType = "path"),
        })
        @GetMapping(value = "/tenants/{tenantId}", produces = "application/json;charset=UTF-8")
        public String getDevice(@PathVariable String tenantId) { 
            if (rateLimitSvc.execRateLimit(tenantId)) { 
                //这里实际中应该是调用设备服务查询数据库，本示例为了简化直接new了一个对象
                Device device = new Device();
                device.setId("001");
                device.setName("一号设备");
                return JSONObject.toJSONString(device, SerializerFeature.WriteMapNullValue);
            } else { 
                JSONObject json = new JSONObject();
                json.put("errMsg", "too many requests");
                return json.toJSONString();
            }
        }
    }

完整的代码在[这里][Link 1]，欢迎fork， 加星。 谢谢！

# 效果截图 #

1， swagger访问效果  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1c3NsZQ_size_16_color_FFFFFF_t_70]  
2， 日志信息  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1c3NsZQ_size_16_color_FFFFFF_t_70 1]

[Link 1]: https://github.com/yqbjtu/springboot/tree/master/rateLimitDemo
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1c3NsZQ_size_16_color_FFFFFF_t_70]: /images/20211018/33233e478aa641198c83f0bc267c5659.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1c3NsZQ_size_16_color_FFFFFF_t_70 1]: /images/20211018/dd897e45c5f14d89a48a20766f77b6df.png