热点探测技术架构设计与实践

public synchronized static void startPipeline(BurningCommonProperties burningCommonProperties) {
    if (STARTED.get() == Boolean.FALSE) {        DwLogger.info("start pipeline");        // 设置参数上下文        setToContext(burningCommonProperties);        // 配置中心启动        EtcdConfigFactory.buildConfigCenter(burningCommonProperties.getConfigServer());        ConfigStarter starter = EtcdConfigStarter.getInstance();        starter.start();        // 注册中心启动        RegisterFactory.buildRegisterCenter(burningCommonProperties);        RegisterStarter registerStarter = RegisterStarter.getInstance();        registerStarter.start();        // 热点探测启动        DetectFactory.startDetect(burningCommonProperties.getPushPeriod());        // 开启worker重连器        WorkerRetryConnector.retryConnectWorkers();        // 注册事件监听        registEventBus();        // 开启监控        MetricsFactory.startMetrics();        STARTED.set(Boolean.TRUE);    }
}

public static Object dynamicGetValue(String key, KeyType keyType) {    try {        //如果没有为该key配置规则，就不用上报key        Boolean dynamicRule = dynamicRule(key);        if (dynamicRule == null) {            return null;        }        Object userValue = null;
        ValueModel value = getValueSimple(key);
        if (value == null) {            HotKeyPusher.push(key, keyType);        } else {            //临近过期了，也发            if (isNearExpire(value)) {                HotKeyPusher.push(key, keyType);            }            Object object = value.getValue();            //如果是默认值，也返回null            if (object instanceof Integer && Constant.MAGIC_NUMBER == (int) object) {                userValue = null;            } else if (Boolean.FALSE.equals(dynamicRule)) {                userValue = null;            } else {                userValue = object;            }        }
        //统计计数        MetricsFactory.metrics(new KeyHotModel(key, value != null));
        return userValue;    } catch (Exception e) {        DwLogger.error(DwHotKeyStore.class, "get value error");        return null;    }}

public void startNettyServer(int port) throws Exception {    //boss单线程    EventLoopGroup bossGroup = new NioEventLoopGroup(1);    //worker节点组    EventLoopGroup WorkerGroup = new NioEventLoopGroup(CpuNum.WorkerCount());    try {        ServerBootstrap bootstrap = new ServerBootstrap();        bootstrap.group(bossGroup, WorkerGroup)                .channel(NioServerSocketChannel.class)                .handler(new LoggingHandler(LogLevel.INFO))                .option(ChannelOption.SO_BACKLOG, 1024)                //保持长连接                .childOption(ChannelOption.SO_KEEPALIVE, true)                //出来网络io事件，如记录日志、对消息编解码等                .childHandler(new ChildChannelHandler());        //绑定端口，同步等待成功        ChannelFuture future = bootstrap.bind(port).sync();        Runtime.getRuntime().addShutdownHook(new Thread(() -> {            bossGroup.shutdownGracefully (1000, 3000, TimeUnit.MILLISECONDS);            WorkerGroup.shutdownGracefully (1000, 3000, TimeUnit.MILLISECONDS);        }));        //等待服务器监听端口关闭        future.channel().closeFuture().sync();    } catch (Exception e) {        DwLogger.error("netty server start error.", e);    } finally {        //优雅退出，释放线程池资源        bossGroup.shutdownGracefully();        WorkerGroup.shutdownGracefully();    }}

public void beginConsume() {    while (true) {        try {            HotKeyModel model = QUEUE.take();            if (model.isRemove()) {                iKeyListener.removeKey(model, KeyEventOriginal.CLIENT);            } else {                iKeyListener.newKey(model, KeyEventOriginal.CLIENT);            }            //处理完毕，将数量加1            totalDealCount.increment();        } catch (Exception e) {            DwLogger.error("consumer error.", e);        }    }}

@Overridepublic void newKey(HotKeyModel hotKeyModel, KeyEventOriginal original) {    //cache里的key    String key = buildKey(hotKeyModel);    String name = StringUtils.isBlank(hotKeyModel.getGroup()) ? hotKeyModel.getAppName() : hotKeyModel.getGroup();
    //判断是不是刚热不久    Object o = hotCache.getIfPresent(key);    if (o != null) {        return;    }    SlidingWindow slidingWindow = checkWindow(hotKeyModel, key, name);    //看看hot没    boolean hot = slidingWindow.addCount(hotKeyModel.getCount());
    if (!hot) {        //如果没hot，重新put，cache会自动刷新过期时间        CaffeineCacheHolder.getCache(name).put(key, slidingWindow);    } else {        hotCache.put(key, 1);
        //删掉该key        CaffeineCacheHolder.getCache(name).invalidate(key);
        //开启推送        hotKeyModel.setCreateTime(SystemClock.now());
        //当开关打开时，打印日志。大促时关闭日志，就不打印了        if (ConfigStarter.LOGGER_ON) {            DwLogger.info(NEW_KEY_EVENT + hotKeyModel.getKey());        }
        //分别推送到各client和etcd        for (IPusher pusher : iPushers) {            pusher.push(hotKeyModel);        }
    }
}

@Overridepublic void removeKey(HotKeyModel hotKeyModel, KeyEventOriginal original) {    //cache里的key    String key = buildKey(hotKeyModel);    String name = StringUtils.isBlank(hotKeyModel.getGroup()) ? hotKeyModel.getAppName() : hotKeyModel.getGroup();    hotCache.invalidate(key);    CaffeineCacheHolder.getCache(name).invalidate(key);
    //推送所有client删除    hotKeyModel.setCreateTime(SystemClock.now());    DwLogger.info(DELETE_KEY_EVENT + hotKeyModel.getKey());
    for (IPusher pusher : iPushers) {        pusher.remove(hotKeyModel);    }
}

@PostConstructpublic void batchPushToClient() {    AsyncPool.asyncDo(() -> {        while (true) {            try {                List<HotKeyModel> tempModels = new ArrayList<>();                //每10ms推送一次                Queues.drain(hotKeyStoreQueue, tempModels, 10, 10, TimeUnit.MILLISECONDS);                if (CollectionUtil.isEmpty(tempModels)) {                    continue;                }
                Map<String, List<HotKeyModel>> allAppHotKeyModels = Maps.newHashMap();                Map<String, List<HotKeyModel>> allGroupHotKeyModels = Maps.newHashMap();
                //拆分出每个app的热key集合，按app分堆                for (HotKeyModel hotKeyModel : tempModels) {                    if (StringUtils.isNotBlank(hotKeyModel.getGroup())) {                        List<HotKeyModel> groupModels = allGroupHotKeyModels.computeIfAbsent(hotKeyModel.getGroup(), (key) -> new ArrayList<>());                        groupModels.add(hotKeyModel);                    } else {                        List<HotKeyModel> oneAppModels = allAppHotKeyModels.computeIfAbsent(hotKeyModel.getAppName(), (key) -> new ArrayList<>());                        oneAppModels.add(hotKeyModel);                    }                }
                CustomizedMetricsProcessor processor = CustomizedMetricsProcessor.builder(MetricsConstant.BURNING_NETTY_OUT).build();
                // group hot key push                pushGroup(processor, allGroupHotKeyModels);
                // app hot key push                pushApp(processor, allAppHotKeyModels);
            } catch (Exception e) {                DwLogger.error("push to client error.", e);            }        }    });}

@Componentpublic class Cache {    @EnableBurning(prefix = "hot_Key_", cache = false, hitHandler = ExceptionHitHandler.class)    public String getResult2(String Key) {        return "这是一个测试结果" + Key;    }}

public class ExceptionHitHandler implements HitHandler {   @Override   public Object handle(String Key, ProceedingJoinPoint joinPoint) {       //此处可自定义实现      throw new RuntimeException("对不起，您没有权限访问: " + Key);   }}

@Componentpublic class Cache {
    @Resource    private RedisTemplate<String, String> RedisTemplate;
    @EnableBurning    public String getResult(String Key) {        return RedisTemplate.opsForValue().get(Key);    }}

@Repositorypublic class SmsSignRepo {
   @Autowired   private SmsSignMapper smsSignMapper;
   @EnableBurning(prefix = "SMS_SIGN", dynamic = false, KeyType = DATABASE_Key)   public List<SmsSign> getAll() {      Example example = new Example(SmsSign.class);      Example.Criteria criteria = example.createCriteria();      criteria.andEqualTo("status", 1);      return smsSignMapper.selectByExample(example);   }}