Mqtt注解式开发,轻便emqx消息服务器,发布订阅轻松搞定

MQTT啊EMQX啊这些概念性问题,博主不再复制粘贴了,需要的老铁直接去百度吧.
直接上测试老代码:

模拟服务端发

    /**
     * 构造函数
     * 
     * @throws MqttException
     */
    public PahoServer() throws MqttException {
        // MemoryPersistence设置clientid的保存形式，默认为以内存保存
        client = new MqttClient(HOST, clientid, new MemoryPersistence());        
        connect();
    }

    /**
     * 用来连接服务器
     */
    private void connect() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        options.setUserName(userName);
        options.setPassword(passWord.toCharArray());
        // 设置超时时间
        options.setConnectionTimeout(60);
        // 设置会话心跳时间
        options.setKeepAliveInterval(30);
        try {
            client.setCallback(new PushCallback());
            topic11 = client.getTopic(TOPIC);
            client.connect(options);
            int[] Qos = { 0 };
            String[] topic2 = { TOPIC2 };
            client.subscribe(topic2, Qos);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 
     * @param topic
     * @param message
     * @throws MqttPersistenceException
     * @throws MqttException
     */
    public void publish(MqttTopic topic, MqttMessage message) throws MqttPersistenceException, MqttException {
        MqttDeliveryToken token = topic.publish(message);
        System.out.println("这是服务端发送的消息:"+token.getMessage());
        token.waitForCompletion();
//        client.disconnect();
        System.out.println("message is published completely! " + token.isComplete());
    }

模拟客户端收

// host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
        client = new MqttClient(HOST, clientid, new MemoryPersistence());
        client.setManualAcks(false);
        // MQTT的连接设置
        options = new MqttConnectOptions();
        // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，设置为true表示每次连接到服务器都以新的身份连接
        options.setCleanSession(true);
        // 设置连接的用户名
        options.setUserName(userName);
        // 设置连接的密码
        options.setPassword(passWord.toCharArray());
        // 设置超时时间 单位为秒
        options.setConnectionTimeout(60);
        // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
        options.setKeepAliveInterval(30);
        // 设置回调
        client.setCallback(new MqttCallback() {
            public void connectionLost(Throwable cause) {
                // 连接丢失后，一般在这里面进行重连
                System.out.println("连接断开，正在重连");
                while (true) {
                    try {// 如果没有发生异常说明连接成功，如果发生异常，则死循环
                        Thread.sleep(1000);
                        System.out.println("正在重新连接...");
                        NewPaho newPaho = new NewPaho();
                        newPaho.start();
                        break;
                    } catch (Exception e) {
                        continue;
                    }
                }
            }
            public void deliveryComplete(IMqttDeliveryToken token) {
                try {
                    token.waitForCompletion();
                } catch (MqttException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
                System.out.println("deliveryComplete---------" + token.isComplete());
            }

            public void messageArrived(String topic, MqttMessage message) throws Exception {
                try {
                    str = message.toString();
//                    System.out.println(" 从服务器收到的消息为:" + message.toString());
                      System.out.println("接收消息主题 : " + topic);  
                        System.out.println("接收消息Qos : " + message.getQos());  
                        System.out.println("接收消息内容 : " + new String(message.getPayload()));
                        MqttMessage message2 = new MqttMessage();
                        message2.setQos(0);
                        message2.setRetained(false);
                        message2.setPayload(resc().getBytes());
                        client.publish("mqtt/abc", message2);

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
         MqttTopic topic = client.getTopic(TOPIC1);
        // setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息
         options.setWill(topic, "客户端掉线了啊".getBytes(), 0, false); //遗嘱
         client.connect(options);
        System.out.println("客户端连接成功......连接地址:"+client.getCurrentServerURI());
        // 订阅消息
        int[] Qos = { 1 };
        String[] topic1 = { TOPIC1 };
        client.subscribe(topic1, Qos);

从上述部分代码可以看出,每次都要频繁的setCallback,重复性的操作很多,所以决定将其,封装.
封装代码
之后只需要在配置文件配置连接地址就行了.然后再实际代码中应用如下

    @MqttSubscribe("terminal/dataUp/{tid}")
    public void dataUpSub(@PathValue("tid") Integer tid, @Payload String json){
        System.out.println("dataUp"+tid);
        System.out.println(ZipUtils.gunzip(json));
    }
    @MqttSubscribe("terminal/dataUpEvent/{tid}")
    public void dataUpEventSub(@PathValue("tid") Integer tid, @Payload String json){
        System.out.println("dataUpEvent"+tid);
        System.out.println(ZipUtils.gunzip(json));
    }
    @MqttSubscribe("terminal/metadata/{tid}")
    public void metadataSub(@PathValue("tid") Integer tid, @Payload(Base64GunzipConverter.class) String json){
        System.out.println("metadata"+tid);
        System.out.println(json);
    }

    @MqttSubscribe("terminal/common/+")
    public void commonSub(@Payload CommonEntity commonEntity){
        System.out.println("common");
        Integer eventId = commonEntity.getEventId();
        Integer status = commonEntity.getStatus();
        if (status == 0) {
            mqttEventQueueService.commonEventSucceeStatus(eventId);
        } else
            mqttEventQueueService.commonEventFailStatus(eventId);
    }

@MqttSubscribe("xxx")代表订阅括xxx主题.
@Payload 代表可以直接将消息内容转为实体,如果不使用该注解的话无法使用,并且字符串会默认是主题.
至于发布private MqttPublisher mqttPublisher;引入该类就行.mqttPublisher.send("xxx",json); 就能直接发布.
是不是很方便.

{
  "groups": [
    {
      "name": "mqtt",
      "type": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties",
      "description": "MQTT配置.",
      "sourceType": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties"
    },
    {
      "name": "mqtt.persistence",
      "type": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties$PersistenceProperties",
      "description": "MqttClientPersistence配置.",
      "sourceType": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties",
      "sourceMethod": "getPersistence()"
    },
    {
      "name": "mqtt.will",
      "type": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties$WillProperties",
      "description": "遗愿相关配置.",
      "sourceType": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties",
      "sourceMethod": "getWill()"
    }
  ],
  "properties": [
    {
      "name": "mqtt.automatic-reconnect",
      "type": "java.lang.Boolean",
      "description": "断开是否重新连接.",
      "sourceType": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties",
      "defaultValue": true
    },
    {
      "name": "mqtt.clean-session",
      "type": "java.lang.Boolean",
      "description": "是否清除会话.",
      "sourceType": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties",
      "defaultValue": true
    },
    {
      "name": "mqtt.client-id",
      "type": "java.lang.String",
      "description": "客户端ID, 默认为MqttAsyncClient.generateClientId().",
      "sourceType": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties"
    },
    {
      "name": "mqtt.connection-timeout",
      "type": "java.lang.Integer",
      "description": "连接超时时间.",
      "sourceType": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties",
      "defaultValue": 30
    },
    {
      "name": "mqtt.disable",
      "type": "java.lang.Boolean",
      "description": "是否禁用.",
      "sourceType": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties",
      "defaultValue": false
    },
    {
      "name": "mqtt.executor-service-timeout",
      "type": "java.lang.Integer",
      "description": "发送超时时间.",
      "sourceType": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties",
      "defaultValue": 10
    },
    {
      "name": "mqtt.keep-alive-interval",
      "type": "java.lang.Integer",
      "description": "KeepAlive 周期.",
      "sourceType": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties",
      "defaultValue": 30
    },
    {
      "name": "mqtt.max-reconnect-delay",
      "type": "java.lang.Integer",
      "description": "最大重连等待时间.",
      "sourceType": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties",
      "defaultValue": 30
    },
    {
      "name": "mqtt.password",
      "type": "java.lang.String",
      "description": "密码.",
      "sourceType": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties"
    },
    {
      "name": "mqtt.persistence.dir",
      "type": "java.lang.String",
      "description": "当 mqtt.persistence.type=file 时使用, 默认System.getProperty(\"user.dir\").",
      "sourceType": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties$PersistenceProperties",
      "defaultValue": "user.dir"
    },
    {
      "name": "mqtt.persistence.type",
      "type": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties$PersistenceType",
      "description": "类型.",
      "sourceType": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties$PersistenceProperties",
      "defaultValue": "memory"
    },
    {
      "name": "mqtt.uri",
      "type": "java.lang.String[]",
      "description": "MQTT服务器地址, 可以配置多个.",
      "sourceType": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties",
      "defaultValue": "tcp:\/\/127.0.0.1:1883"
    },
    {
      "name": "mqtt.username",
      "type": "java.lang.String",
      "description": "用户名.",
      "sourceType": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties"
    },
    {
      "name": "mqtt.will.payload",
      "type": "java.lang.String",
      "description": "遗愿消息内容.",
      "sourceType": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties$WillProperties",
      "defaultValue": "default will msg."
    },
    {
      "name": "mqtt.will.qos",
      "type": "java.lang.Integer",
      "description": "遗愿消息QOS.",
      "sourceType": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties$WillProperties",
      "defaultValue": 0
    },
    {
      "name": "mqtt.will.retained",
      "type": "java.lang.Boolean",
      "description": "遗愿消息是否保留.",
      "sourceType": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties$WillProperties",
      "defaultValue": false
    },
    {
      "name": "mqtt.will.topic",
      "type": "java.lang.String",
      "description": "遗愿主题.",
      "sourceType": "cn.wellsgroup.intelligent.building.cloud.device.manager.mqtt.impl.autoconfigure.MqttProperties$WillProperties"
    }
  ],
  "hints": [
    {
      "name": "mqtt.persistence.type",
      "values": [
        {
          "value": "memory",
          "description": "MemoryPersistence."
        },
        {
          "value": "file",
          "description": "MqttDefaultFilePersistence."
        }
      ]
    }
  ]
}

在项目resourceMETA-INF下面加入spring-configuration-metadata.json这个文件,将上面内容复制进去!sourceType记得改成自己项目包路径

<dependency>
    <groupId>com.github.tocrhz</groupId>
    <artifactId>mqtt-spring-boot-starter</artifactId>
    <version>1.0.0.RELEASE</version>
</dependency>

直接项目中引用该依赖就可以了.具体使用详情,可以进群或者加博主QQ咨询.