mqtt websocket优势？
一、mqtt websocket优势？
MQTT和WebSocket都是用于实现实时通信的协议，但它们有不同的优势。MQTT是一种轻量级的发布/订阅协议，适用于低带宽和不稳定网络环境。它具有低的网络开销和较小的数据包大小，适合在物联网设备之间传输消息。
WebSocket是一种全双工通信协议，通过长连接实现实时双向通信。它可以在浏览器和服务器之间建立持久连接，实现实时的双向数据传输，适用于Web应用程序的实时更新和交互。
WebSocket提供更高的实时性和更低的延迟，适合需要频繁交换数据的应用场景，如在线聊天、实时游戏等。选择使用MQTT还是WebSocket取决于具体的应用需求和网络环境。
二、mqtt protobuf，区别？
mqtt protobuf的区别是：文本格式不同。
1.文本聊天内容传输时，超过280字节，zip压缩比较有意义；
2.少量数据传输(<420字节），protbuffer压缩比更高，比较有优势；
3.内容越多，文本传输量越大，zip压缩优势越明显；
4.建议对内容超过一定数量的信息可以再进行zip压缩，以便缩小传输量；（参见600汉字，1000汉字聊天内容对比）
三、mqtt配置方法？
配置 MQTT 协议的方法包括以下几个步骤：
首先，确定所需的 MQTT 服务器和端口号，并连接到服务器。
接下来，设置客户端的名称和相关的身份验证信息，如用户名和密码。
然后，选择发布和订阅的主题，并设置订阅和发布的 QoS 等级。
最后，根据需要配置其他高级选项，如保持连接和重新连接机制，以确保稳定的通信。通过这些步骤，可以成功地配置和使用 MQTT 协议实现消息传递和通信。
四、flink如何对接mqtt？
flink同自定义数据源emqtt可以按照以下方式对接：
测试环境 ：
单机服务器：8核12G，
设置并行度为2，
测试结果： 执行3分钟， 大概1秒4万的并发量， 未延迟， 只是简单测试， 并未达到极限。
Client11.java （主要用来处理emqtt的配置）
package com.flink.utils.emqtt;
import java.net.URISyntaxException;
import java.util.ArrayList;
import org.fusesource.mqtt.client.Future;
import org.fusesource.mqtt.client.FutureConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.Message;
import org.fusesource.mqtt.client.QoS;
import org.fusesource.mqtt.client.Topic;
/**
* 客户端订阅消息
*/
public class Client11 {
private final static String CONNECTION_STRING = "tcp://192.168.3.101:61613";
private final static boolean CLEAN_START = true;
private final static short KEEP_ALIVE = 30;// 低耗网络，但是又需要及时获取数据，心跳30s
// private final static String CLIENT_ID = "client11";
public static Topic[] topics = {
new Topic("$share/group/0001/#", QoS.AT_LEAST_ONCE), // 2 只有一次
new Topic("mqtt/aaa", QoS.AT_LEAST_ONCE), // 1 至少一次
new Topic("mqtt/ccc", QoS.AT_MOST_ONCE)}; // 0 至多一次
public final static long RECONNECTION_ATTEMPT_MAX = 6;
public final static long RECONNECTION_DELAY = 2000;
public final static int SEND_BUFFER_SIZE = 64;// 发送最大缓冲为2M
public ArrayList<String> list = new ArrayList<String>();
public FutureConnection start() {
String CLIENT_ID = (int)(Math.random()*100) + "";
// 创建MQTT对象
MQTT mqtt = new MQTT();
try {
// 设置mqtt broker的ip和端口
mqtt.setHost(CONNECTION_STRING);
// 连接前清空会话信息
mqtt.setCleanSession(CLEAN_START);
// 设置重新连接的次数
mqtt.setReconnectAttemptsMax(RECONNECTION_ATTEMPT_MAX);
// 设置重连的间隔时间
mqtt.setReconnectDelay(RECONNECTION_DELAY);
// 设置心跳时间
mqtt.setKeepAlive(KEEP_ALIVE);
// 设置缓冲的大小
mqtt.setSendBufferSize(SEND_BUFFER_SIZE);
//设置客户端id
mqtt.setClientId(CLIENT_ID);
// 获取mqtt的连接对象BlockingConnection ,采用Future模式 订阅主题
// final FutureConnection connection = mqtt.futureConnection();
FutureConnection connection = mqtt.futureConnection();
connection.connect();
connection.subscribe(topics);
return connection;
} catch (URISyntaxException e) {
e.printStackTrace();
} catch (Exception e) {
e.printStackTrace();
} finally {
}
return null;
}
}
SourceTest.java （flink的自定义数据源+ 数据存储redis）
package com.flink;
import com.google.gson.Gson;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import java.util.*;
import org.fusesource.mqtt.client.Message;
import org.fusesource.mqtt.client.Future;
import com.flink.utils.emqtt.Client11;
import org.fusesource.mqtt.client.FutureConnection;
public class SourceTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStream<String> inputStream= env.addSource(new EmqttSource());
// inputStream.print();
DataStream<List<deviceData>> redisData = inputStream.rebalance().map(new MapFunction<String, List<deviceData>>() {
@Override
public List<deviceData> map(String s) throws Exception {
String[] array = s.split("@@");
String topic = (String) array[1];
String message = (String) array[0];
return RulesEngine(message, topic);
}
});
// redisData.addSink(new OpnetsdbWriter());
redisData.addSink(new redisWriter());
env.execute("Intsmaze Custom Source");
}
public static List<deviceData> RulesEngine(String message, String topic){
try {
// String topic = "3333/D4:36:39:1A:0D:D3/Send/Data/FOCAS";
List<deviceData> d = new ArrayList<>();
Gson gson = new Gson();
Map<String, Object> map = new HashMap<String, Object>();
map = gson.fromJson(message, map.getClass());
String dataType = (String) map.get("type");
if(dataType.equals("Data")||dataType.equals("data")) {
ArrayList dataList = (ArrayList) map.get("values");
String[] array = topic.split("/");
for (int i = 0; i < dataList.size(); i++) {
deviceData d1 = new deviceData();
Map<String, String> dataDict = (Map<String, String>) dataList.get(i);
d1.machID = dataDict.get("machID");
d1.compID = array[0];
d1.gateMac = array[1];
d1.Type = dataType;
d1.operationValue = dataDict.get("name");
d1.operationData = dataDict.get("data");
d1.gatherTime = dataDict.get("time");
d.add(d1);
}
return d;
}else{
System.out.println("无法解析该类型数据");
}
} catch (Throwable t) {
t.printStackTrace();
}
return null;
}
// SourceFunction<String>
public static class EmqttSource implements ParallelSourceFunction<String> {
private static final long serialVersionUID = 1L;
private volatile boolean isRunning = true;
@Override
public void run(SourceContext<String> ctx) throws Exception {
Client11 client = new Client11();
FutureConnection connection = client.start();
int Num = 0;
String msg;
while (isRunning) {
Future<Message> futrueMessage = connection.receive();
Message message = futrueMessage.await();
Num++;
// System.out.println("MQTTFutureClient.Receive Message " + "Topic Title :" + message.getTopic() + " context :"
// + String.valueOf(message.getPayloadBuffer()));
// ctx.collect(Num + " context :" + String.valueOf(message.getPayloadBuffer()));
msg = String.valueOf(message.getPayloadBuffer()).substring(6);
ctx.collect(msg + "@@" + message.getTopic());
}
}
@Override
public void cancel() {
isRunning = false;
}
}
}。
五、mqtt与netty区别？
mqtt
MQTT(消息队列遥测传输)是ISO标准(ISO/IEC PRF 20922)下基于发布/订阅范式的消息协议。它工作在 TCP/IP协议族上，是为硬件性能低下的远程设备以及网络状况糟糕的情况下而设计的发布/订阅型消息协议，为此，它需要一个消息中间件。
netty
Netty是由JBOSS提供的一个java开源框架。Netty提供异步的、事件驱动的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序。
六、mqtt数据命名规则？
MQTT(Message Queuing Telemetry Transport)是一种分布式消息传递协议，用于在分布式系统中传输消息。在MQTT中，数据命名规则是非常重要的，以下是MQTT数据命名规则的一些基本要素：
1. 主题(Topic):MQTT数据的主题是指包含消息的特定数据结构。主题通常由三个部分组成：元数据(Header)、消息内容(Body)和标识符(識別符)。
2. 标识符(識別符):MQTT标识符用于唯一标识一个主题。标识符可以是数字、字母或下划线，通常是一个单词的第一个字母。
3. 订阅(Subsciber)：订阅是指订阅特定主题的消息传递。订阅可以指定要接收消息的客户端。
4. 发布(Publish)：发布是指将消息从一个主题发送到另一个主题。发布可以指定要发送的消息内容、消息标识符和订阅者。
5. 消息类型(Message Type):MQTT消息类型用于指示消息的内容类型。MQTT消息类型包括命令(Command)、消息(Message)、请求(Request)和应答(Response)。
6. 数据结构(Data Structure):MQTT数据结构是指包含消息内容的特定数据结构。例如，一个主题可以包含一个元数据、一个消息内容和一个标识符。
MQTT数据命名规则的目的是使消息易于理解和处理。通过遵循这些规则，可以更容易地识别和检索消息。
七、mqtt与webservices区别？
MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器（M2M）通信和物联网（IoT）。
Web Service是一个平台独立的，低耦合的，自包含的、基于可编程的web的应用程序，可使用开放的XML（标准通用标记语言下的一个子集）标准来描述、发布、发现、协调和配置这些应用程序，用于开发分布式的交互操作的应用程序。
八、mqtt与tcp区别？
MQTT是建立在TCP协议之上的一层应用层协议，是不同层面的。我想你可能是想问MQTT比socket报文交互好在哪里？这样的话有几点重要原因:
1.最大优点是降低开发复杂度和开发成本，解决了网络编程中重连机制，报文解析中粘包处理，字节流处理，高并发处理，保证数据到达，保证数据唯一到达，等等问题。
2.MQTT使用json作为交互数据格式，便于理解和对接，使得不同系统之间，不同设备和系统之间交互更加简单，降低了开发和沟通复杂度。
3.但在效率上，MQTT还是比不过直接网络编程，用netty开发也难度不高
九、mqtt通讯协议详解？
MQTT是一种轻量级的通信协议，能够实现设备与设备之间的相互通信。它具有低开销、高效率、可靠性强等特点，适用于各种物联网应用场景，例如智能家居、工业自动化等。
MQTT采用发布/订阅模式，设备可以订阅一个主题，其他设备可以发布消息到这个主题，实现设备之间的数据传输，同时也支持点对点通信，具有很高的灵活性和可扩展性。
十、mqtt activemq优缺点？
优点:对tps要求比较低，可控性比较好
缺点:不支持消息自动分片机制，无分片功能 吞吐量低