【发布时间】：2019-08-09 09:05:25
【问题描述】：
我目前正在使用 Eclipse Paho 为更大的软件开发 MQTT 客户端服务，但遇到了性能问题。我收到一堆想要发布到代理的事件，并且我正在使用 GSON 对这些事件进行序列化。我已经对序列化和发布进行了多线程处理。根据基本基准，序列化和发布最多需要 1 毫秒。 我正在使用 ExecutorService 和大小为 10 的固定线程池（目前）。
我的代码目前每秒向 ExecutorService 提交大约 50 个 Runnable，但我的 Broker 每秒只报告大约 5-10 条消息。 我之前对我的 MQTT 设置进行了基准测试，并设法以非多线程方式每秒发送大约 9000 多条 MQTT 消息。
线程池是否有那么大的开销，我只能从中获得这么少量的发布？
public class MqttService implements IMessagingService{
protected int PORT = 1883;
protected String HOST = "localhost";
protected final String SERVICENAME = "MQTT";
protected static final String COMMANDTOPIC = "commands";
protected static final String REMINDSPREFIX = "Reminds/";
protected static final String VIOLATIONTOPIC = "violations/";
protected static final String WILDCARDTOPIC = "Reminds/#";
protected static final String TCPPREFIX = "tcp://";
protected static final String SSLPREFIX = "ssl://";
private MqttClient client;
private MqttConnectOptions optionsPublisher = new MqttConnectOptions();
private ExecutorService pool = Executors.newFixedThreadPool(10);
public MqttService() {
this("localhost", 1883);
}
public MqttService(String host, int port) {
this.HOST = host;
this.PORT = port;
}
@Override
public void setPort(int port) {
this.PORT = port;
}
@Override
public void setHost(String host) {
this.HOST = host;
}
@Override
public void sendMessage(AbstractMessage message) {
pool.submit(new SerializeJob(client,message));
}
@Override
public void connect() {
try {
client = new MqttClient(TCPPREFIX + HOST + ":" + PORT, IDPublisher);
optionsPublisher.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);
client.connect(optionsPublisher);
client.setCallback(new MessageCallback());
client.subscribe(WILDCARDTOPIC, 0);
} catch (MqttException e1) {
e1.printStackTrace();
}
}
}
以下代码是ExecutorService执行的Runnable。不过，这本身应该不是问题，因为它最多只需要 1-2 毫秒即可完成。
class SerializeJob implements Runnable {
private AbstractMessage message;
private MqttClient client;
public SerializeJob(MqttClient client, AbstractMessage m) {
this.client = client;
this.message = m;
}
@Override
public void run() {
String serializedMessage = MessageSerializer.serializeMessage(message);
MqttMessage wireMessage = new MqttMessage();
wireMessage.setQos(message.getQos());
wireMessage.setPayload(serializedMessage.getBytes());
if (client.isConnected()) {
StringBuilder topic = new StringBuilder();
topic.append(MqttService.REMINDSPREFIX);
topic.append(MqttService.VIOLATIONTOPIC);
try {
client.publish(topic.toString(), wireMessage);
} catch (MqttPersistenceException e) {
e.printStackTrace();
} catch (MqttException e) {
e.printStackTrace();
}
}
}
}
我不太确定是什么阻碍了我。 MQTT 本身似乎允许很多事件，这些事件也可能有很大的负载，而且网络也不可能成为问题，因为我目前将代理本地托管在与客户端相同的机器上。
通过进一步测试进行编辑：
我现在已经对我自己的设置进行了综合基准测试，该设置由本地托管的 HiveMQ 和 Mosquitto 代理组成，这些代理在机器上“本地”运行。使用 Paho 库，我以 1000 条为一批发送了越来越大的消息。对于每批，我计算了从第一条消息到最后一条消息的消息吞吐量。这个场景没有使用任何多线程。有了这个，我想出了以下性能图表：
同时运行客户端和代理的机器是具有 i7 6700 和 32 GB RAM 的桌面。代理可以访问其 VM 的所有内核和 8 GB 内存。
为了进行基准测试，我使用了以下代码：
import java.util.Random;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
public class MqttBenchmarker {
protected static int PORT = 1883;
protected static String HOST = "localhost";
protected final String SERVICENAME = "MQTT";
protected static final String COMMANDTOPIC = "commands";
protected static final String REMINDSPREFIX = "Reminds/";
protected static final String VIOLATIONTOPIC = "violations/";
protected static final String WILDCARDTOPIC = "Reminds/#";
protected static final String TCPPREFIX = "tcp://";
protected static final String SSLPREFIX = "ssl://";
private static MqttClient client;
private static MqttConnectOptions optionsPublisher = new MqttConnectOptions();
private static String IDPublisher = MqttClient.generateClientId();
private static int messageReceived = 0;
private static long timesent = 0;
private static int count = 2;
private static StringBuilder out = new StringBuilder();
private static StringBuilder in = new StringBuilder();
private static final int runs = 1000;
private static boolean receivefinished = false;
public static void main(String[] args) {
connect();
Thread sendThread=new Thread(new Runnable(){
@Override
public void run() {
Random rd = new Random();
for (int i = 2; i < 1000000; i += i) {
byte[] arr = new byte[i];
// System.out.println("Starting test run for byte Array of size:
// "+arr.length);
long startt = System.currentTimeMillis();
System.out.println("Test for size: " + i + " started.");
for (int a = 0; a <= runs; a++) {
rd.nextBytes(arr);
try {
client.publish(REMINDSPREFIX, arr, 1, false);
} catch (MqttPersistenceException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (MqttException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
try {
while (!receivefinished) {
Thread.sleep(10);
}
receivefinished = false;
System.out.println("Test for size: " + i + " finished.");
out.append("Sending Payload size: " + arr.length + " achieved "
+ runs / ((System.currentTimeMillis() - startt) / 1000d) + " messages per second.\n");
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
System.out.println(out.toString());
System.out.println(in.toString());
}
});
sendThread.start();
}
private static class MessageCallback implements MqttCallback {
@Override
public void messageArrived(String arg0, MqttMessage arg1) throws Exception {
if (messageReceived == 0) {
timesent = System.currentTimeMillis();
}
messageReceived++;
if (messageReceived >= runs) {
receivefinished = true;
in.append("Receiving payload size " + count + " achieved "
+ runs / ((System.currentTimeMillis() - timesent) / 1000d) + " messages per second.\n");
count += count;
messageReceived = 0;
}
}
@Override
public void deliveryComplete(IMqttDeliveryToken arg0) {
// TODO Auto-generated method stub
}
@Override
public void connectionLost(Throwable arg0) {
// TODO Auto-generated method stub
}
}
public static void connect() {
try {
client = new MqttClient(TCPPREFIX + HOST + ":" + PORT, IDPublisher);
optionsPublisher.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);
optionsPublisher.setAutomaticReconnect(true);
optionsPublisher.setCleanSession(false);
optionsPublisher.setMaxInflight(65535);
client.connect(optionsPublisher);
while (!client.isConnected()) {
try {
Thread.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
client.setCallback(new MessageCallback());
client.subscribe(WILDCARDTOPIC, 0);
} catch (MqttException e1) {
e1.printStackTrace();
}
}
}
奇怪的是，我想从我的应用程序发送的序列化消息只使用大约 4000 个字节。所以理论上的吞吐量应该是每秒大约 200 条消息。这仍然是回调函数内部计算时间较长引起的问题吗？我已经使用 mosquitto broker 取得了更好的结果，我将进一步测试，我可以将性能提升到什么程度。
感谢您的任何建议！
【问题讨论】：
-
什么是MessageSerializer？
-
那么你试过不使用 ExecutorService 吗？将线程设置为 1，或者只使用 1 个手动创建的工作线程，或者甚至尝试完全不使用线程？我相信 Paho 有自己的循环。我对您的用例有些怀疑。 MQTT 不是为大量并发的动态消息而设计的。此外，您的基准测试和此次执行是否具有相同的 QoS 级别？
-
我越来越怀疑 MQTT 是否适合您的用例。你有明显的大消息，你有很多，交付证明对你来说是必不可少的。 MQTT 可以满足一项要求，但不能满足所有要求。它还有其他设计目标。您是否有权使用或提议一个真正的消息队列（AMQP 或其他）？
-
还可以考虑在 QoS=0 的前提下设计应用层。 MQTT 是作为冗余数据的传输而诞生的，当您的应用程序可以处理偶尔的间隙和最终的一致性时，它真的会大放异彩。请注意，QoS>0 不是向目标消费者交付的证明，而是向最近的代理交付的证明，而且性能损失仍然令人瞠目结舌 - 正如您已经发现的那样。
-
我读到MqttClient 是一个阻塞实现，改用MqttAsyncClient 可能会有所改善。但是您确实需要一个持久的、持久的消息队列（是的，就像 AMQP，无论是 RabbitMQ 还是 ActiveMQ 或其他）。 MQTT 持久性只是最终的，优点是内存少、cpu 少、开销小。这些对你都没有用。
标签： java multithreading mqtt runnable executorservice