ZkRegister的主要作用： 1. 根据特定的interface从zk中取出与之相关的所有provider的host并且监听 provider的变化，当发生变化时更新最新的hosts到本地缓存； 2. 当从zk中获取指定的interface的provider的时候，把当前进程作为此interface 的一个consumer注册到zk中，并设置此节点的状态为ephemeral；
| 88 | |
| 89 | |
| 90 | class ZkRegister(object): |
| 91 | """ |
| 92 | ZkRegister的主要作用： |
| 93 | 1. 根据特定的interface从zk中取出与之相关的所有provider的host并且监听 |
| 94 | provider的变化，当发生变化时更新最新的hosts到本地缓存； |
| 95 | 2. 当从zk中获取指定的interface的provider的时候，把当前进程作为此interface |
| 96 | 的一个consumer注册到zk中，并设置此节点的状态为ephemeral； |
| 97 | """ |
| 98 | |
| 99 | def __init__(self, hosts, application_name='search_platform'): |
| 100 | """ |
| 101 | :param hosts: Zookeeper的地址 |
| 102 | :param application_name: 当前客户端的名称 |
| 103 | """ |
| 104 | zk = KazooClient(hosts=hosts) |
| 105 | # 对zookeeper连接状态的监控 |
| 106 | zk.add_listener(self.state_listener) |
| 107 | zk.start() |
| 108 | |
| 109 | self.zk = zk |
| 110 | self.hosts = {} |
| 111 | self.weights = {} |
| 112 | self.application_name = application_name |
| 113 | self.lock = threading.Lock() |
| 114 | |
| 115 | @staticmethod |
| 116 | def state_listener(state): |
| 117 | logger.debug('Current state -> {}'.format(state)) |
| 118 | if state == KazooState.LOST: |
| 119 | logger.debug('The session to register has lost.') |
| 120 | elif state == KazooState.SUSPENDED: |
| 121 | logger.debug('Disconnected from zookeeper.') |
| 122 | else: |
| 123 | logger.debug('Connected or disconnected to zookeeper.') |
| 124 | |
| 125 | def get_provider_host(self, interface): |
| 126 | """ |
| 127 | 从zk中可以根据接口名称获取到此接口某个provider的host |
| 128 | :param interface: |
| 129 | :return: |
| 130 | """ |
| 131 | if interface not in self.hosts: |
| 132 | self.lock.acquire() |
| 133 | try: |
| 134 | if interface not in self.hosts: |
| 135 | path = DUBBO_ZK_PROVIDERS.format(interface) |
| 136 | if self.zk.exists(path): |
| 137 | self._get_providers_from_zk(path, interface) |
| 138 | self._get_configurators_from_zk(interface) |
| 139 | else: |
| 140 | raise RegisterException('No providers for interface {0}'.format(interface)) |
| 141 | finally: |
| 142 | self.lock.release() |
| 143 | return self._routing_with_wight(interface) |
| 144 | |
| 145 | def _get_providers_from_zk(self, path, interface): |
| 146 | """ |
| 147 | 从zk中根据interface获取到providers信息 |
no outgoing calls