MQTT发送消息至EMQX桥接clickhuose 数据丢失未落表

wenyuzdl · 2023 年9 月 22 日 10:39

我在网上看到可用于高并发场景的MQTT平台有EMQX
但是在测试过程中 500并发的消息处理不掉有数据丢失
具体操作是本地代码每1秒发送500条消息到emqx emqx配置桥接clickhuose的数据处理消息接收和发出测试正常落数据库没有完全落

Shawn · 2023 年9 月 26 日 02:43

有数据丢失还是有发送错误？EMQX Dashboard 上数据桥接的部分有一些统计和计数，你看看是 dropp 还是什么，还需要看一下日志文件里是否有错误日志。

wenyuzdl · 2023 年9 月 26 日 02:54

wenyuzdl · 2023 年9 月 26 日 03:32

@RequestMapping(value = “/publishTopic”)
public String publishTopic(String topic, int sendMessage) {
long start = System.currentTimeMillis();
log.info("task start time " + start);
long end = System.currentTimeMillis(); // 结束时间为当前时间 + 20 秒
while (secondsCount < sendMessage) {
try {
int numTasks = 500; // 并发任务数量
ExecutorService executor = Executors.newFixedThreadPool(numTasks); // 创建固定大小的线程池
for (int i = 0; i < numTasks; i++) {
final int taskId = i;
Runnable task = new Runnable() {
@Override
public void run() {
// 在这里编写并发执行的代码逻辑
String s = TemperatureSensor.readTemperature();
try {
mqttConfig.publish(false,topic,s);
log.info("Task " + taskId + “: Executing”);
} catch (Exception e) {
log.info("Task " + taskId + “: error”);
e.printStackTrace();
}
}
};
executor.execute(task); // 提交任务给线程池执行
}
executor.shutdown(); // 关闭线程池
} catch (Exception e) {
e.printStackTrace();
}
try {
Thread.sleep(1000);
secondsCount++;
} catch (InterruptedException e) {
e.printStackTrace();
}
end = System.currentTimeMillis();
log.info(“Task " + (end - start) + " is delivered.”);
}
secondsCount=0;
System.out.println(“topic:” + topic);
System.out.println(“message:” + sendMessage);
return “topic:” + topic + “\nmessage:” + sendMessage;
}