|
| 1 | +--- |
| 2 | +redirectTo: Programming-MQTT_apache.html |
| 3 | +--- |
1 | 4 | <!-- |
2 | 5 |
|
3 | 6 | Licensed to the Apache Software Foundation (ASF) under one |
|
17 | 20 | specific language governing permissions and limitations |
18 | 21 | under the License. |
19 | 22 |
|
20 | | ---> |
21 | | -# MQTT Protocol |
22 | | - |
23 | | -## 1. Overview |
24 | | - |
25 | | -MQTT (Message Queuing Telemetry Transport) is a lightweight messaging protocol designed for IoT and low-bandwidth environments. It operates on a Publish/Subscribe (Pub/Sub) model, enabling efficient and reliable bidirectional communication between devices. Its core objectives are low power consumption, minimal bandwidth usage, and high real-time performance, making it ideal for unstable networks or resource-constrained scenarios (e.g., sensors, mobile devices). |
26 | | - |
27 | | -IoTDB provides deep integration with the MQTT protocol, fully compliant with MQTT v3.1 (OASIS International Standard). The IoTDB server includes a built-in high-performance MQTT Broker module, eliminating the need for third-party middleware. Devices can directly write time-series data into the IoTDB storage engine via MQTT messages. |
28 | | - |
29 | | - |
30 | | - |
31 | | - |
32 | | -## 2. Configuration |
33 | | - |
34 | | -By default, the IoTDB MQTT service loads configurations from `${IOTDB_HOME}/${IOTDB_CONF}/iotdb-system.properties`. |
35 | | - |
36 | | -| **Property** | **Description** | **Default** | |
37 | | -| ------------------------ | ---------------------------------------------------------------------------------------------------------------------- | ------------------- | |
38 | | -| `enable_mqtt_service` | Enable/ disable the MQTT service. | FALSE | |
39 | | -| `mqtt_host` | Host address bound to the MQTT service. | 127.0.0.1 | |
40 | | -| `mqtt_port` | Port bound to the MQTT service. | 1883 | |
41 | | -| `mqtt_handler_pool_size` | Thread pool size for processing MQTT messages. | 1 | |
42 | | -| **`mqtt_payload_formatter`** | **Formatting method for MQTT message payloads. ****Options: `json` (tree model), `line` (table model).** | **json** | |
43 | | -| `mqtt_max_message_size` | Maximum allowed MQTT message size (bytes). | 1048576 | |
44 | | - |
45 | | -## 3. Write Protocol |
46 | | - |
47 | | -* Line Protocol Syntax |
48 | | - |
49 | | -```JavaScript |
50 | | -<measurement>[,<tag_key>=<tag_value>[,<tag_key>=<tag_value>]][ <attribute_key>=<attribute_value>[,<attribute_key>=<attribute_value>]] <field_key>=<field_value>[,<field_key>=<field_value>] [<timestamp>] |
51 | | -``` |
52 | | - |
53 | | -* Example |
54 | | - |
55 | | -```JavaScript |
56 | | -myMeasurement,tag1=value1,tag2=value2 attr1=value1,attr2=value2 fieldKey="fieldValue" 1556813561098000000 |
57 | | -``` |
58 | | - |
59 | | - |
60 | | - |
61 | | -## 4. Naming Conventions |
62 | | - |
63 | | -* Database Name |
64 | | - |
65 | | -The first segment of the MQTT topic (split by `/`) is used as the database name. |
66 | | - |
67 | | -```Properties |
68 | | -topic: stock/Legacy |
69 | | -databaseName: stock |
70 | | - |
71 | | - |
72 | | -topic: stock/Legacy/# |
73 | | -databaseName:stock |
74 | | -``` |
75 | | - |
76 | | -* Table Name |
77 | | - |
78 | | -The table name is derived from the `<measurement>` in the line protocol. |
79 | | - |
80 | | -* Type Identifiers |
81 | | - |
82 | | -| Filed Value | IoTDB Data Type | |
83 | | -|--------------------------------------------------------------------| ----------------- | |
84 | | -| 1<br>1.12 | DOUBLE | |
85 | | -| 1`f`<br>1.12`f` | FLOAT | |
86 | | -| 1`i`<br>123`i` | INT64 | |
87 | | -| 1`u`<br>123`u` | INT64 | |
88 | | -| 1`i32`<br>123`i32` | INT32 | |
89 | | -| `"xxx"` | TEXT | |
90 | | -| `t`,`T`,`true`,`True`,`TRUE`<br> `f`,`F`,`false`,`False`,`FALSE` | BOOLEAN | |
91 | | - |
92 | | - |
93 | | -## 5. Coding Examples |
94 | | -The following is an example which a mqtt client send messages to IoTDB server. |
95 | | - |
96 | | - ```java |
97 | | -MQTT mqtt = new MQTT(); |
98 | | -mqtt.setHost("127.0.0.1", 1883); |
99 | | -mqtt.setUserName("root"); |
100 | | -mqtt.setPassword("root"); |
101 | | - |
102 | | -BlockingConnection connection = mqtt.blockingConnection(); |
103 | | -String DATABASE = "myMqttTest"; |
104 | | -connection.connect(); |
105 | | - |
106 | | -String payload = |
107 | | - "test1,tag1=t1,tag2=t2 attr3=a5,attr4=a4 field1=\"fieldValue1\",field2=1i,field3=1u 1"; |
108 | | -connection.publish(DATABASE + "/myTopic", payload.getBytes(), QoS.AT_LEAST_ONCE, false); |
109 | | -Thread.sleep(10); |
110 | | - |
111 | | -payload = "test1,tag1=t1,tag2=t2 field4=2,field5=2i32,field6=2f 2"; |
112 | | -connection.publish(DATABASE, payload.getBytes(), QoS.AT_LEAST_ONCE, false); |
113 | | -Thread.sleep(10); |
114 | | - |
115 | | -payload = "# It's a remark\n " + "test1,tag1=t1,tag2=t2 field4=2,field5=2i32,field6=2f 6"; |
116 | | - connection.publish(DATABASE + "/myTopic", payload.getBytes(), QoS.AT_LEAST_ONCE, false); |
117 | | - Thread.sleep(10); |
118 | | - |
119 | | -//batch write example |
120 | | -payload = |
121 | | - "test1,tag1=t1,tag2=t2 field7=t,field8=T,field9=true 3 \n " |
122 | | - + "test1,tag1=t1,tag2=t2 field7=f,field8=F,field9=FALSE 4"; |
123 | | -connection.publish(DATABASE + "/myTopic", payload.getBytes(), QoS.AT_LEAST_ONCE, false); |
124 | | -Thread.sleep(10); |
125 | | - |
126 | | -//batch write example |
127 | | -payload = |
128 | | - "test1,tag1=t1,tag2=t2 attr1=a1,attr2=a2 field1=\"fieldValue1\",field2=1i,field3=1u 4 \n " |
129 | | - + "test1,tag1=t1,tag2=t2 field4=2,field5=2i32,field6=2f 5"; |
130 | | -connection.publish(DATABASE + "/myTopic", payload.getBytes(), QoS.AT_LEAST_ONCE, false); |
131 | | -Thread.sleep(10); |
132 | | - |
133 | | -connection.disconnect(); |
134 | | - ``` |
135 | | - |
136 | | - |
137 | | - |
138 | | -## 6. Customize your MQTT Message Format |
139 | | - |
140 | | -If you do not like the above Line format, you can customize your MQTT Message format by just writing several lines |
141 | | -of codes. An example can be found in [example/mqtt-customize](https://github.com/apache/iotdb/tree/master/example/mqtt-customize) project. |
142 | | - |
143 | | -Steps: |
144 | | -1. Create a java project, and add dependency: |
145 | | -```xml |
146 | | - <dependency> |
147 | | - <groupId>org.apache.iotdb</groupId> |
148 | | - <artifactId>iotdb-server</artifactId> |
149 | | - <version>2.0.4-SNAPSHOT</version> |
150 | | - </dependency> |
151 | | -``` |
152 | | -2. Define your implementation which implements `org.apache.iotdb.db.protocol.mqtt.PayloadFormatter` |
153 | | - e.g., |
154 | | - |
155 | | -```java |
156 | | -package org.apache.iotdb.mqtt.server; |
157 | | - |
158 | | -import io.netty.buffer.ByteBuf; |
159 | | -import org.apache.iotdb.db.protocol.mqtt.Message; |
160 | | -import org.apache.iotdb.db.protocol.mqtt.PayloadFormatter; |
161 | | - |
162 | | -import java.nio.charset.StandardCharsets; |
163 | | -import java.util.ArrayList; |
164 | | -import java.util.Arrays; |
165 | | -import java.util.List; |
166 | | - |
167 | | -public class CustomizedLinePayloadFormatter implements PayloadFormatter { |
168 | | - |
169 | | - @Override |
170 | | - public List<Message> format(String topic, ByteBuf payload) { |
171 | | - // Suppose the payload is a line format |
172 | | - if (payload == null) { |
173 | | - return null; |
174 | | - } |
175 | | - |
176 | | - String line = payload.toString(StandardCharsets.UTF_8); |
177 | | - // parse data from the line and generate Messages and put them into List<Meesage> ret |
178 | | - List<Message> ret = new ArrayList<>(); |
179 | | - // this is just an example, so we just generate some Messages directly |
180 | | - for (int i = 0; i < 3; i++) { |
181 | | - long ts = i; |
182 | | - TableMessage message = new TableMessage(); |
183 | | - |
184 | | - // Parsing Database Name |
185 | | - message.setDatabase("db" + i); |
186 | | - |
187 | | - //Parsing Table Names |
188 | | - message.setTable("t" + i); |
189 | | - |
190 | | - // Parsing Tags |
191 | | - List<String> tagKeys = new ArrayList<>(); |
192 | | - tagKeys.add("tag1" + i); |
193 | | - tagKeys.add("tag2" + i); |
194 | | - List<Object> tagValues = new ArrayList<>(); |
195 | | - tagValues.add("t_value1" + i); |
196 | | - tagValues.add("t_value2" + i); |
197 | | - message.setTagKeys(tagKeys); |
198 | | - message.setTagValues(tagValues); |
199 | | - |
200 | | - // Parsing Attributes |
201 | | - List<String> attributeKeys = new ArrayList<>(); |
202 | | - List<Object> attributeValues = new ArrayList<>(); |
203 | | - attributeKeys.add("attr1" + i); |
204 | | - attributeKeys.add("attr2" + i); |
205 | | - attributeValues.add("a_value1" + i); |
206 | | - attributeValues.add("a_value2" + i); |
207 | | - message.setAttributeKeys(attributeKeys); |
208 | | - message.setAttributeValues(attributeValues); |
209 | | - |
210 | | - // Parsing Fields |
211 | | - List<String> fields = Arrays.asList("field1" + i, "field2" + i); |
212 | | - List<TSDataType> dataTypes = Arrays.asList(TSDataType.FLOAT, TSDataType.FLOAT); |
213 | | - List<Object> values = Arrays.asList("4.0" + i, "5.0" + i); |
214 | | - message.setFields(fields); |
215 | | - message.setDataTypes(dataTypes); |
216 | | - message.setValues(values); |
217 | | - |
218 | | - //// Parsing timestamp |
219 | | - message.setTimestamp(ts); |
220 | | - ret.add(message); |
221 | | - } |
222 | | - return ret; |
223 | | - } |
224 | | - |
225 | | - @Override |
226 | | - public String getName() { |
227 | | - // set the value of mqtt_payload_formatter in iotdb-system.properties as the following string: |
228 | | - return "CustomizedLine"; |
229 | | - } |
230 | | -} |
231 | | -``` |
232 | | -3. modify the file in `src/main/resources/META-INF/services/org.apache.iotdb.db.protocol.mqtt.PayloadFormatter`: |
233 | | - clean the file and put your implementation class name into the file. |
234 | | - In this example, the content is: `org.apache.iotdb.mqtt.server.CustomizedLinePayloadFormatter` |
235 | | -4. compile your implementation as a jar file: `mvn package -DskipTests` |
236 | | - |
237 | | - |
238 | | -Then, in your server: |
239 | | -1. Create ${IOTDB_HOME}/ext/mqtt/ folder, and put the jar into this folder. |
240 | | -2. Update configuration to enable MQTT service. (`enable_mqtt_service=true` in `conf/iotdb-system.properties`) |
241 | | -3. Set the value of `mqtt_payload_formatter` in `conf/iotdb-system.properties` as the value of getName() in your implementation |
242 | | - , in this example, the value is `CustomizedLine` |
243 | | -4. Launch the IoTDB server. |
244 | | -5. Now IoTDB will use your implementation to parse the MQTT message. |
245 | | - |
246 | | -More: the message format can be anything you want. For example, if it is a binary format, |
247 | | -just use `payload.forEachByte()` or `payload.array` to get bytes content. |
248 | | - |
249 | | - |
250 | | -## 7. Caution |
251 | | - |
252 | | -To avoid compatibility issues caused by a default client_id, always explicitly supply a unique, non-empty client_id in every MQTT client. |
253 | | -Behavior varies when the client_id is missing or empty. Common examples: |
254 | | -1. Explicitly sending an empty string |
255 | | -• MQTTX: When client_id="", IoTDB silently discards the message. |
256 | | -• mosquitto_pub: When client_id="", IoTDB receives the message normally. |
257 | | -2. Omitting client_id entirely |
258 | | -• MQTTX: IoTDB accepts the message. |
259 | | -• mosquitto_pub: IoTDB rejects the connection. |
260 | | -Therefore, explicitly assigning a unique, non-empty client_id is the simplest way to eliminate these discrepancies and ensure reliable message delivery. |
| 23 | +--> |
0 commit comments