emqttd java代码测试

柔情只为你懂 2022-08-09 09:13 165阅读 0赞

emqttd 集群在上一节我们已经配置完毕。emqttd默认占用1883端口用于MQTT连接，8083端口用于HTTP接口，通过HTTP接口从应用程序向MQTT客户端发布消息。下面我们通过java实现消息的发布和订阅。

**1.pom**

使用mqtt-client java客户端，添加依赖jar

<dependency>
    			<groupId>org.fusesource.mqtt-client</groupId>
    			<artifactId>mqtt-client</artifactId>
    			<version>1.12</version>
    		</dependency>

2.代码实现

订阅者订阅node2节点端口

package mqtt;
    
    import java.net.URISyntaxException;
    
    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 Client {
    	
    	private final static String CONNECTION_STRING = "tcp://192.168.1.91:1883";  
        private final static boolean CLEAN_START = true;  
        private final static short KEEP_ALIVE = 30;// 低耗网络，但是又需要及时获取数据，心跳30s  
        private final static String CLIENT_ID = "client";  
        public static Topic[] topics = {   
            new Topic("mqtt/aaa", QoS.EXACTLY_ONCE), //  2 只有一次
            new Topic("mqtt/bbb", 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 static void main(String[] args) {
    		  // 创建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();  
                connection.connect();  
                connection.subscribe(topics);  
                while (true) {  
                    Future<Message> futrueMessage = connection.receive();  
                    Message message = futrueMessage.await();    
                    System.out.println("MQTTFutureClient.Receive Message " + "Topic Title :" + message.getTopic() + " context :"  
                            + String.valueOf(message.getPayloadBuffer()));  
                }  
            } catch (URISyntaxException e) {  
                e.printStackTrace();  
            } catch (Exception e) {  
                e.printStackTrace();  
            } finally {  
            }  
    	}
    
    }

发布者通过node1发布消息

package mqtt;
    
    import java.net.URISyntaxException;
    
    import org.fusesource.mqtt.client.FutureConnection;
    import org.fusesource.mqtt.client.MQTT;
    import org.fusesource.mqtt.client.QoS;
    import org.fusesource.mqtt.client.Topic;
    
    /**
     * 服务端发布消息
     */
    public class Server {
    
    	private final static String CONNECTION_STRING = "tcp://192.168.1.90:1883";
    	private final static boolean CLEAN_START = true;
    	private final static String CLIENT_ID = "server";
    	private final static short KEEP_ALIVE = 30;// 低耗网络，但是又需要及时获取数据，心跳30s
    
    	public static Topic[] topics = { 
    	new Topic("mqtt/aaa", QoS.EXACTLY_ONCE), // 2  只有一次
            new Topic("mqtt/bbb", 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;// 发送最大缓冲
    
    	public static void main(String[] args) {
    		 MQTT mqtt = new MQTT();    
             try {    
                 //==MQTT设置说明  
                 //设置服务端的ip    
                 mqtt.setHost(CONNECTION_STRING);    
                 //连接前清空会话信息 ,若设为false，MQTT服务器将持久化客户端会话的主体订阅和ACK位置，默认为true  
                 mqtt.setCleanSession(CLEAN_START);    
                 //设置心跳时间 ,定义客户端传来消息的最大时间间隔秒数，服务器可以据此判断与客户端的连接是否已经断开，从而避免TCP/IP超时的长时间等待  
                 mqtt.setKeepAlive(KEEP_ALIVE);    
                 //设置客户端id,用于设置客户端会话的ID。在setCleanSession(false);被调用时，MQTT服务器利用该ID获得相应的会话。  
                 //此ID应少于23个字符，默认根据本机地址、端口和时间自动生成  
                 mqtt.setClientId(CLIENT_ID);  
                   
                 //==失败重连接设置说明  
                  //设置重新连接的次数 ,客户端已经连接到服务器，但因某种原因连接断开时的最大重试次数，超出该次数客户端将返回错误。-1意为无重试上限，默认为-1  
                 mqtt.setReconnectAttemptsMax(RECONNECTION_ATTEMPT_MAX);    
                 //设置重连的间隔时间  ,首次重连接间隔毫秒数，默认为10ms  
                 mqtt.setReconnectDelay(RECONNECTION_DELAY);    
                 //设置socket发送缓冲区大小，默认为65536（64k）  
                 mqtt.setSendBufferSize(SEND_BUFFER_SIZE);    
                 设置发送数据包头的流量类型或服务类型字段，默认为8，意为吞吐量最大化传输  
                 mqtt.setTrafficClass(8);  
                   
                 //==带宽限制设置说明  
                 mqtt.setMaxReadRate(0);//设置连接的最大接收速率，单位为bytes/s。默认为0，即无限制  
                 mqtt.setMaxWriteRate(0);//设置连接的最大发送速率，单位为bytes/s。默认为0，即无限制  
                   
                 //使用Future创建连接     
                 final FutureConnection connection= mqtt.futureConnection();    
                 connection.connect();    
                 int count=1;    
                 while(true){    
                     count++;    
                     // 用于发布消息，目前手机段不需要向服务端发送消息    
                     //主题的内容    
                     String message="Hello "+count+" MQTT...";    
                     String topic = "mqtt/bbb";    
                     connection.publish(topic, message.getBytes(), QoS.AT_LEAST_ONCE,    
                             false);    
                     System.out.println("MQTTFutureServer.publish Message "+"Topic Title :"+topic+" context :"+message);    
                 }    
             } catch (URISyntaxException e) {    
                 e.printStackTrace();    
             } catch (Exception e) {    
                 e.printStackTrace();    
             }finally{
             }
    	}
    
    }

运行代码测试正常接收、发布消息。  
也可以通过http请求的方式发布消息，如下：

curl -v --basic -u user:passwd -d "qos=2&retain=0&topic=mqtt/aaa&message=hello" -k http://192.168.1.90:8083/mqtt/publish

<table style="border-collapse:collapse; border-spacing:0px; margin-top:0px; display:block; width:640px; overflow:auto; word-break:keep-all; color:rgb(51,51,51); font-family:'Helvetica Neue',Helvetica,'Segoe UI',Arial,freesans,sans-serif; font-size:16px; line-height:25.6000003814697px; margin-bottom:0px!important"> 
 <thead style=""> 
  <tr style="border-top-width:1px; border-top-style:solid; border-top-color:rgb(204,204,204)"> 
   <th style="padding:6px 13px; border:1px solid rgb(221,221,221)">Name</th> 
   <th style="padding:6px 13px; border:1px solid rgb(221,221,221)">Description</th> 
  </tr> 
 </thead> 
 <tbody style=""> 
  <tr style="border-top-width:1px; border-top-style:solid; border-top-color:rgb(204,204,204)"> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221)">client</td> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221)">ClientId</td> 
  </tr> 
  <tr style="border-top-width:1px; border-top-style:solid; border-top-color:rgb(204,204,204); background-color:rgb(248,248,248)"> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221)">qos</td> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221)">QoS(0, 1, 2)</td> 
  </tr> 
  <tr style="border-top-width:1px; border-top-style:solid; border-top-color:rgb(204,204,204)"> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221)">retain</td> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221)">Retain(0, 1)</td> 
  </tr> 
  <tr style="border-top-width:1px; border-top-style:solid; border-top-color:rgb(204,204,204); background-color:rgb(248,248,248)"> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221)">topic</td> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221)">Topic</td> 
  </tr> 
  <tr style="border-top-width:1px; border-top-style:solid; border-top-color:rgb(204,204,204)"> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221)">message</td> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221)">Message</td> 
  </tr> 
 </tbody> 
</table>

没有添加认证，用户密码可以随便填写。

**RETAIN(保持)**  
仅针对PUBLISH消息。不同值，不同含义：  
1：表示发送的消息需要一直持久保存（不受服务器重启影响），不但要发送给当前的订阅者，并且以后新来的订阅了此Topic name的订阅者会马上得到推送。  
备注：新来乍到的订阅者，只会取出最新的一个RETAIN flag = 1的消息推送。  
0：仅仅为当前订阅者推送此消息。  
假如服务器收到一个空消息体(zero-length payload)、RETAIN = 1、已存在Topic name的PUBLISH消息，服务器可以删除掉对应的已被持久化的PUBLISH消息。

参考：[https://github.com/emqtt/emqttd/wiki/HTTP%20Publish][https_github.com_emqtt_emqttd_wiki_HTTP_20Publish]

[https_github.com_emqtt_emqttd_wiki_HTTP_20Publish]: https://github.com/emqtt/emqttd/wiki/HTTP%20Publish