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

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);
        }
    }
}

@Override
public 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);
        }


    }


}

@Override
public 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);
    }


}

@PostConstruct
public 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);
            }
        }
    });
}

@Component
public 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);
   }
}

@Component
public class Cache {


    @Resource
    private RedisTemplate<String, String> RedisTemplate;


    @EnableBurning
    public String getResult(String Key) {
        return RedisTemplate.opsForValue().get(Key);
    }
}

@Repository
public 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);
   }
}