# emqx开启CoAP网关
请参考【https://blog.csdn.net/chenhz2284/article/details/139562749?spm=1001.2014.3001.5502】
# 写一个emqx的客户端程序，不断地往topic【server/1】发消息
【pom.xml】
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
<version>2.3.12.RELEASE</version>
</dependency>
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.mqttv5.client</artifactId>
<version>1.2.5</version>
</dependency>
<dependency>
<groupId>com.alibaba.fastjson2</groupId>
<artifactId>fastjson2</artifactId>
<version>2.0.49</version>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>1.2.11</version>
</dependency>
【MyDemo2MqttCallback.java】
package com.chz.myMqttV5.demo2;
@Slf4j
public class MyDemo2MqttCallback implements MqttCallback
{
private String clientId;
public MyDemo2MqttCallback(String clientId)
{
this.clientId = clientId;
}
public void connectComplete(boolean reconnect, String serverURI) {
log.info("{}::MyMqttCallback, reconnect={}, serverURI={}", clientId, reconnect, serverURI);
}
public void disconnected(MqttDisconnectResponse disconnectResponse) {
log.info("{}::disconnected, disconnectResponse={}", clientId, disconnectResponse.getReasonString());
}
public void deliveryComplete(IMqttToken token) {
log.info("{}::deliveryComplete, disconnectResponse={}", clientId, token.isComplete());
}
public void messageArrived(String topic, MqttMessage message) throws Exception {
log.info("{}::messageArrived, topic={}, qos={}, message={}", clientId, topic, message.getQos(), new String(message.getPayload()));
}
public void mqttErrorOccurred(MqttException exception) {
log.info("{}::mqttErrorOccurred, disconnectResponse={}", clientId, exception.getMessage());
}
public void authPacketArrived(int reasonCode, MqttProperties properties) {
log.info("{}::authPacketArrived, reasonCode={}", clientId, reasonCode);
}
}
【MyDemo2MqttV5Sender.java】
package com.chz.myMqttV5.demo2;
@Slf4j
public class MyDemo2MqttV5Sender {
private static String clientId = MyDemo2MqttV5Sender.class.getSimpleName();
public static void main(String[] args) throws InterruptedException {
String broker = "tcp://192.168.44.228:1883";
int subQos = 1;
int pubQos = 1;
String msg;
try {
MqttClient client = new MqttClient(broker, clientId);
MqttConnectionOptions options = new MqttConnectionOptions();
client.setCallback(new MyDemo2MqttCallback(clientId));
client.connect(options);
client.subscribe("device/#", subQos);
for(int i=0; i<10000; i++){
msg = "I am "+clientId+":" + i + "\r\n";
MqttMessage message = new MqttMessage(msg.getBytes());
message.setQos(pubQos);
client.publish("server/1", message);
Thread.sleep(3000L);
}
client.disconnect();
client.close();
} catch (MqttException e) {
e.printStackTrace();
}
}
}
启动【MyDemo2MqttV5Sender】，每隔3秒不断地往topic【server/1】发消息。
# 下面使用libcoap对emqx进行测试
创建与emqx的连接，执行：
coap-client -m post -e "" "coap://192.168.44.228/mqtt/connection?clientid=123&username=admin&password=public"
注意url里面带有clientid，下面还会继续用到。
返回的那个数据是token，下面也会用到。
向emqx发送消息，执行：
coap-client -m post -e "Hi, this is libcoap" "coap://192.168.44.228/ps/device/1?clientid=123&token=3167748641"
注意url参数里面带有token，clientid也必须跟上面建立连接时候的clientid一样
从截图可以看到mqtt的消息订阅者接收到libcoap发出的消息了。
使用libcoap从emqx订阅消息，执行：
coap-client -m get -s 6000 -O 6,0x00 -o - -T "obstoken" "coap://192.168.44.228/ps/server/%23?clientid=123&token=3167748641"
注意订阅的主题【server/%23】是经过url编码的，编码之前的主题名是【server/#】
从截图可以看到消费到消息了
# 总结
从上面的测试可以看出，只要emqx安装了CoAP网关，通过CoAP协议也是可以跟emqx进行通讯，发布和订阅消息的。不过尽量使用mqtt协议比较好
