zookeeper测试

# -*- coding:utf-8 -*-
from kazoo.client import KazooClient
class PyZooConn(object):
# init function include connection method
def __init__(self):
self.zk = KazooClient(hosts='192.168.4.170:30405')
self.zk.start()
# get node data
def get_data(self, param):
result = self.zk.get(param)
print(result)
# create a node and input a value in this node
def create_node(self, node, value):
self.zk.create(node, value)
# close the connection
def close(self):
self.zk.stop()
if __name__ == '__main__':
path = "/abc/abc"
value = b"peter"
pz = PyZooConn()
pz.create_node(path, value)
pz.get_data(path)
pz.close()
压测

1.运行一个zk的压力测试容器

docker run --rm --name ztest -it daocloud.io/daocloud/zookeeper:feature-pressure_test /bin/bash


2.编辑配置文件

vi benchmark.conf（修改zookeeper服务器地址以及其他相关参数）


3.运行测试工具

./runBenchmark.sh test1111 ./benchmark.conf
robot脚本

*** Settings ***
Library XML
library ZookeeperLibrary
*** Variable ***
${aa}
*** Test Cases ***
#
ZookeeperLibrary.Connect To Zookeeper 192.168.124.115:12181
${aa} ZookeeperLibrary.Get Value /abc
BuiltIn.Log ${aa}