前些日子写了 Java Redis stream 的基本 API 实践和封装文章：Redis stream Java API 实践，自然地也需要对这些 API 进行性能测试。

总的来说，Redis stream 的 API 跟 list 操作的 API 差不多，就是添加、读取、删除一类，对于消费组的 API，由于实际工作中并没有使用，所以暂时搁置。

性能测试思路

这里我设计了三个用例：

1. 添加消息
2. 读取消息
3. 添加、删除消息

前两个用例比较简单，所以这里就不演示了，本次用例展示两种动态测试模型的使用方式：动态线程和动态QPS。

简述一下第三个用例的思路：首先就是创建一个添加消息，然后成功之后返回一个redis.clients.jedis.StreamEntryID，再根据这个返回值进行删除操作。实际工作中暂时没想到这个实际的场景，以后再遇到其他使用场景再来进行性能测试。

性能测试用例

这里我先分享一下动态线程模型的用例，我把用到的参数都写成了静态的变量形式，这样比较方便，如果是实际业务场景中，应该传比较复杂的。不过问题不大，都可以写在com.funtest.groovytest.RedisStreamTest.FunTester#doing里面，或者在动态QPS的闭包中。

package com.funtest.groovytest

import com.funtester.base.constaint.FunThread
import com.funtester.db.redis.RedisBase
import com.funtester.frame.SourceCode
import com.funtester.frame.execute.FunConcurrent
import redis.clients.jedis.params.XAddParams

import java.util.concurrent.atomic.AtomicInteger

class RedisStreamTest extends SourceCode {

    static RedisBase driver

    static def index = new AtomicInteger(0)

    static def desc = "Redis stream性能测试实践"

    static def map

    static def key = "FunTesterStream"

    static def params

    public static void main(String[] args) {
        driver = new RedisBase("127.0.0.1", 6379)
        params = XAddParams.xAddParams().maxLen(1000)
        map = new HashMap<String, Integer>()
        map.put("key1", 1)
        map.put("key2", 2)
        driver.xadd(key, params, map)
        params = XAddParams.xAddParams()
        output driver.xlen(key)
        new FunConcurrent(new FunTester(),).start()

    }

    private static class FunTester extends FunThread {


        FunTester() {
            super(null, desc + index.getAndIncrement())
        }

        @Override
        protected void doing() throws Exception {
            def xadd = driver.xadd(key, params, map)
            driver.xdel(key,xadd)
        }

        @Override
        FunTester clone() {
            return new FunTester()
        }
    }
}

下面分享一下动态 QPS 模型的压测用例：

package com.funtest.groovytest

import com.funtester.db.redis.RedisBase
import com.funtester.frame.SourceCode
import com.funtester.frame.execute.FunQpsConcurrent
import redis.clients.jedis.params.XAddParams

class RedisStreamQPSTest extends SourceCode {

    static RedisBase driver

    static def desc = "Redis stream性能测试实践"

    static def map

    static def key = "FunTesterStream"

    static def params

    public static void main(String[] args) {
        driver = new RedisBase("127.0.0.1", 6379)
        params = XAddParams.xAddParams().maxLen(1000)
        map = new HashMap<String, Integer>()
        map.put("key1", 1)
        map.put("key2", 2)
        driver.xadd(key, params, map)
        params = XAddParams.xAddParams()
        def test  = {
            def xadd = driver.xadd(key, params, map)
            driver.xdel(key,xadd)
        }
        new FunQpsConcurrent(test).start()

    }

}

控制台输出

15:50:41.091 main 
  ###### #     #  #    # ####### ######  #####  ####### ######  #####
  #      #     #  ##   #    #    #       #         #    #       #    #
  ####   #     #  # #  #    #    ####    #####     #    ####    #####
  #      #     #  #  # #    #    #            #    #    #       #   #
  #       #####   #    #    #    ######  #####     #    ######  #    #

15:50:41.474 main redis连接池IP：127.0.0.1，端口：6379，超时设置：5000
15:50:42.728 Deamon 守护线程开启!
15:50:45.761 main 当前设计QPS:1,实际QPS:0 活跃线程数:0 单线程效率:0
15:50:50.780 main 当前设计QPS:1,实际QPS:1 活跃线程数:1 单线程效率:1
15:50:55.798 main 当前设计QPS:1,实际QPS:1 活跃线程数:0 单线程效率:1
进程已结束,退出代码130 (interrupted by signal 2: SIGINT)

BUG 挖掘机·性能征服者·头顶锅盖

• 性能测试专题
• Java、Groovy、Go、Python
• 单测&白盒
• FunTester 社群风采
• 测试理论鸡汤
• 接口功能测试专题
• FunTester 视频专题
• 案例分享：方案、BUG、爬虫
• UI 自动化专题
• 测试工具专题

阅读原文，跳转我的仓库地址