客户端并发测试 Mqtt Broker

备份转载金总的Note

概述

本文描述了对 example broker 进行性能测试时, 客户端所需进行配置的相关步骤。

测试软件

example broker 作为一个 Mqtt 服务器, 对它进行测试时需要一款对应的 Mqtt 客户端。考虑到便捷性我们采用了基于 nodejs 的开源 Mqtt 客户端: MQTT.js。在运行测试程序前需要安装如下软件。

1. Git
2. Node.js
3. npm

在自己预设的工作目录执行 git clone https://github.com/mqttjs/MQTT.js 下载最新的 MQTT.js。进入 MQTT.js 执行 npm install 下载所需的依赖包。

由于测试程序需要支持自定义 ip, 因此需要修改 MQTT.js 程序，允许增加额外的参数。使用文档中所付的 tcp.js 替换目录 MQTT.js 目录下的同名文件。复制文档中所付的 performace_test.js 至 MQTT.js 目录下。

执行测试程序时需要额外的参数，以下是参数示例说明: node performance_test.js joshua 192.168.1.100 10.65.81.60:12306 50000 其中 joshua 为 mqtt client id 的前缀, 测试程序会在后面根据序号添加数字，保证每个唯一连接的 mqtt client id。192.168.1.100 为本机 ip, 应保证该 ip 为本机可用 ip 之一, 且可以正常的连通 example broker ip。10.65.81.60:12306 为需要测试的 mqtt 服务器地址及端口, 50000 为本次测试连接的客户端数目。以上参数可根据测试所需自行修改。

操作系统

在使用测试服务器进行测试时, 如果所需并发量在 100,000 以下, 可以使用 Mac OS X 作为测试客户端所用操作系统。如果大于 100, 000 则建议使用 Linux 操作系统, 并添加多个 ip 地址的形式进行测试。 下面会针对这两种操作系统进行相关配置的介绍, 部分命令可能需要 root 权限。

Mac OS X

操作系统的参数修改有两大类, 即 tcp/ip 相关的网络参数及文件描述符的相关参数。

执行 sysctl -a | grep port, 查看 net.inet.ip.portrange.first, net.inet.ip.portrange.hifirst 的值。这两项值定义了操作系统种用户可用端口的起始范围, 根据需要连接的并发量进行修改, 我们建议修改为 10,000。修改的命令如下: sysctl -w net.inet.ip.portrange.first=10000, sysctl -w net.inet.ip.portrange.hifirst=10000。

执行 sysctl -a | grep files, 查看 kern.maxfiles, kern.maxfilesperproc 两项的值。这两项值决定了用户可以打开的最大文件数和每个进程允许打开的最大文件数。我们建议修改为 1,000,000 和 100,000。修改的命令如下: sysctl -w kern.maxfiles=1000000, sysctl -w kern.maxfilesperproc=100000。

在 4 core, 16G 内存的 rMBP 上, 单个 ip 可以稳定的保持最多 50,000 个长连接。如果需要继续增加连接个数, 可以采取增加 ip 的方式, 但是根据实际的测试结果, 单台 rMBP 能够维持的稳定的长连接个数在 70,000 个左右, 因此如果需要大量长连接测试的情况, 我们推荐使用 Linux 服务器。

Linux

与 Mac OS X 类似, 首先执行 sysctl -a | grep port 查看 net.ipv4.ip_local_port_range 的值, 该项参数决定了用户在系统内可以端口的范围, 我们建议修改的下限为 10,000。

执行 sysctl -a | grep file 查看 fs.file-max 的值, 该项参数决定了用户在系统中允许打开的最大文件个数, 我们建议修改为 1,500,000。

在 Linux 中可以通过添加虚拟 ip 的形式来增大测试的连接数, 在我们的测试案例中, 我们给单台 Linux 服务器添加了 5 个虚拟 ip, 共计 6 个 ip, 单个 ip 可以稳定的维持 50,000 个长连接, 因此单台服务器可以产生 300,000 个稳定的长连接。添加虚拟 ip 的命令示例如下: ifconfig eth0:1 192.168.190.151, 该命令在网卡 eth0 上添加了虚拟 ip 192.168.190.151。

运行测试程序

将工作目录切换至 MQTT.js 的根目录, 执行之前介绍的客户端命令, 即开始运行测试的客户端程序。

var mqtt = require('./');
var EventEmitter = require('events').EventEmitter;
var connectingCount = 0;
var connectedCount = 0;
var connectClosedCount = 0;
var connectErrorCount = 0;
var tcpConnectedCount = 0;

var cid = 0;


var cmdOpts = process.argv.splice(2);
var clid = cmdOpts[0];
var localhostIp = cmdOpts[1];
var serverIp = cmdOpts[2];
var limit = cmdOpts[3];

console.log(clid);
console.log(localhostIp);
console.log(limit);

var doconnectInterval = null;

function addTimer(){
  doconnectInterval = setInterval(doConnect, 10);
}

function delTimer(){
  clearInterval(doconnectInterval);
}

addTimer();

setInterval(
  function(){
    if(connectedCount + connectingCount < limit){
      addTimer();
    }
    console.log("src[" + localhostIp
        + "] tcpConnectedCount[" + tcpConnectedCount
                + "] mqttConnectedCount[" + connectedCount
        + "] mqttConnectingCount[" + connectingCount
        + "] tcpConnectClosedCount[" + connectClosedCount
        + "] tcpConnectErrorCount[" + connectErrorCount + "]");
  },1000
);

function doConnect() {

  for(var i = 0; i < 8192; i++) {
    if( connectedCount + connectingCount >= limit || connectingCount >= 4096){
      delTimer();
      return true;
    }
    mqttTest();
  }
}

function mqttTest(){
  var client = mqtt.connect('tcp://' + serverIp, {
    clientId: 'mqtt-' + clid + '-' + (cid++),
    clean: false,
    keepalive: 300,
    reconnectPeriod: -1,
    localAddress: localhostIp
  });

  client.stream.on('connect', function(){
    tcpConnectedCount++;
  });
  connectingCount++;
  client.isConnected = false;

  client.on('connect', function(){
    connectedCount++;
    connectingCount--;
    client.isConnected = true;
    client.subscribe('hello');
  });

  client.on('close', function(data) {
    console.log('connection[' + (client.isConnected ? 'connected':'connecting') + '] close: ' + data);
    connectClosedCount++;
    if(client.isConnected){
      connectedCount--;
      tcpConnectedCount--;
    }else{
      connectingCount--
    }
  });

  client.on('message', function(topic, payload){
    console.log(topic + '\n' + payload);
  });

  client.on('error', function(error) {
    connectErrorCount++;
    console.log('mqtt error: ' + error);
  });
};

'use strict';
var net = require('net');

/*
  variables port and host can be removed since
  you have all required information in opts object
*/
function buildBuilder (client, opts) {
  var port, host, localAddress;
  opts.port = opts.port || 1883;
  opts.hostname = opts.hostname || opts.host || 'localhost';

  port = opts.port;
  host = opts.hostname;
  localAddress = opts.localAddress;

  if (!opts.localAddress) {
    return net.createConnection(port, host);
  }
  return net.createConnection({'port': port, 'host': host, 'localAddress': localAddress});
}

module.exports = buildBuilder;