const M = N * 2
//生成随机数
random = GenRandom(0, M)
//构造备份新key
bakHotKey = hotKey + “_” + random
data = redis.GET(bakHotKey)
if data == NULL {
  data = GetFromDB()
  redis.SET(bakHotKey, expireTime + GenRandom(0,5))
}

@PostConstruct
public void init() {
    ClientStarter.Builder builder = new ClientStarter.Builder();
    ClientStarter starter = builder.setAppName(appName).setEtcdServer(etcd).build();
    starter.startPipeline();
}

/**
 * 启动监听etcd
 */
public void startPipeline() {
    JdLogger.info(getClass(), "etcdServer:" + etcdServer);
    //设置caffeine的最大容量
    Context.CAFFEINE_SIZE = caffeineSize;

    //设置etcd地址
    EtcdConfigFactory.buildConfigCenter(etcdServer);
    //开始定时推送
    PushSchedulerStarter.startPusher(pushPeriod);
    PushSchedulerStarter.startCountPusher(10);
    //开启worker重连器
    WorkerRetryConnector.retryConnectWorkers();

    registEventBus();

    EtcdStarter starter = new EtcdStarter();
    //与etcd相关的监听都开启
    starter.start();
}

//设置caffeine的最大容量
Context.CAFFEINE_SIZE = caffeineSize;

//设置etcd地址
EtcdConfigFactory.buildConfigCenter(etcdServer);

public class Context {
    public static String APP_NAME;

    public static int CAFFEINE_SIZE;
}

public class EtcdConfigFactory {
    private static IConfigCenter configCenter;

    private EtcdConfigFactory() {}

    public static IConfigCenter configCenter() {
        return configCenter;
    }

    public static void buildConfigCenter(String etcdServer) {
        //连接多个时，逗号分隔
        configCenter = JdEtcdBuilder.build(etcdServer);
    }
}

//开始定时推送
PushSchedulerStarter.startPusher(pushPeriod);//每0.5秒推送一次待测key
PushSchedulerStarter.startCountPusher(10);//每10秒推送一次数量统计,不可配置

/**
     * 每0.5秒推送一次待测key
     */
    public static void startPusher(Long period) {
        if (period == null || period <= 0) {
            period = 500L;
        }
        @SuppressWarnings("PMD.ThreadPoolCreationRule")
        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("hotkey-pusher-service-executor", true));
        scheduledExecutorService.scheduleAtFixedRate(() -> {
            //热key的收集器
            IKeyCollector<HotKeyModel, HotKeyModel> collectHK = KeyHandlerFactory.getCollector();
            //这里相当于每0.5秒，通过netty来给worker来推送收集到的热key的信息，主要是一些热key的元数据信息(热key来源的app和key的类型和是否是删除事件，还有该热key的上报次数)
            //这里面还有就是该热key在每次上报的时候都会生成一个全局的唯一id，还有该热key每次上报的创建时间是在netty发送的时候来生成，同一批次的热key时间是相同的
            List<HotKeyModel> hotKeyModels = collectHK.lockAndGetResult();
            if(CollectionUtil.isNotEmpty(hotKeyModels)){
                //积攒了半秒的key集合，按照hash分发到不同的worker
                KeyHandlerFactory.getPusher().send(Context.APP_NAME, hotKeyModels);
                collectHK.finishOnce();
            }

        },0, period, TimeUnit.MILLISECONDS);
    }
    /**
     * 每10秒推送一次数量统计
     */
    public static void startCountPusher(Integer period) {
        if (period == null || period <= 0) {
            period = 10;
        }
        @SuppressWarnings("PMD.ThreadPoolCreationRule")
        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("hotkey-count-pusher-service-executor", true));
        scheduledExecutorService.scheduleAtFixedRate(() -> {
            IKeyCollector<KeyHotModel, KeyCountModel> collectHK = KeyHandlerFactory.getCounter();
            List<KeyCountModel> keyCountModels = collectHK.lockAndGetResult();
            if(CollectionUtil.isNotEmpty(keyCountModels)){
                //积攒了10秒的数量，按照hash分发到不同的worker
                KeyHandlerFactory.getPusher().sendCount(Context.APP_NAME, keyCountModels);
                collectHK.finishOnce();
            }
        },0, period, TimeUnit.SECONDS);
    }

public class DefaultKeyHandler {
    //推送HotKeyMsg消息到Netty的推送者
    private IKeyPusher iKeyPusher = new NettyKeyPusher();
    //待测key的收集器，这里面包含两个map，key主要是热key的名字,value主要是热key的元数据信息(比如:热key来源的app和key的类型和是否是删除事件)
    private IKeyCollector<HotKeyModel, HotKeyModel> iKeyCollector = new TurnKeyCollector();
    //数量收集器,这里面包含两个map，这里面key是相应的规则,HitCount里面是这个规则的总访问次数和热后访问次数
    private IKeyCollector<KeyHotModel, KeyCountModel> iKeyCounter = new TurnCountCollector();

    public IKeyPusher keyPusher() {
        return iKeyPusher;
    }
    public IKeyCollector<HotKeyModel, HotKeyModel> keyCollector() {
        return iKeyCollector;
    }
    public IKeyCollector<KeyHotModel, KeyCountModel> keyCounter() {
        return iKeyCounter;
    }
}

/**
 * 将msg推送到netty的pusher
 * @author wuweifeng wrote on 2020-01-06
 * @version 1.0
 */
public class NettyKeyPusher implements IKeyPusher {
    @Override
    public void send(String appName, List<HotKeyModel> list) {
        //积攒了半秒的key集合，按照hash分发到不同的worker
        long now = System.currentTimeMillis();

        Map<Channel, List<HotKeyModel>> map = new HashMap<>();
        for(HotKeyModel model : list) {
            model.setCreateTime(now);
            Channel channel = WorkerInfoHolder.chooseChannel(model.getKey());
            if (channel == null) {
                continue;
            }
            List<HotKeyModel> newList = map.computeIfAbsent(channel, k -> new ArrayList<>());
            newList.add(model);
        }

        for (Channel channel : map.keySet()) {
            try {
                List<HotKeyModel> batch = map.get(channel);
                HotKeyMsg hotKeyMsg = new HotKeyMsg(MessageType.REQUEST_NEW_KEY, Context.APP_NAME);
                hotKeyMsg.setHotKeyModels(batch);
                channel.writeAndFlush(hotKeyMsg).sync();
            } catch (Exception e) {
                try {
                    InetSocketAddress insocket = (InetSocketAddress) channel.remoteAddress();
                    JdLogger.error(getClass(),"flush error " + insocket.getAddress().getHostAddress());
                } catch (Exception ex) {
                    JdLogger.error(getClass(),"flush error");
                }
            }
        }
    }
    @Override
    public void sendCount(String appName, List<KeyCountModel> list) {
        //积攒了10秒的数量，按照hash分发到不同的worker
        long now = System.currentTimeMillis();
        Map<Channel, List<KeyCountModel>> map = new HashMap<>();
        for(KeyCountModel model : list) {
            model.setCreateTime(now);
            Channel channel = WorkerInfoHolder.chooseChannel(model.getRuleKey());
            if (channel == null) {
                continue;
            }
            List<KeyCountModel> newList = map.computeIfAbsent(channel, k -> new ArrayList<>());
            newList.add(model);
        }
        for (Channel channel : map.keySet()) {
            try {
                List<KeyCountModel> batch = map.get(channel);
                HotKeyMsg hotKeyMsg = new HotKeyMsg(MessageType.REQUEST_HIT_COUNT, Context.APP_NAME);
                hotKeyMsg.setKeyCountModels(batch);
                channel.writeAndFlush(hotKeyMsg).sync();
            } catch (Exception e) {
                try {
                    InetSocketAddress insocket = (InetSocketAddress) channel.remoteAddress();
                    JdLogger.error(getClass(),"flush error " + insocket.getAddress().getHostAddress());
                } catch (Exception ex) {
                    JdLogger.error(getClass(),"flush error");
                }
            }
        }
    }
}

/**
 * 对hotkey进行聚合
 * @author wuweifeng wrote on 2020-01-06
 * @version 1.0
 */
public interface IKeyCollector<T, V> {
    /**
     * 锁定后的返回值
     */
    List<V> lockAndGetResult();
    /**
     * 输入的参数
     */
    void collect(T t);

    void finishOnce();
}

public class TurnKeyCollector implements IKeyCollector<HotKeyModel, HotKeyModel> {
    //这map里面的key主要是热key的名字,value主要是热key的元数据信息(比如:热key来源的app和key的类型和是否是删除事件)
    private ConcurrentHashMap<String, HotKeyModel> map0 = new ConcurrentHashMap<>();
    private ConcurrentHashMap<String, HotKeyModel> map1 = new ConcurrentHashMap<>();
    private AtomicLong atomicLong = new AtomicLong(0);

    @Override
    public List<HotKeyModel> lockAndGetResult() {
        //自增后，对应的map就会停止被写入，等待被读取
        atomicLong.addAndGet(1);
        List<HotKeyModel> list;
        //可以观察这里与collect方法里面的相同位置，会发现一个是操作map0一个是操作map1，这样保证在读map的时候，不会阻塞写map，
        //两个map同时提供轮流提供读写能力，设计的很巧妙，值得学习
        if (atomicLong.get() % 2 == 0) {
            list = get(map1);
            map1.clear();
        } else {
            list = get(map0);
            map0.clear();
        }
        return list;
    }
    private List<HotKeyModel> get(ConcurrentHashMap<String, HotKeyModel> map) {
        return CollectionUtil.list(false, map.values());
    }
    @Override
    public void collect(HotKeyModel hotKeyModel) {
        String key = hotKeyModel.getKey();
        if (StrUtil.isEmpty(key)) {
            return;
        }
        if (atomicLong.get() % 2 == 0) {
            //不存在时返回null并将key-value放入，已有相同key时，返回该key对应的value，并且不覆盖
            HotKeyModel model = map0.putIfAbsent(key, hotKeyModel);
            if (model != null) {
                //增加该hotMey上报的次数
                model.add(hotKeyModel.getCount());
            }
        } else {
            HotKeyModel model = map1.putIfAbsent(key, hotKeyModel);
            if (model != null) {
                model.add(hotKeyModel.getCount());
            }
        }
    }
    @Override
    public void finishOnce() {}
}

//开启worker重连器
WorkerRetryConnector.retryConnectWorkers();
public class WorkerRetryConnector {

    /**
     * 定时去重连没连上的workers
     */
    public static void retryConnectWorkers() {
        @SuppressWarnings("PMD.ThreadPoolCreationRule")
        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("worker-retry-connector-service-executor", true));
        //开启拉取etcd的worker信息，如果拉取失败，则定时继续拉取
        scheduledExecutorService.scheduleAtFixedRate(WorkerRetryConnector::reConnectWorkers, 30, 30, TimeUnit.SECONDS);
    }

    private static void reConnectWorkers() {
        List<String> nonList = WorkerInfoHolder.getNonConnectedWorkers();
        if (nonList.size() == 0) {
            return;
        }
        JdLogger.info(WorkerRetryConnector.class, "trying to reConnect to these workers :" + nonList);
        NettyClient.getInstance().connect(nonList);//这里会触发netty连接方法channelActive
    }
}

/**
 * 保存worker的ip地址和Channel的映射关系，这是有序的。每次client发送消息时，都会根据该map的size进行hash
 * 如key-1就发送到workerHolder的第1个Channel去，key-2就发到第2个Channel去
 */
private static final List<Server> WORKER_HOLDER = new CopyOnWriteArrayList<>();

private void registEventBus() {
    //netty连接器会关注WorkerInfoChangeEvent事件
    EventBusCenter.register(new WorkerChangeSubscriber());
    //热key探测回调关注热key事件
    EventBusCenter.register(new ReceiveNewKeySubscribe());
    //Rule的变化的事件
    EventBusCenter.register(new KeyRuleHolder());
}

/**
 * 监听worker信息变动
 */
@Subscribe
public void connectAll(WorkerInfoChangeEvent event) {
    List<String> addresses = event.getAddresses();
    if (addresses == null) {
        addresses = new ArrayList<>();
    }

    WorkerInfoHolder.mergeAndConnectNew(addresses);
}
/**
 * 当client与worker的连接断开后，删除
 */
@Subscribe
public void channelInactive(ChannelInactiveEvent inactiveEvent) {
    //获取断线的channel
    Channel channel = inactiveEvent.getChannel();
    InetSocketAddress socketAddress = (InetSocketAddress) channel.remoteAddress();
    String address = socketAddress.getHostName() + ":" + socketAddress.getPort();
    JdLogger.warn(getClass(), "this channel is inactive : " + socketAddress + " trying to remove this connection");

    WorkerInfoHolder.dealChannelInactive(address);
}

private ReceiveNewKeyListener receiveNewKeyListener = new DefaultNewKeyListener();

@Subscribe
public void newKeyComing(ReceiveNewKeyEvent event) {
    HotKeyModel hotKeyModel = event.getModel();
    if (hotKeyModel == null) {
        return;
    }
    //收到新key推送
    if (receiveNewKeyListener != null) {
        receiveNewKeyListener.newKey(hotKeyModel);
    }
}

@Override
public void newKey(HotKeyModel hotKeyModel) {
    long now = System.currentTimeMillis();
    //如果key到达时已经过去1秒了，记录一下。手工删除key时，没有CreateTime
    if (hotKeyModel.getCreateTime() != 0 && Math.abs(now - hotKeyModel.getCreateTime()) > 1000) {
        JdLogger.warn(getClass(), "the key comes too late : " + hotKeyModel.getKey() + " now " +
                +now + " keyCreateAt " + hotKeyModel.getCreateTime());
    }
    if (hotKeyModel.isRemove()) {
        //如果是删除事件，就直接删除
        deleteKey(hotKeyModel.getKey());
        return;
    }
    //已经是热key了，又推过来同样的热key，做个日志记录，并刷新一下
    if (JdHotKeyStore.isHot(hotKeyModel.getKey())) {
        JdLogger.warn(getClass(), "receive repeat hot key ：" + hotKeyModel.getKey() + " at " + now);
    }
    addKey(hotKeyModel.getKey());
}
private void deleteKey(String key) {
        CacheFactory.getNonNullCache(key).delete(key);
}
private void addKey(String key) {
  ValueModel valueModel = ValueModel.defaultValue(key);
  if (valueModel == null) {
      //不符合任何规则
      deleteKey(key);
      return;
  }
  //如果原来该key已经存在了，那么value就被重置，过期时间也会被重置。如果原来不存在，就新增的热key
   JdHotKeyStore.setValueDirectly(key, valueModel);
}

/**
 * 保存超时时间和caffeine的映射，key是超时时间，value是caffeine[(String,Object)]
 */
private static final ConcurrentHashMap<Integer, LocalCache> RULE_CACHE_MAP = new ConcurrentHashMap<>();
/**
 * 这里KEY_RULES是保存etcd里面该appName所对应的所有rule
 */
private static final List<KeyRule> KEY_RULES = new ArrayList<>();

@Subscribe
public void ruleChange(KeyRuleInfoChangeEvent event) {
    JdLogger.info(getClass(), "new rules info is :" + event.getKeyRules());
    List<KeyRule> ruleList = event.getKeyRules();
    if (ruleList == null) {
        return;
    }

    putRules(ruleList);
}

/**
 * 所有的规则，如果规则的超时时间变化了，会重建caffeine
 */
public static void putRules(List<KeyRule> keyRules) {
    synchronized (KEY_RULES) {
        //如果规则为空，清空规则表
        if (CollectionUtil.isEmpty(keyRules)) {
            KEY_RULES.clear();
            RULE_CACHE_MAP.clear();
            return;
        }
        KEY_RULES.clear();
        KEY_RULES.addAll(keyRules);
        Set<Integer> durationSet = keyRules.stream().map(KeyRule::getDuration).collect(Collectors.toSet());
        for (Integer duration : RULE_CACHE_MAP.keySet()) {
            //先清除掉那些在RULE_CACHE_MAP里存的，但是rule里已没有的
            if (!durationSet.contains(duration)) {
                RULE_CACHE_MAP.remove(duration);
            }
        }
        //遍历所有的规则
        for (KeyRule keyRule : keyRules) {
            int duration = keyRule.getDuration();
            //这里如果RULE_CACHE_MAP里面没有超时时间为duration的value，则新建一个放入到RULE_CACHE_MAP里面
            //比如RULE_CACHE_MAP本来就是空的，则在这里来构建RULE_CACHE_MAP的映射关系
            //TODO 如果keyRules里面包含相同duration的keyRule，则也只会建一个key为duration,value为caffeine，其中caffeine是(string,object)
            if (RULE_CACHE_MAP.get(duration) == null) {
                LocalCache cache = CacheFactory.build(duration);
                RULE_CACHE_MAP.put(duration, cache);
            }
        }
    }
}

EtcdStarter starter = new EtcdStarter();
//与etcd相关的监听都开启
starter.start();

public void start() {
fetchWorkerInfo();
fetchRule();
startWatchRule();
//监听热key事件，只监听手工添加、删除的key
startWatchHotKey();
}

/**
 * 每隔30秒拉取worker信息
 */
private void fetchWorkerInfo() {
    ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
    //开启拉取etcd的worker信息，如果拉取失败，则定时继续拉取
    scheduledExecutorService.scheduleAtFixedRate(() -> {
        JdLogger.info(getClass(), "trying to connect to etcd and fetch worker info");
        fetch();

    }, 0, 30, TimeUnit.SECONDS);
}

private void fetchRule() {
    ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
    //开启拉取etcd的worker信息，如果拉取失败，则定时继续拉取
    scheduledExecutorService.scheduleAtFixedRate(() -> {
        JdLogger.info(getClass(), "trying to connect to etcd and fetch rule info");
        boolean success = fetchRuleFromEtcd();
        if (success) {
            //拉取已存在的热key
            fetchExistHotKey();
            //这里如果拉取规则和拉取手动配置的hotKey成功之后，则该定时执行线程停止
            scheduledExecutorService.shutdown();
        }
    }, 0, 5, TimeUnit.SECONDS);
}

/**
 * 异步监听rule规则变化
 */
private void startWatchRule() {
    ExecutorService executorService = Executors.newSingleThreadExecutor();
    executorService.submit(() -> {
        JdLogger.info(getClass(), "--- begin watch rule change ----");
        try {
            IConfigCenter configCenter = EtcdConfigFactory.configCenter();
            KvClient.WatchIterator watchIterator = configCenter.watch(ConfigConstant.rulePath + Context.APP_NAME);
            //如果有新事件，即rule的变更，就重新拉取所有的信息
            while (watchIterator.hasNext()) {
                //这句必须写，next会让他卡住，除非真的有新rule变更
                WatchUpdate watchUpdate = watchIterator.next();
                List<Event> eventList = watchUpdate.getEvents();
                JdLogger.info(getClass(), "rules info changed. begin to fetch new infos. rule change is " + eventList);

                //全量拉取rule信息
                fetchRuleFromEtcd();
            }
        } catch (Exception e) {
            JdLogger.error(getClass(), "watch err");
        }
    });
}

/**
 * 异步开始监听热key变化信息，该目录里只有手工添加的key信息
 */
private void startWatchHotKey() {
    ExecutorService executorService = Executors.newSingleThreadExecutor();
    executorService.submit(() -> {
        JdLogger.info(getClass(), "--- begin watch hotKey change ----");
        IConfigCenter configCenter = EtcdConfigFactory.configCenter();
        try {
            KvClient.WatchIterator watchIterator = configCenter.watchPrefix(ConfigConstant.hotKeyPath + Context.APP_NAME);
            //如果有新事件，即新key产生或删除
            while (watchIterator.hasNext()) {
                WatchUpdate watchUpdate = watchIterator.next();

                List<Event> eventList = watchUpdate.getEvents();
                KeyValue keyValue = eventList.get(0).getKv();
                Event.EventType eventType = eventList.get(0).getType();
                try {
                    //从这个地方可以看出,etcd给的返回是节点的全路径，而我们需要的key要去掉前缀
                    String key = keyValue.getKey().toStringUtf8().replace(ConfigConstant.hotKeyPath + Context.APP_NAME + "/", "");
                    //如果是删除key，就立刻删除
                    if (Event.EventType.DELETE == eventType) {
                        HotKeyModel model = new HotKeyModel();
                        model.setRemove(true);
                        model.setKey(key);
                        EventBusCenter.getInstance().post(new ReceiveNewKeyEvent(model));
                    } else {
                        HotKeyModel model = new HotKeyModel();
                        model.setRemove(false);
                        String value = keyValue.getValue().toStringUtf8();
                        //新增热key
                        JdLogger.info(getClass(), "etcd receive new key : " + key + " --value:" + value);
                        //如果这是一个删除指令，就什么也不干
                        //TODO 这里有个疑问，监听到worker自动探测发出的惰性删除指令，这里之间跳过了，但是本地缓存没有更新吧？
                        //TODO 所以我猜测在客户端使用判断缓存是否存在的api里面，应该会判断相关缓存的value值是否为"#[DELETE]#"删除标记
                        //解疑:这里确实只监听手工配置的hotKey，etcd的/jd/hotkeys/+$appName该地址只是手动配置hotKey，worker自动探测的hotKey是直接通过netty通道来告知client的
                        if (Constant.DEFAULT_DELETE_VALUE.equals(value)) {
                            continue;
                        }
                        //手工创建的value是时间戳
                        model.setCreateTime(Long.valueOf(keyValue.getValue().toStringUtf8()));
                        model.setKey(key);
                        EventBusCenter.getInstance().post(new ReceiveNewKeyEvent(model));
                    }
                } catch (Exception e) {
                    JdLogger.error(getClass(), "new key err ：" + keyValue);
                }

            }
        } catch (Exception e) {
            JdLogger.error(getClass(), "watch err");
        }
    });

}

public static void push(String key, KeyType keyType, int count, boolean remove) {
    if (count <= 0) {
        count = 1;
    }
    if (keyType == null) {
        keyType = KeyType.REDIS_KEY;
    }
    if (key == null) {
        return;
    }
    //这里之所以用LongAdder是为了保证多线程计数的线程安全性，虽然这里是在方法内调用的，但是在TurnKeyCollector的两个map里面，
    //存储了HotKeyModel的实例对象，这样在多个线程同时修改count的计数属性时，会存在线程安全计数不准确问题
    LongAdder adderCnt = new LongAdder();
    adderCnt.add(count);

    HotKeyModel hotKeyModel = new HotKeyModel();
    hotKeyModel.setAppName(Context.APP_NAME);
    hotKeyModel.setKeyType(keyType);
    hotKeyModel.setCount(adderCnt);
    hotKeyModel.setRemove(remove);
    hotKeyModel.setKey(key);


    if (remove) {
        //如果是删除key，就直接发到etcd去，不用做聚合。但是有点问题现在，这个删除只能删手工添加的key，不能删worker探测出来的
        //因为各个client都在监听手工添加的那个path，没监听自动探测的path。所以如果手工的那个path下，没有该key，那么是删除不了的。
        //删不了，就达不到集群监听删除事件的效果，怎么办呢？可以通过新增的方式，新增一个热key，然后删除它
        //TODO 这里为啥不直接删除该节点，难道worker自动探测处理的hotKey不会往该节点增加新增事件吗？
        //释疑:worker根据探测配置的规则，当判断出某个key为hotKey后，确实不会往keyPath里面加入节点，他只是单纯的往本地缓存里面加入一个空值，代表是热点key
        EtcdConfigFactory.configCenter().putAndGrant(HotKeyPathTool.keyPath(hotKeyModel), Constant.DEFAULT_DELETE_VALUE, 1);
        EtcdConfigFactory.configCenter().delete(HotKeyPathTool.keyPath(hotKeyModel));//TODO 这里很巧妙待补充描述
        //也删worker探测的目录
        EtcdConfigFactory.configCenter().delete(HotKeyPathTool.keyRecordPath(hotKeyModel));
    } else {
        //如果key是规则内的要被探测的key，就积累等待传送
        if (KeyRuleHolder.isKeyInRule(key)) {
            //积攒起来，等待每半秒发送一次
            KeyHandlerFactory.getCollector().collect(hotKeyModel);
        }
    }
}

public class NettyClient {
    private static final NettyClient nettyClient = new NettyClient();

    private Bootstrap bootstrap;

    public static NettyClient getInstance() {
        return nettyClient;
    }

    private NettyClient() {
        if (bootstrap == null) {
            bootstrap = initBootstrap();
        }
    }

    private Bootstrap initBootstrap() {
        //少线程
        EventLoopGroup group = new NioEventLoopGroup(2);

        Bootstrap bootstrap = new Bootstrap();
        NettyClientHandler nettyClientHandler = new NettyClientHandler();
        bootstrap.group(group).channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ByteBuf delimiter = Unpooled.copiedBuffer(Constant.DELIMITER.getBytes());
                        ch.pipeline()
                                .addLast(new DelimiterBasedFrameDecoder(Constant.MAX_LENGTH, delimiter))//这里就是定义TCP多个包之间的分隔符，为了更好的做拆包
                                .addLast(new MsgDecoder())
                                .addLast(new MsgEncoder())
                                //30秒没消息时，就发心跳包过去
                                .addLast(new IdleStateHandler(0, 0, 30))
                                .addLast(nettyClientHandler);
                    }
                });
        return bootstrap;
    }
}

@ChannelHandler.Sharable
public class NettyClientHandler extends SimpleChannelInboundHandler<HotKeyMsg> {
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
            //这里表示如果读写都挂了
            if (idleStateEvent.state() == IdleState.ALL_IDLE) {
                //向服务端发送消息
                ctx.writeAndFlush(new HotKeyMsg(MessageType.PING, Context.APP_NAME));
            }
        }

        super.userEventTriggered(ctx, evt);
    }
    //在Channel注册EventLoop、绑定SocketAddress和连接ChannelFuture的时候都有可能会触发ChannelInboundHandler的channelActive方法的调用
    //类似TCP三次握手成功之后触发
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        JdLogger.info(getClass(), "channelActive:" + ctx.name());
        ctx.writeAndFlush(new HotKeyMsg(MessageType.APP_NAME, Context.APP_NAME));
    }
    //类似TCP四次挥手之后，等待2MSL时间之后触发(大概180s)，比如channel通道关闭会触发(channel.close())

    //客户端channel主动关闭连接时，会向服务端发送一个写请求，然后服务端channel所在的selector会监听到一个OP_READ事件，然后
    //执行数据读取操作，而读取时发现客户端channel已经关闭了，则读取数据字节个数返回-1，然后执行close操作，关闭该channel对应的底层socket，
    //并在pipeline中，从head开始，往下将InboundHandler，并触发handler的channelInactive和channelUnregistered方法的执行，以及移除pipeline中的handlers一系列操作。
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        //断线了，可能只是client和server断了，但都和etcd没断。也可能是client自己断网了，也可能是server断了
        //发布断线事件。后续10秒后进行重连，根据etcd里的worker信息来决定是否重连，如果etcd里没了，就不重连。如果etcd里有，就重连
        notifyWorkerChange(ctx.channel());
    }
    private void notifyWorkerChange(Channel channel) {
        EventBusCenter.getInstance().post(new ChannelInactiveEvent(channel));
    }
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, HotKeyMsg msg) {
        if (MessageType.PONG == msg.getMessageType()) {
            JdLogger.info(getClass(), "heart beat");
            return;
        }
        if (MessageType.RESPONSE_NEW_KEY == msg.getMessageType()) {
            JdLogger.info(getClass(), "receive new key : " + msg);
            if (CollectionUtil.isEmpty(msg.getHotKeyModels())) {
                return;
            }
            for (HotKeyModel model : msg.getHotKeyModels()) {
                EventBusCenter.getInstance().post(new ReceiveNewKeyEvent(model));
            }
        }
    }
}

@PostConstruct
public void watchLog() {
    AsyncPool.asyncDo(() -> {
        try {
            //取etcd的是否开启日志配置，地址/jd/logOn
            String loggerOn = configCenter.get(ConfigConstant.logToggle);
            LOGGER_ON = "true".equals(loggerOn) || "1".equals(loggerOn);
        } catch (StatusRuntimeException ex) {
            logger.error(ETCD_DOWN);
        }
        //监听etcd地址/jd/logOn是否开启日志配置，并实时更改开关
        KvClient.WatchIterator watchIterator = configCenter.watch(ConfigConstant.logToggle);
        while (watchIterator.hasNext()) {
            WatchUpdate watchUpdate = watchIterator.next();
            List<Event> eventList = watchUpdate.getEvents();
            KeyValue keyValue = eventList.get(0).getKv();
            logger.info("log toggle changed : " + keyValue);
            String value = keyValue.getValue().toStringUtf8();
            LOGGER_ON = "true".equals(value) || "1".equals(value);
        }
    });
}

/**
 * 启动回调监听器，监听rule变化
 */
@PostConstruct
public void watch() {
    AsyncPool.asyncDo(() -> {
        KvClient.WatchIterator watchIterator;
        if (isForSingle()) {
            watchIterator = configCenter.watch(ConfigConstant.rulePath + workerPath);
        } else {           
            watchIterator = configCenter.watchPrefix(ConfigConstant.rulePath);
        }
        while (watchIterator.hasNext()) {
            WatchUpdate watchUpdate = watchIterator.next();
            List<Event> eventList = watchUpdate.getEvents();
            KeyValue keyValue = eventList.get(0).getKv();
            logger.info("rule changed : " + keyValue);
            try {
                ruleChange(keyValue);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    });
}
/**
     * rule发生变化时，更新缓存的rule
     */
    private synchronized void ruleChange(KeyValue keyValue) {
        String appName = keyValue.getKey().toStringUtf8().replace(ConfigConstant.rulePath, "");
        if (StrUtil.isEmpty(appName)) {
            return;
        }
        String ruleJson = keyValue.getValue().toStringUtf8();
        List<KeyRule> keyRules = FastJsonUtils.toList(ruleJson, KeyRule.class);
        KeyRuleHolder.put(appName, keyRules);
    }

/**
 * 启动回调监听器，监听白名单变化，只监听自己所在的app，白名单key不参与热key计算，直接忽略
 */
@PostConstruct
public void watchWhiteList() {
    AsyncPool.asyncDo(() -> {
        //从etcd配置中获取所有白名单
        fetchWhite();
        KvClient.WatchIterator watchIterator = configCenter.watch(ConfigConstant.whiteListPath + workerPath);
        while (watchIterator.hasNext()) {
            WatchUpdate watchUpdate = watchIterator.next();
            logger.info("whiteList changed ");
            try {
                fetchWhite();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    });
}

/**
 * 每隔一会去check一下，自己还在不在etcd里
 */
@PostConstruct
public void makeSureSelfOn() {
    //开启上传worker信息
    ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
    scheduledExecutorService.scheduleAtFixedRate(() -> {
        try {
            if (canUpload) {
                uploadSelfInfo();
            }
        } catch (Exception e) {
            //do nothing
        }
    }, 0, 5, TimeUnit.SECONDS);
}

@Component
public class DashboardPusher implements IPusher {
    /**
     * 热key集中营
     */
    private static LinkedBlockingQueue<HotKeyModel> hotKeyStoreQueue = new LinkedBlockingQueue<>();

    @PostConstruct
    public void uploadToDashboard() {
        AsyncPool.asyncDo(() -> {
            while (true) {
                try {
                    //要么key达到1千个，要么达到1秒，就汇总上报给etcd一次
                    List<HotKeyModel> tempModels = new ArrayList<>();
                    Queues.drain(hotKeyStoreQueue, tempModels, 1000, 1, TimeUnit.SECONDS);
                    if (CollectionUtil.isEmpty(tempModels)) {
                        continue;
                    }

                    //将热key推到dashboard
                    DashboardHolder.flushToDashboard(FastJsonUtils.convertObjectToJSON(tempModels));
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }
}

public class AppServerPusher implements IPusher {
    /**
     * 热key集中营
     */
    private static LinkedBlockingQueue<HotKeyModel> hotKeyStoreQueue = new LinkedBlockingQueue<>();

    /**
     * 和dashboard那边的推送主要区别在于，给app推送每10ms一次，dashboard那边1s一次
     */
    @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 = new HashMap<>();
                    //拆分出每个app的热key集合，按app分堆
                    for (HotKeyModel hotKeyModel : tempModels) {
                        List<HotKeyModel> oneAppModels = allAppHotKeyModels.computeIfAbsent(hotKeyModel.getAppName(), (key) -> new ArrayList<>());
                        oneAppModels.add(hotKeyModel);
                    }
                    //遍历所有app，进行推送
                    for (AppInfo appInfo : ClientInfoHolder.apps) {
                        List<HotKeyModel> list = allAppHotKeyModels.get(appInfo.getAppName());
                        if (CollectionUtil.isEmpty(list)) {
                            continue;
                        }
                        HotKeyMsg hotKeyMsg = new HotKeyMsg(MessageType.RESPONSE_NEW_KEY);
                        hotKeyMsg.setHotKeyModels(list);

                        //整个app全部发送
                        appInfo.groupPush(hotKeyMsg);
                    }
                    //推送完，及时清理不使用内存
                    allAppHotKeyModels = null;
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }
}

public class NodesServerStarter {
    @Value("${netty.port}")
    private int port;
    private Logger logger = LoggerFactory.getLogger(getClass());
    @Resource
    private IClientChangeListener iClientChangeListener;
    @Resource
    private List<INettyMsgFilter> messageFilters;

    @PostConstruct
    public void start() {
        AsyncPool.asyncDo(() -> {
            logger.info("netty server is starting");
            NodesServer nodesServer = new NodesServer();
            nodesServer.setClientChangeListener(iClientChangeListener);
            nodesServer.setMessageFilters(messageFilters);
            try {
                nodesServer.startNettyServer(port);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }
}

public interface IClientChangeListener {
    /**
     * 发现新连接
     */
    void newClient(String appName, String channelId, ChannelHandlerContext ctx);
    /**
     * 客户端掉线
     */
    void loseClient(ChannelHandlerContext ctx);
}

/**
 * 对netty来的消息，进行过滤处理
 * @author wuweifeng wrote on 2019-12-11
 * @version 1.0
 */
public interface INettyMsgFilter {
    boolean chain(HotKeyMsg message, ChannelHandlerContext ctx);
}

APP_NAME((byte) 1),
REQUEST_NEW_KEY((byte) 2),
RESPONSE_NEW_KEY((byte) 3),
REQUEST_HIT_COUNT((byte) 7), //命中率
REQUEST_HOT_KEY((byte) 8), //热key，worker->dashboard
PING((byte) 4), PONG((byte) 5),
EMPTY((byte) 6);

/**
 * 每隔1分钟拉取一次，所有的app的rule
 */
@Scheduled(fixedRate = 60000)
public void pullRules() {
    try {
        if (isForSingle()) {
            String value = configCenter.get(ConfigConstant.rulePath + workerPath);
            if (!StrUtil.isEmpty(value)) {
                List<KeyRule> keyRules = FastJsonUtils.toList(value, KeyRule.class);
                KeyRuleHolder.put(workerPath, keyRules);
            }
        } else {
            List<KeyValue> keyValues = configCenter.getPrefix(ConfigConstant.rulePath);
            for (KeyValue keyValue : keyValues) {
                ruleChange(keyValue);
            }
        }
    } catch (StatusRuntimeException ex) {
        logger.error(ETCD_DOWN);
    }
}

/**
     * 每隔10秒上传一下client的数量到etcd中
     */
    @Scheduled(fixedRate = 10000)
    public void uploadClientCount() {
        try {
            String ip = IpUtils.getIp();
            for (AppInfo appInfo : ClientInfoHolder.apps) {
                String appName = appInfo.getAppName();
                int count = appInfo.size();
                //即便是full gc也不能超过3秒，因为这里给的过期时间是13s，由于该定时任务每隔10s执行一次，如果full gc或者说上报给etcd的时间超过3s，
                //则在dashboard查询不到client的数量
                configCenter.putAndGrant(ConfigConstant.clientCountPath + appName + "/" + ip, count + "", 13);
            }
            configCenter.putAndGrant(ConfigConstant.caffeineSizePath + ip, FastJsonUtils.convertObjectToJSON(CaffeineCacheHolder.getSize()), 13);
            //上报每秒QPS（接收key数量、处理key数量）
            String totalCount = FastJsonUtils.convertObjectToJSON(new TotalCount(HotKeyFilter.totalReceiveKeyCount.get(), totalDealCount.longValue()));
            configCenter.putAndGrant(ConfigConstant.totalReceiveKeyCount + ip, totalCount, 13);
            logger.info(totalCount + " expireCount:" + expireTotalCount + " offerCount:" + totalOfferCount);
            //如果是稳定一直有key发送的应用，建议开启该监控，以避免可能发生的网络故障
            if (openMonitor) {
                checkReceiveKeyCount();
            }
//            configCenter.putAndGrant(ConfigConstant.bufferPoolPath + ip, MemoryTool.getBufferPool() + "", 10);
        } catch (Exception ex) {
            logger.error(ETCD_DOWN);
        }
    }

/**
 * 每隔30秒去获取一下dashboard的地址
 */
@Scheduled(fixedRate = 30000)
public void fetchDashboardIp() {
    try {
        //获取DashboardIp
        List<KeyValue> keyValues = configCenter.getPrefix(ConfigConstant.dashboardPath);
        //是空，给个警告
        if (CollectionUtil.isEmpty(keyValues)) {
            logger.warn("very important warn !!! Dashboard ip is null!!!");
            return;
        }
        String dashboardIp = keyValues.get(0).getValue().toStringUtf8();
        NettyClient.getInstance().connect(dashboardIp);
    } catch (Exception e) {
        e.printStackTrace();
    }
}

@Component
public class KeyProducer {
    public void push(HotKeyModel model, long now) {
        if (model == null || model.getKey() == null) {
            return;
        }
        //5秒前的过时消息就不处理了
        if (now - model.getCreateTime() > InitConstant.timeOut) {
            expireTotalCount.increment();
            return;
        }
        try {
            QUEUE.put(model);
            totalOfferCount.increment();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

public class KeyConsumer {
    private IKeyListener iKeyListener;
    public void setKeyListener(IKeyListener iKeyListener) {
        this.iKeyListener = iKeyListener;
    }
    public void beginConsume() {
        while (true) {
            try {
                //从这里可以看出，这里的生产者消费者模型，本质上还是拉模式，之所以不使用EventBus，是因为需要队列来做缓冲
                HotKeyModel model = QUEUE.take();
                if (model.isRemove()) {
                    iKeyListener.removeKey(model, KeyEventOriginal.CLIENT);
                } else {
                    iKeyListener.newKey(model, KeyEventOriginal.CLIENT);
                }
                //处理完毕，将数量加1
                totalDealCount.increment();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
@Override
public void removeKey(HotKeyModel hotKeyModel, KeyEventOriginal original) {
   //cache里的key,appName+keyType+key
   String key = buildKey(hotKeyModel);
   hotCache.invalidate(key);
   CaffeineCacheHolder.getCache(hotKeyModel.getAppName()).invalidate(key);
   //推送所有client删除
   hotKeyModel.setCreateTime(SystemClock.now());
   logger.info(DELETE_KEY_EVENT + hotKeyModel.getKey());
   for (IPusher pusher : iPushers) {
       //这里可以看到，删除热key的netty消息只给client端发了过去，没有给dashboard发过去(DashboardPusher里面的remove是个空方法)
       pusher.remove(hotKeyModel);
   }
}
    @Override
    public void newKey(HotKeyModel hotKeyModel, KeyEventOriginal original) {
        //cache里的key
        String key = buildKey(hotKeyModel);
        //判断是不是刚热不久
        //hotCache对应的caffeine有效期为5s，也就是说该key会保存5s，在5s内不重复处理相同的hotKey。
        //毕竟hotKey都是瞬时流量，可以避免在这5s内重复推送给client和dashboard，避免无效的网络开销
        Object o = hotCache.getIfPresent(key);
        if (o != null) {
            return;
        }

        //********** watch here ************//
        //该方法会被InitConstant.threadCount个线程同时调用，存在多线程问题
        //下面的那句addCount是加了锁的，代表给Key累加数量时是原子性的，不会发生多加、少加的情况，到了设定的阈值一定会hot
        //譬如阈值是2，如果多个线程累加，在没hot前，hot的状态肯定是对的，譬如thread1 加1，thread2加1，那么thread2会hot返回true，开启推送
        //但是极端情况下，譬如阈值是10，当前是9，thread1走到这里时，加1，返回true，thread2也走到这里，加1，此时是11，返回true，问题来了
        //该key会走下面的else两次，也就是2次推送。
        //所以出现问题的原因是hotCache.getIfPresent(key)这一句在并发情况下，没return掉，放了两个key+1到addCount这一步时，会有问题
        //测试代码在TestBlockQueue类，直接运行可以看到会同时hot

        //那么该问题用解决吗，NO，不需要解决，1 首先要发生的条件极其苛刻，很难触发，以京东这样高的并发量，线上我也没见过触发连续2次推送同一个key的
        //2 即便触发了，后果也是可以接受的，2次推送而已，毫无影响，客户端无感知。但是如果非要解决，就要对slidingWindow实例加锁了，必然有一些开销

        //所以只要保证key数量不多计算就可以，少计算了没事。因为热key必然频率高，漏计几次没事。但非热key，多计算了，被干成了热key就不对了
        SlidingWindow slidingWindow = checkWindow(hotKeyModel, key);//从这里可知，每个app的每个key都会对应一个滑动窗口
        //看看hot没
        boolean hot = slidingWindow.addCount(hotKeyModel.getCount());

        if (!hot) {
            //如果没hot，重新put，cache会自动刷新过期时间
            CaffeineCacheHolder.getCache(hotKeyModel.getAppName()).put(key, slidingWindow);
        } else {
            //这里之所以放入的value为1，是因为hotCache是用来专门存储刚生成的hotKey
            //hotCache对应的caffeine有效期为5s，也就是说该key会保存5s，在5s内不重复处理相同的hotKey。
            //毕竟hotKey都是瞬时流量，可以避免在这5s内重复推送给client和dashboard，避免无效的网络开销
            hotCache.put(key, 1);

            //删掉该key
            //这个key从实际上是专门针对slidingWindow的key，他的组合逻辑是appName+keyType+key，而不是给client和dashboard推送的hotKey
            CaffeineCacheHolder.getCache(hotKeyModel.getAppName()).invalidate(key);

            //开启推送
            hotKeyModel.setCreateTime(SystemClock.now());

            //当开关打开时，打印日志。大促时关闭日志，就不打印了
            if (EtcdStarter.LOGGER_ON) {
                logger.info(NEW_KEY_EVENT + hotKeyModel.getKey());
            }

            //分别推送到各client和etcd
            for (IPusher pusher : iPushers) {
                pusher.push(hotKeyModel);
            }

        }

    }