Zookeeper客户端kazoo的watch流程详解

@zookeeper.DataWatch(path)
def changed(data, stat):
	fun(data)
	type(stat)

def __init__(xxx):
	if func is not None:
		self._used = True
		self._client.add_listener(self._session_watcher)
		self._get_data()

def get_async(self, path, watch=None):
	"""Asynchronously get the value of a node. Takes the same
	arguments as :meth:`get`.

	:rtype: :class:`~kazoo.interfaces.IAsyncResult`

	"""
	if not isinstance(path, string_types):
		raise TypeError("Invalid type for 'path' (string expected)")
	if watch and not callable(watch):
		raise TypeError("Invalid type for 'watch' (must be a callable)")

	async_result = self.handler.async_result()
	self._call(GetData(_prefix_root(self.chroot, path), watch),
			   async_result)
	return async_result

def _read_socket(self, read_timeout):
	"""Called when there's something to read on the socket"""
	client = self.client

	header, buffer, offset = self._read_header(read_timeout)
	if header.xid == PING_XID:
		...
	elif header.xid == WATCH_XID:
		self._read_watch_event(buffer, offset)
	...

		return self._read_response(header, buffer, offset)

def _read_watch_event(self, buffer, offset):
	client = self.client
	watch, offset = Watch.deserialize(buffer, offset)
	path = watch.path

	self.logger.debug('Received EVENT: %s', watch)

	watchers = []

	if watch.type in (CREATED_EVENT, CHANGED_EVENT):
		watchers.extend(client._data_watchers.pop(path, []))
	elif watch.type == DELETED_EVENT:
		watchers.extend(client._data_watchers.pop(path, []))
		watchers.extend(client._child_watchers.pop(path, []))
	elif watch.type == CHILD_EVENT:
		watchers.extend(client._child_watchers.pop(path, []))
	else:
		self.logger.warn('Received unknown event %r', watch.type)
		return

	# Strip the chroot if needed
	path = client.unchroot(path)
	ev = WatchedEvent(EVENT_TYPE_MAP[watch.type], client._state, path)

	# Last check to ignore watches if we've been stopped
	if client._stopped.is_set():
		return

	# Dump the watchers to the watch thread
	for watch in watchers:
		client.handler.dispatch_callback(Callback('watch', watch, (ev,)))

def dispatch_callback(self, callback):
	"""Dispatch to the callback object

	The callback is put on separate queues to run depending on the
	type as documented for the :class:`SequentialThreadingHandler`.

	"""
	self.callback_queue.put(lambda: callback.func(*callback.args))

def start(self):
	"""Start the worker threads."""
	with self._state_change:
		if self._running:
			return

		# Spawn our worker threads, we have
		# - A callback worker for watch events to be called
		# - A completion worker for completion events to be called
		for queue in (self.completion_queue, self.callback_queue):
			w = self._create_thread_worker(queue)
			self._workers.append(w)
		self._running = True
		python2atexit.register(self.stop)

def _create_thread_worker(self, queue):
	def _thread_worker():  # pragma: nocover
		while True:
			try:
				func = queue.get()
				try:
					if func is _STOP:
						break
					func()
				except Exception:
					log.exception("Exception in worker queue thread")
				finally:
					queue.task_done()
			except self.queue_empty:
				continue
	t = self.spawn(_thread_worker)
	return t

def _read_response(self, header, buffer, offset):
	client = self.client
	request, async_object, xid = client._pending.popleft()
	if header.zxid and header.zxid > 0:
		client.last_zxid = header.zxid
	if header.xid != xid:
		exc = RuntimeError('xids do not match, expected %r '
						   'received %r', xid, header.xid)
		async_object.set_exception(exc)
		raise exc

	# Determine if its an exists request and a no node error
	exists_error = (header.err == NoNodeError.code and
					request.type == Exists.type)

	# Set the exception if its not an exists error
	if header.err and not exists_error:
		callback_exception = EXCEPTIONS[header.err]()
		self.logger.debug(
			'Received error(xid=%s) %r', xid, callback_exception)
		if async_object:
			async_object.set_exception(callback_exception)
	elif request and async_object:
		if exists_error:
			# It's a NoNodeError, which is fine for an exists
			# request
			async_object.set(None)
		else:
			try:
				response = request.deserialize(buffer, offset)
			except Exception as exc:
				self.logger.exception(
					"Exception raised during deserialization "
					"of request: %s", request)
				async_object.set_exception(exc)
				return
			self.logger.debug(
				'Received response(xid=%s): %r', xid, response)

			# We special case a Transaction as we have to unchroot things
			if request.type == Transaction.type:
				response = Transaction.unchroot(client, response)

			async_object.set(response)

		# Determine if watchers should be registered
		watcher = getattr(request, 'watcher', None)
		if not client._stopped.is_set() and watcher:
			if isinstance(request, (GetChildren, GetChildren2)):
				client._child_watchers[request.path].add(watcher)
			else:
				client._data_watchers[request.path].add(watcher)

	if isinstance(request, Close):
		self.logger.log(BLATHER, 'Read close response')
		return CLOSE_RESPONSE

class AsyncResult(object):
    """A one-time event that stores a value or an exception"""
    def __init__(self, handler, condition_factory, timeout_factory):
        self._handler = handler
        self._exception = _NONE
        self._condition = condition_factory()
        self._callbacks = []
        self._timeout_factory = timeout_factory
        self.value = None

    ...
	
    def set(self, value=None):
        """Store the value. Wake up the waiters."""
        with self._condition:
            self.value = value
            self._exception = None
            for callback in self._callbacks:
                self._handler.completion_queue.put(
                    functools.partial(callback, self)
                )
            self._condition.notify_all()

	...
	
    def get(self, block=True, timeout=None):
        """Return the stored value or raise the exception.

        If there is no value raises TimeoutError.

        """
        with self._condition:
            if self._exception is not _NONE:
                if self._exception is None:
                    return self.value
                raise self._exception
            elif block:
                self._condition.wait(timeout)
                if self._exception is not _NONE:
                    if self._exception is None:
                        return self.value
                    raise self._exception

            # if we get to this point we timeout
            raise self._timeout_factory()

self.get_async(path, watch=watch).get()

def _read_response(self, header, buffer, offset):
	...
	try:
		response = request.deserialize(buffer, offset)
	except Exception as exc:
		self.logger.exception(
			"Exception raised during deserialization "
			"of request: %s", request)
		async_object.set_exception(exc)
		return
	self.logger.debug(
		'Received response(xid=%s): %r', xid, response)

	# We special case a Transaction as we have to unchroot things
	if request.type == Transaction.type:
		response = Transaction.unchroot(client, response)

	async_object.set(response)
	...