1 watch、watch_once

我们这里开一个进程watch某一个key，在另外一个进程里进行替换或者删除key。

import etcd3
import json
import time
etcd = etcd3.client(host='127.0.0.1', port=2379)

# 查看当前集群的kv
list_of_kv = list(etcd.get_all())
for value, kv_msg in list_of_kv:
    print("key:{}, value:{}" .format(kv_msg.key, value))
# key:b'key', value:b'hello'

# 对一个key进行watch
# 这里是阻塞的，我们在其他的程序中进行key的修改
events_iterator, cancel = etcd.watch('key')
for event in events_iterator:
    print(event)
print(cancel)
etcd.close()

import etcd3
import json
etcd = etcd3.client(host='127.0.0.1', port=2379)

# 修改key的value
print(etcd.replace('key', initial_value='hello', new_value='hello2'))
print(etcd.get('key'))
# 删除key
print(etcd.delete('key'))
print(etcd.get('key'))
etcd.close()

1.1 replace

1.2 delete

可以看到在接收到事件后watch进程仍在监听状态

我们可以使用watch_once在第一个事件后停止

监视一系列key

import etcd3
import json
import time
etcd = etcd3.client(host='127.0.0.1', port=2379)

print(etcd.put('demo/bar1', 'doot'))
print(etcd.put('demo/bar2', 'doot'))
print(etcd.put('demo/bar3', 'doot'))
print(etcd.put('demo/bar4', 'doot'))
# 查看当前集群的kv
list_of_kv = list(etcd.get_all())
for value, kv_msg in list_of_kv:
    print("key:{}, value:{}" .format(kv_msg.key, value))
# key:b'key', value:b'hello'

# 对一个key进行watch
# 这里是阻塞的，我们在其他的程序中进行key的修改
events_iterator, cancel = etcd.watch_prefix('demo/')
for event in events_iterator:
    print(event)
print(cancel)
etcd.close()

3 cancel_watch

当我们不想继续watch后，我们可以显式取消watch

这里我们在第3次处理watch事件后进行取消

import etcd3
import json
import time
etcd = etcd3.client(host='127.0.0.1', port=2379)

# 查看当前集群的kv
list_of_kv = list(etcd.get_all())
for value, kv_msg in list_of_kv:
    print("key:{}, value:{}" .format(kv_msg.key, value))
# key:b'key', value:b'hello'

# 对一个key进行watch
# 这里是阻塞的，我们在其他的程序中进行key的修改
cnt = 0
events_iterator, cancel = etcd.watch('key')
for event in events_iterator:
    cnt += 1
    print(event)
    print("handle times:{}" .format(cnt))
    if 3 == cnt:
        cancel() # cancel底层调用的就是cancel_watch(watchid)
        print("cancel watch")
etcd.close()

很显然，如果我们在主线程进行event监控，会造成主线程阻塞。

我们可以向watch加入回调函数，这样就不会阻塞主进程了。

import etcd3
import time
etcd = etcd3.client(host='127.0.0.1', port=2379)

def callback(resp):
    # resp is a events_iterator
    for event in resp.events:
        print("Key:{}发生改变, 新的value是:{}" .format(event.key, event.value))

etcd.add_watch_callback('key', callback)
# 程序主流程
while True:
    time.sleep(1)

5 参考

https://www.modb.pro/db/53046