简单的redis并发测试

我会带着你远行 2022-03-28 11:54 446阅读 0赞

用到的高并发模块：gevent

一、给redis中写入8万个key。每个key的大小约12k

#filename:redis1.py
    import redis
    import  os,time,threading
    import uuid
    import random
    # pool =redis.ConnectionPool(host='192.168.1.3',port=6379)
    # r = redis.Redis(connection_pool=pool)
    
    r=redis.Redis(host='192.168.1.3', port=6379)
    def setKey():
        num = 1
        set_start = time.clock()
        while num <80000:
    
            mykey = 'test' + str(num)
            num = num + 1
            myvalue=str(uuid.uuid1())*250 + str(random.randint(100000000,1000000000))
    
            r.set(mykey,myvalue)
            # print(r[mykey])
            # print(r.get(mykey))
    
        set_end = time.clock()
        set_times = set_end - set_start
        print('单线程存8万个key的时间:' + str(set_times)+ 's')
    if __name__ == '__main__':
        setKey()

二、并发测试脚本：

#filename:redis2.py
    from gevent import monkey
    monkey.patch_all()
    import gevent
    import redis
    import  time
    
    r = redis.Redis(host="192.168.1.3",port=6379)
    
    def getFunc(key):
        """取key"""
        v=r.get(key)
        # print(v)
    
    def call_gevent(count):
        """调用gevent 模拟高并发"""
        begin_time = time.time()
        run_gevent_list = []
        num = 1
        for i in range(count):
            print('--------------%d--Test-------------' % i)
            mykey = 'test' + str(num)
            run_gevent_list.append(gevent.spawn(getFunc,mykey))
            num = num + 1
        gevent.joinall(run_gevent_list)
        end = time.time()
        print('测试并发量'+ str(count))
        print('单次测试时间（平均）s:', (end - begin_time) / count)
        print('累计测试时间 s:', end - begin_time)
    
    if __name__ == '__main__':
        # 并发请求数量
        test_count = 20000  #改变并发量查看测试效果。。我这里取7000，10000，20000进行测试。记得将rdis的最大连接数改为30000并重启redis。
        # while 1:
        call_gevent(count=test_count)

测试思路：执行上面的脚本，同时另外再执行一个脚本（计算另外单独读取一个key需要的时间）。脚本内容如下：

#filename:redis3.py
    '''get a key'''
    
    import redis
    import  os,time,threading
    import uuid
    import random
    
    
    def get_key():
        start = time.time()
        r = redis.Redis(host='192.168.1.3', port=6379)
    
        r.get('test3000')
        end = time.time()
        times =end -start
        print('另外建立一个连接,取一个key的时间:' + str(times))
    
    if __name__ == '__main__':
        get_key()

三、测试结果：

执行命令：python3.6  redis2.py   && python3.6 redis3.py

查看连接数：netstat -an | grep :6379 | awk -F ':' '\{print $2\}' | awk -F " " '\{print $2\}' | awk '\{sum\[$1\]+=1\} END \{for(k in sum) print sum\[k\] "  "k\}'  | sort -nr | head   | column  -t

结果展示：

测试并发量20000  
单次测试时间（平均）s: 0.0005465076088905335  
累计测试时间 s: 10.930152177810669  
另外建立一个连接,取一个key的时间:0.1315782070159912

测试并发量10000  
单次测试时间（平均）s: 0.0004366687774658203  
累计测试时间 s: 4.366687774658203  
另外建立一个连接,取一个key的时间:0.0015645027160644531

测试并发量7000  
单次测试时间（平均）s: 0.0004212432588849749  
累计测试时间 s: 2.948702812194824  
另外建立一个连接,取一个key的时间:0.001592397689819336