public synchronized void start() {
       if (future == null) {
           Runnable task = new Runnable() {
                               public void run() {
                                   try {
                                     // 主要做计算统计数据
                                      accumulator.publish();
                                   } catch (Exception e) {
                                       handleException(e);
                                   }
                               }
                           };
           // 从上下文看，延迟1秒定期执行一次
           future = getExecutor().scheduleWithFixedDelay(task,
                                                         delayMillis, delayMillis,
                                                         TimeUnit.MILLISECONDS);
       }
   }
   protected synchronized ScheduledExecutorService getExecutor() {
       if (sharedExecutor == null) {
           sharedExecutor = Executors.newScheduledThreadPool(1, new PublishThreadFactory());
       }
       return sharedExecutor;
   }

public void initialize(Server server) {
       serverFailureCounts = new MeasuredRate(failureCountSlidingWindowInterval);
       requestCountInWindow = new MeasuredRate(300000L);
       if (publisher == null) {
           // DataBuffer实际存在于DataDistribution中，bufferSize=1000
           dataDist = new DataDistribution(getBufferSize(), PERCENTS);
           // DataPublisher间接引用了DataBuffer
           publisher = new DataPublisher(dataDist, getPublishIntervalMillis());
           // 触发调用
           publisher.start();
       }
       this.server = server;
   }

private ServerStats createServerStats(Server server) {
        // 创建ServerStats并初始化
        ServerStats ss = new ServerStats(this);
        ss.setBufferSize(1000);
        ss.setPublishInterval(1000);                   
        ss.initialize(server);
        return ss;        
    }
private ServerStats getServerStats(Server server) {
        try {
            // 从localcache缓存中获取，如果没有
            // 则调用createServerStats创建ServerStats并缓存
            return serverStatsCache.get(server);
        } catch (ExecutionException e) {
            ServerStats stats = createServerStats(server);
            serverStatsCache.asMap().putIfAbsent(server, stats);
            return serverStatsCache.asMap().get(server);
        }
    }
 // localcache在这里。。。，key是Server, value是ServerStats
 private final LoadingCache<Server, ServerStats> serverStatsCache = 
        CacheBuilder.newBuilder()
            .expireAfterAccess(SERVERSTATS_EXPIRE_MINUTES.get(), TimeUnit.MINUTES)
            .removalListener(new RemovalListener<Server, ServerStats>() {
                @Override
                public void onRemoval(RemovalNotification<Server, ServerStats> notification) {
                    notification.getValue().close();
                }
            })
            .build(
                new CacheLoader<Server, ServerStats>() {
                    // 缓存没有则创建ServerStats
                    public ServerStats load(Server server) {
                        return createServerStats(server);
                    }
                });

public String call(final String path, Map<String, Object> param, Integer uploadType) {
      return LoadBalancerCommand.<String> builder().withLoadBalancer(loadBalancer).build()
          .submit(new ServerOperation<String>() {
            @Override
            public Observable<String> call(Server server) {
              URL url;
              try {
                String s = HttpClientUtil.post("http://" + server.getHost() + ":" + server.getPort() + path, param);
                return Observable.just(s);
              } catch (Exception e) {
                logger.info("python identify voice failed! host {}", server.getHost());
                return Observable.error(e);
              }
            }
          }).toBlocking().first();
  }

 public Observable<T> submit(final ServerOperation<T> operation) {
        // Use the load balancer
        Observable<T> o = 
                (server == null ? selectServer() : Observable.just(server))
                .concatMap(new Func1<Server, Observable<T>>() {
                    @Override
                    // Called for each server being selected
                    public Observable<T> call(Server server) {
                        context.setServer(server);
                        // getServerStats的入口在这里。。。
                        final ServerStats stats = loadBalancerContext.getServerStats(server);
}

public final ServerStats getServerStats(Server server) {
       ServerStats serverStats = null;
       // LoadBalancerContext持有LoadBalancer对象
       ILoadBalancer lb = this.getLoadBalancer();
       if (lb instanceof AbstractLoadBalancer){
           // LoadBalancer持有LoadBalancerStats对象
           LoadBalancerStats lbStats = ((AbstractLoadBalancer) lb).getLoadBalancerStats();
           serverStats = lbStats.getSingleServerStat(server);
       }
       return serverStats;
   }

public void afterPropertiesSet() throws Exception {
    initLoadbalance();
    // 每隔3分钟重新创建Loadbalance
    service.scheduleAtFixedRate(new Runnable() {
      @Override
      public void run() {
        initLoadbalance();
      }
    }, 3, 3, TimeUnit.MINUTES);
  }
/**
   * 初始化普通机器负载均衡策略
   */
  private void initLoadbalance() {
    try {
      List<Server> serverList = new ArrayList<>();
      String machine = footballConfigFamilyService.getConfigMap().get("voice-analyze");
      List<Weight> list = JSON.parseObject(machine, new TypeReference<List<Weight>>() {
      });
      // serverList只有10来个
      for (Weight weight : list) {
        serverList.add(new WeightServer(weight.getHost(), weight.getPort(), weight.getWeight()));
      }
      // 创建loadBalancer
      loadBalancer = LoadBalancerBuilder.newBuilder().withRule(new WeightedRule())
          .buildFixedServerListLoadBalancer(serverList);
      
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

 public BaseLoadBalancer buildFixedServerListLoadBalancer(List<T> servers) {
        if (rule == null) {
            rule = createRuleFromConfig(config);
        }
        // 创建BaseLoadBalancer
        BaseLoadBalancer lb = new BaseLoadBalancer(config, rule, ping);
        // 设置负载均衡服务器列表
        lb.setServersList(servers);
        return lb;
    }
 public BaseLoadBalancer(IClientConfig config, IRule rule, IPing ping) {
        initWithConfig(config, rule, ping);
    }
void initWithConfig(IClientConfig clientConfig, IRule rule, IPing ping) {
        setLoadBalancerStats(new LoadBalancerStats(clientName));
        rule.setLoadBalancer(this);
        if (ping instanceof AbstractLoadBalancerPing) {
            ((AbstractLoadBalancerPing) ping).setLoadBalancer(this);
        }
    }

 public Server(String host, int port) {
        this.host = host;
        this.port = port;
        this.id = host + ":" + port;
        isAliveFlag = false;
    }
public boolean equals(Object obj) {
        if (this == obj)
            return true;
        if (!(obj instanceof Server))
            return false;
        Server svc = (Server) obj;
        return svc.getId().equals(this.getId());
    }
    public int hashCode() {
        int hash = 7;
        hash = 31 * hash + (null == this.getId() ? 0 : this.getId().hashCode());
        return hash;
    }

public BaseLoadBalancer(IClientConfig config, IRule rule, IPing ping) {
        initWithConfig(config, rule, ping);
    }
void initWithConfig(IClientConfig clientConfig, IRule rule, IPing ping) {
        setLoadBalancerStats(new LoadBalancerStats(clientName));
        rule.setLoadBalancer(this);
        if (ping instanceof AbstractLoadBalancerPing) {
            ((AbstractLoadBalancerPing) ping).setLoadBalancer(this);
        }
    }

public synchronized void start() {
       if (future == null) {
           Runnable task = new Runnable() {
                               public void run() {
                                   try {
                                      accumulator.publish();
                                   } catch (Exception e) {
                                       handleException(e);
                                   }
                               }
                           };
           future = getExecutor().scheduleWithFixedDelay(task,
                                                         delayMillis, delayMillis,
                                                         TimeUnit.MILLISECONDS);
       }
   }
   protected synchronized ScheduledExecutorService getExecutor() {
       if (sharedExecutor == null) {
           sharedExecutor = Executors.newScheduledThreadPool(1, new PublishThreadFactory());
       }
       return sharedExecutor;
   }

// 过期时间默认30分钟
private final LoadingCache<Server, ServerStats> serverStatsCache = 
        CacheBuilder.newBuilder()
            .expireAfterAccess(SERVERSTATS_EXPIRE_MINUTES.get(), TimeUnit.MINUTES)
            .removalListener(new RemovalListener<Server, ServerStats>() {
                @Override
                public void onRemoval(RemovalNotification<Server, ServerStats> notification) {
                    notification.getValue().close();
                }
            })
            .build(
                new CacheLoader<Server, ServerStats>() {
                    // 缓存没有则创建ServerStats
                    public ServerStats load(Server server) {
                        return createServerStats(server);
                    }
                });

private void initLoadbalance() {
    try {
      List<Server> serverList = new ArrayList<>();
      String machine = footballConfigFamilyService.getConfigMap().get("voice-analyze");
      List<Weight> list = JSON.parseObject(machine, new TypeReference<List<Weight>>() {
      });
      // serverList只有10来个
      for (Weight weight : list) {
        serverList.add(new WeightServer(weight.getHost(), weight.getPort(), weight.getWeight()));
      }
      // **********代码修复处************
      if(loadBalancer == null){
        loadBalancer = LoadBalancerBuilder.newBuilder().withRule(new WeightedRule())
          .buildFixedServerListLoadBalancer(serverList);
      }else{
        // 覆盖旧的server list
        loadBalancer.setServersList(serverList);
      }
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
          .childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch) throws Exception {
              // ...
            }
          }).option(ChannelOption.SO_BACKLOG, 128).childOption(ChannelOption.SO_KEEPALIVE, true);
ChannelFuture channelFuture = bootstrap.bind(new InetSocketAddress(port)).sync();

private ChannelFuture doBind(final SocketAddress localAddress) {
         // 创建channel并且注册selector
        final ChannelFuture regFuture = initAndRegister();
        final Channel channel = regFuture.channel();
        if (regFuture.cause() != null) {
            return regFuture;
        }
        if (regFuture.isDone()) {
            // At this point we know that the registration was complete and successful.
            ChannelPromise promise = channel.newPromise();
            // 将ServerSocketChannel绑定到本地的端口
            doBind0(regFuture, channel, localAddress, promise);
            return promise;
        } else {
           // ...
        }
    }

Channel channel = null;
try {
    // 创建channel
    channel = channelFactory.newChannel();
    // 初始化channel
    init(channel);
} catch (Throwable t) {
    if (channel != null) {
        channel.unsafe().closeForcibly();
        return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
    }
    return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t);
}
// 注册channel到selector
ChannelFuture regFuture = config().group().register(channel);

public NioServerSocketChannel() {
    //调用下面这个方法
    this(newSocket(DEFAULT_SELECTOR_PROVIDER));
}
public NioServerSocketChannel(ServerSocketChannel channel) {
    //注意，这里是初始化了一个对SelectionKey.OP_ACCEPT感兴趣的管道
    //Boss的Reactor线程轮询到都是ACCEPT事件
    super(null, channel, SelectionKey.OP_ACCEPT);
    config = new NioServerSocketChannelConfig(this, javaChannel().socket());
}
private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();
private static ServerSocketChannel newSocket(SelectorProvider provider) {
        return provider.openServerSocketChannel();
}

protected AbstractChannel(Channel parent) {
    this.parent = parent;
    // ChannelId是一个全局唯一的值
    id = newId();
    unsafe = newUnsafe();
    pipeline = newChannelPipeline();
}

void init(Channel channel) throws Exception {
        // 添加一个handle－ChannelInitializer
        p.addLast(new ChannelInitializer<Channel>() {
            @Override
            // 这里还并未执行initChannel方法，该方法是在执行回调方法handlerAdded后调用的
            public void initChannel(final Channel ch) throws Exception {
                final ChannelPipeline pipeline = ch.pipeline();
                ChannelHandler handler = config.handler();
                if (handler != null) {
                    pipeline.addLast(handler);
                }
                ch.eventLoop().execute(new Runnable() {
                    @Override
                    public void run() {
                        pipeline.addLast(new ServerBootstrapAcceptor(
                                ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                    }
                });
            }
        });
    }

ChannelFuture regFuture = config().group().register(channel);

 public ChannelFuture register(Channel channel) {
        return next().register(channel);
}

promise.channel().unsafe().register(this, promise);
return promise;

private void register0(ChannelPromise promise) {
            try {
                boolean firstRegistration = neverRegistered;// 是否为首次注册
                // 实际的注册动作, 把Channel感兴趣的事件注册到Eventloop.selector上面
                doRegister();
                neverRegistered = false;
                registered = true;
                pipeline.invokeHandlerAddedIfNeeded();// 将注册之前加入的handler加入进来
                safeSetSuccess(promise); // 注册成功，通知promise
                pipeline.fireChannelRegistered();// pipeline通知触发注册成功
      
                if (isActive()) { // 是否已经绑定 因为register和bind阶段是异步的
                    if (firstRegistration) { 
                        pipeline.fireChannelActive(); // 首次注册，通知
                    } else if (config().isAutoRead()) {
                        beginRead();
                    }
                }
            } catch (Throwable t) {
                closeForcibly();
                closeFuture.setClosed();
                safeSetFailure(promise, t);
            }
        }

public final ChannelPipeline fireChannelRegistered() {
        AbstractChannelHandlerContext.invokeChannelRegistered(head);
        return this;
    }
    
private void invokeChannelRegistered() {
        if (invokeHandler()) { // 状态是否正确
            try {
                ((ChannelInboundHandler) handler()).channelRegistered(this); // 触发
            } catch (Throwable t) {
                notifyHandlerException(t);
            }
        } else {
            fireChannelRegistered();// 状态不正确，通知下一个Handler
        }
    }

private static void doBind0(
           final ChannelFuture regFuture, final Channel channel,
           final SocketAddress localAddress, final ChannelPromise promise) {
       // This method is invoked before channelRegistered() is triggered.  Give user handlers a chance to set up
       // the pipeline in its channelRegistered() implementation.
       channel.eventLoop().execute(new Runnable() {
           @Override
           public void run() {
               if (regFuture.isSuccess()) {
                   channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
               } else {
                   promise.setFailure(regFuture.cause());
               }
           }
       });
   }

public final ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
        return tail.bind(localAddress, promise);
    }
public ChannelFuture bind(final SocketAddress localAddress, final ChannelPromise promise) {
        // 找出下一个outbound,这里找出来的是HeadContext
        final AbstractChannelHandlerContext next = findContextOutbound();
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            next.invokeBind(localAddress, promise);
        } else {
            safeExecute(executor, new Runnable() {
                @Override
                public void run() {
                    next.invokeBind(localAddress, promise);
                }
            }, promise, null);
        }
        return promise;
    }

// AbstractUnsafe.class
public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
            boolean wasActive = isActive();
            try {
              // 把channel bind到本地地址
                doBind(localAddress);
            } catch (Throwable t) {
                safeSetFailure(promise, t);
                closeIfClosed();
                return;
            }
            // bind之后，这里的isActive会返回true
            if (!wasActive && isActive()) {
                invokeLater(new Runnable() {
                    @Overridem   
                    public void run() {
                        //在这里触发的是NioServerSocketChannel对应的pipeline的channelActive
                        pipeline.fireChannelActive();
                    }
                });
            }
            safeSetSuccess(promise);
        }

protected void doBind(SocketAddress localAddress) throws Exception {
        if (PlatformDependent.javaVersion() >= 7) {
            javaChannel().bind(localAddress, config.getBacklog());
        } else {
            javaChannel().socket().bind(localAddress, config.getBacklog());
        }
    }

// DefaultChannelPipeline.class
public final ChannelPipeline fireChannelActive() {
        AbstractChannelHandlerContext.invokeChannelActive(head);
        return this;
}
static void invokeChannelActive(final AbstractChannelHandlerContext next) {
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            next.invokeChannelActive();
        } else {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    next.invokeChannelActive();
                }
            });
        }
    }
 private void invokeChannelActive() {
        if (invokeHandler()) {
            try {
              // 这里handler()返回HeadContext
                ((ChannelInboundHandler) handler()).channelActive(this);
            } catch (Throwable t) {
                notifyHandlerException(t);
            }
        } else {
            fireChannelActive();
        }
    }

public void channelActive(ChannelHandlerContext ctx) throws Exception {
            ctx.fireChannelActive();
            readIfIsAutoRead();
}

protected void doBeginRead() throws Exception {
    final SelectionKey selectionKey = this.selectionKey;
    if (!selectionKey.isValid()) {
        return;
    }
    readPending = true;
    final int interestOps = selectionKey.interestOps();
    if ((interestOps & readInterestOp) == 0) { 
        // interestOps是0，按位与结果肯定是0
        selectionKey.interestOps(interestOps | readInterestOp);
    }
}

protected void doRegister() throws Exception {
        boolean selected = false;
        for (;;) {
            try {
              // selectKey赋予了0
        selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
                return;
            } catch (CancelledKeyException e) { 
            }
        }
    }

public NioServerSocketChannel(ServerSocketChannel channel) {
       super(null, channel, SelectionKey.OP_ACCEPT);
       config = new NioServerSocketChannelConfig(this, javaChannel().socket());
   }

ChannelFuture write(Object msg, ChannelPromise promise);

private AbstractChannelHandlerContext findContextOutbound() {
        AbstractChannelHandlerContext ctx = this;
        do {
            // 注意，这里是往前找
            ctx = ctx.prev;
        } while (!ctx.outbound);
        return ctx;
    }