java - 泛美卫生组织客户限制？

Question

我正在使用 mosquitto 代理和客户端以及泛美卫生组织客户端对服务/客户端服务进行一些性能测试。我得到了一些奇怪的结果:

部署说明:

• 3台机器；生产者、经纪人、消费者
• 生产者:6 个尽可能快地使用 mosquitto_pub 的 Python 脚本。见下文。
• 消费者:简单的java客户端如下所示。订阅所有主题。
• 硬件细节没有显示出显着差异。

1) Mosquitto 每秒收到约 1459.5055 条消息，但仅发送 973.9596666666667 条消息。订阅者只会获得 485.5458333333333 。

2) 无论创建多少 paho 客户端实例，性能都不会提高。例如。如果您在一个主题中运行 6 个生产者，在两个主题中运行 2 个消费者，您将得到 485.5458333333333。但是，如果您向另一个主题添加 6 个生产者(已检查消息总量是否增加)，则总性能保持不变，并且每个主题除以二。

3) 如果您对两个独立的 java 应用程序进行精确测试，性能不会下降。每个应用程序都获得最大性能。

在任何情况下，CPU 或内存都会达到任何限制。

Producers.py

from datetime import datetime, date, time
import os,sys,time, json, random, itertools

arg = sys.argv
host="broker"

n=1
if len(arg)>1:
    n = int(arg[1])

i=0
while True :

    payload = {"id":str(n),"Time":datetime.now().strftime("%Y-%m-%dT%H:%M:%S.00Z"),"ResultValue":1.0,"ResultType":"integer","Datastream":{"id":str(n)}}

    os.system( "mosquitto_pub -h "+host+" -t "+("/"+str(payload["id"])+" -m " +str(json.dumps(json.dumps(payload)))+"")

Consumer.java

package eu.linksmart.testing;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.util.UUID;

public class Application implements MqttCallback {
    public Application() {
        id++;
    }

    public static void main(String[] args) {

        try {

            Application app = new Application();
            create("1",new Application());
            create("2",new Application());

            while (true)
                try {
                    Thread.sleep(30000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

        } catch (MqttException e) {
            e.printStackTrace();
        }

    }
    static void create(String id, Application app) throws MqttException {
        MqttClient mqttClient =  new MqttClient("tcp://broker:1883", UUID.randomUUID().toString(), new MemoryPersistence());
        mqttClient.connect();
        mqttClient.subscribe("/"+id+"/#", 1);

        mqttClient.setCallback(app);
    }
    long acc =0;
    int i=0;
    long start= System.nanoTime();
    static  int id=0;
    @Override
    public void connectionLost(Throwable throwable) {

    }

    @Override
    public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {

        i++;
        acc = (System.nanoTime()-start);
        if(acc/1000000>1000){
            start = System.nanoTime();
            System.out.println(String.valueOf((i * 1000000000.0) / acc));
            acc =0;
            i=0;
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

    }

}

例如主题 1 的生产者运行如下:

python Producers.py 1&

什么限制了 java 应用程序内的 paho 客户端？

Answer 1

经过大量调试后，我发现了问题所在。

主题$SYS/broker/load/messages/received/1min在我发送时报告了更多消息。可能正在将协议(protocol)消息计为消息。因此，在空闲状态下，该主题会向一名订阅者报告 3.22。所以我认为我每秒发送 1459.5055，这是由 mosquitto 报告的。但我只发送了 485.5458333333333。

因此，不要相信此主题的应用程序消息负载!