nsq 环境搭建

官方文档
官方文档中文版
官方文档中文版PDF

nsq 虚拟机测试（单个 topic）

测试环境

object	describe
cpu	2
memory	8G
nsqd count	1
nsqlookup	1
message size	1KB
message count	100000
topic count	1
channel count	20/100
client count	20/100

测试报告

channel count	write mb/s	write kops/s	read mb/s	read kops/s	per_channel_read ops/s
20	19.9	15.2	8.0	8.0	400
100	7.73	5.9	33.3	33.3	333

结果分析

测试结果与官方差距极大，仅达到每秒万级（已满足需求），官方测试报告能到达每秒十万甚至百万级，下面是结果误差较大的可能原因

1. 官方测试数据 client 并发低为 9/16/4，该测试 client 并发为 20/100
2. 虚拟机配置问题，2 CPU，会限制降低 client 并发性能
3. 测试数据为 1KB 大于 官方测试中的 0.1 KB
4. 官方的 producer 端为 golang 写的客户端，该测试直接使用 API 接口用 ab 工具发送 http 请求测试，http 连接创建销毁费时

---

官方测试报告

3主机环境下，对于 100B 的消息达到了每秒百万级

3主机 和 4 主机测试

测试环境

object	describe
nsqlookup	1
message size	0.1 KB
topic count	3
channel count	3

测试报告

nsqd(host)	producer count	consumer count	write mb/s	write kops/s	read mb/s	read kops/s
3	9	9	80.315	842	76.946	807
4	16	16	105.898	1110	100.956	1059

单机测试

producer	consumer	write mb/s	write kops/s	read mb/s	read kops/s
1	4	31.738	166	82.500	433
4	4	36.321	190	135.130	708

---

测试脚本相关

启动 nsqd/nsqlookupd/nsqadmin

#!/bin/bash
readonly memQueueSize="${MEM_QUEUE_SIZE:-1000000}"
set -e
set -u

echo "# using --mem-queue-size=$memQueueSize"

if [ ! -x ./nsqd ]
then
  echo "Please copy executable nsqd here"
  exit
fi
if [ ! -x ./nsqlookupd ]
then
  echo "Please copy executable nsqlookupd here"
  exit
fi

echo "# running nsqlookupd and nsqd"
rm -f *.dat
./nsqlookupd --broadcast-address=127.0.0.1 >/dev/null 2>&1 &
nsqlookupd_pid=$!

./nsqd --mem-queue-size=$memQueueSize --lookupd-tcp-address=127.0.0.1:4160 --broadcast-address=127.0.0.1 >/dev/null 2>&1 &
nsqd_pid=$!

./nsqadmin --lookupd-http-address=127.0.0.1:4161 >/dev/null 2>&1 &
nsqadmin_pid=$!

创建 topic/channel

#!/bin/bash
readonly channelCount="${1:-1}"

echo "# using --channel-count=${channelCount}"

echo "# creating topic/channel"
for i in $(seq 1 ${channelCount})
do
    curl --silent -X POST "http://127.0.0.1:4151/topic/create?topic=topic_1" >/dev/null 2>&1
    if [ $i -lt 10 ]
    then
        curl --silent -X POST "http://127.0.0.1:4151/channel/create?topic=topic_1&channel=channel_0${i}" >/dev/null 2>&1
    else
        curl --silent -X POST "http://127.0.0.1:4151/channel/create?topic=topic_1&channel=channel_${i}" >/dev/null 2>&1
    fi
done

测试客户端代码

#include "nsq.h"

static void message_handler(nsqRdr *rdr, nsqdConn *conn, nsqMsg *msg, void *ctx)
{
    //_DEBUG("%s: %lld, %d, %s, %lu, %.*s\n", __FUNCTION__, msg->timestamp, msg->attempts, msg->id,
    //    msg->body_length, (int)msg->body_length, msg->body);
    int ret = 0;
    //TestNsqMsgContext * test_ctx = (TestNsqMsgContext *)ctx;
    //int ret= ctx->process(msg->body, msg->body_length);

    printf("====================================================\n");
    printf("MESSAGE: %d %s\n", (int)msg->body_length, msg->body);
    printf("====================================================\n");
    buffer_reset(conn->command_buf);

    if(ret < 0){
        nsq_requeue(conn->command_buf, msg->id, 100);
    }else{
        nsq_finish(conn->command_buf, msg->id);
    }
    buffered_socket_write_buffer(conn->bs, conn->command_buf);

    buffer_reset(conn->command_buf);
    nsq_ready(conn->command_buf, rdr->max_in_flight);
    buffered_socket_write_buffer(conn->bs, conn->command_buf);

    free_nsq_message(msg);
}

int main(int argc, char **argv)
{
    if (argc < 4) {
        printf("not enough args from command line\n");
        return 1;
    }
    nsqRdr *rdr;
    struct ev_loop *loop;
    void *ctx = NULL; //(void *)(new TestNsqMsgContext());

    loop = ev_default_loop(0);
    rdr = new_nsq_reader(loop, argv[2], argv[3], ctx, NULL, NULL, NULL, message_handler);

#ifdef NSQD_STANDALONE
    nsq_reader_connect_to_nsqd(rdr, argv[1], 4150);
//    nsq_reader_connect_to_nsqd(rdr, "127.0.0.1", 14150);
#else
    nsq_reader_add_nsqlookupd_endpoint(rdr, argv[1], 4161);
#endif
    nsq_run(loop);

    return 0;
}

启动测试客户端

#!/bin/bash
readonly channelCount="${1:-1}"

echo "# using --channel-count=$channelCount"

if [ ! -x ./test-nsqd ]
then
  echo "Please copy executable test_nsqd here"
  exit
fi

echo "# running test_nsqd"
for i in $(seq 1 ${channelCount})
do
    if [ $i -lt 10 ]
    then
        ./test-nsqd 127.0.0.1 topic_1 channel_0${i} >/dev/null 2>&1 &
    else
        ./test-nsqd 127.0.0.1 topic_1 channel_${i} >/dev/null 2>&1 &
    fi
done

ab 压力测试

ab -c 10 -n 100000 -p 1.txt -T '/text/plain' 'http://127.0.0.1:4151/pub?topic=topic_1'