先试试

  1  synchronized (getUpdateLock()) {
  2       cmdExecutor.ensureExists(CLUSTER_STATE, zkClient);
  3       
  4       log.info("Updating cluster state from ZooKeeper... ");
  5       
  6       zkClient.exists(CLUSTER_STATE, new Watcher() {
  7         
  8         @Override
  9         public void process(WatchedEvent event) {
 10           // session events are not change events,
 11           // and do not remove the watcher
 12           if (EventType.None.equals(event.getType())) {
 13             return;
 14           }
 15           log.info("A cluster state change has occurred - updating...");
 16           try {
 17             
 18             // delayed approach
 19             // ZkStateReader.this.updateClusterState(false, false);
 20             synchronized (ZkStateReader.this.getUpdateLock()) {
 21               // remake watch
 22               final Watcher thisWatch = this;
 23               Stat stat = new Stat();
 24               byte[] data = zkClient.getData(CLUSTER_STATE, thisWatch, stat ,
 25                   true);
 26               
 27 // 将zk上的信息实例化成ClusterState对象
 28               ClusterState clusterState = ClusterState.load(stat.getVersion(), data,
 29                   ZkStateReader.this.clusterState.getLiveNodes());
 30               // update volatile 更新volatile变量
 31               ZkStateReader.this.clusterState = clusterState;
 32             }
 33           } catch (KeeperException e) {
 34             if (e.code() == KeeperException.Code.SESSIONEXPIRED
 35                 || e.code() == KeeperException.Code.CONNECTIONLOSS) {
 36               log.warn("ZooKeeper watch triggered, but Solr cannot talk to ZK");
 37               return;
 38             }
 39             log.error("", e);
 40             throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR,
 41                 "", e);
 42           } catch (InterruptedException e) {
 43             // Restore the interrupted status
 44             Thread.currentThread().interrupt();
 45             log.warn("", e);
 46             return;
 47           }
 48         }
 49         
 50       }, true);
 51     }
 52    
 53 
 54 
 55 再获得一次livenodes 重新加载ClusterState 
 56     synchronized (ZkStateReader.this.getUpdateLock()) {
 57       List<String> liveNodes = zkClient.getChildren(LIVE_NODES_ZKNODE,
 58           new Watcher() {
 59             
 60             @Override
 61             public void process(WatchedEvent event) {
 62               // session events are not change events,
 63               // and do not remove the watcher
 64               if (EventType.None.equals(event.getType())) {
 65                 return;
 66               }
 67               log.info("Updating live nodes");
 68               try {
 69                 // delayed approach
 70                 // ZkStateReader.this.updateClusterState(false, true);
 71                 synchronized (ZkStateReader.this.getUpdateLock()) {
 72                   List<String> liveNodes = zkClient.getChildren(
 73                       LIVE_NODES_ZKNODE, this, true);
 74                   Set<String> liveNodesSet = new HashSet<String>();
 75                   liveNodesSet.addAll(liveNodes);
 76                   ClusterState clusterState = new ClusterState(
 77                       ZkStateReader.this.clusterState.getZkClusterStateVersion(),
 78                       liveNodesSet, ZkStateReader.this.clusterState
 79                           .getCollectionStates());
 80                   ZkStateReader.this.clusterState = clusterState;
 81                 }
 82               } catch (KeeperException e) {
 83                 if (e.code() == KeeperException.Code.SESSIONEXPIRED
 84                     || e.code() == KeeperException.Code.CONNECTIONLOSS) {
 85                   log.warn("ZooKeeper watch triggered, but Solr cannot talk to ZK");
 86                   return;
 87                 }
 88                 log.error("", e);
 89                 throw new ZooKeeperException(
 90                     SolrException.ErrorCode.SERVER_ERROR, "", e);
 91               } catch (InterruptedException e) {
 92                 // Restore the interrupted status
 93                 Thread.currentThread().interrupt();
 94                 log.warn("", e);
 95                 return;
 96               }
 97             }
 98             
 99           }, true);
100     
101       Set<String> liveNodeSet = new HashSet<String>();
102       liveNodeSet.addAll(liveNodes);
103       ClusterState clusterState = ClusterState.load(zkClient, liveNodeSet);
104       this.clusterState = clusterState;
105     }
106 
107 
108 
109 
110 
111 
112   if (!sendToLeaders || (sendToLeaders && coreNodeProps.isLeader())) {
113               String url = coreNodeProps.getCoreUrl();
114               urlList.add(url);
115             } else if (sendToLeaders) {
116               String url = coreNodeProps.getCoreUrl();
117               replicas.add(url);
118             }
119 
120 
121 将leader 和replication都追加到thUelList里
122  theUrlList.addAll(urlList);
123 
124 
125  theUrlList.addAll(theReplicas);
126 
127 
128    LBHttpSolrServer.Req req = new LBHttpSolrServer.Req(request, theUrlList);
129     LBHttpSolrServer.Rsp rsp = lbServer.request(req);