MQTT服务（Apollo）搭建和客户端使用

公司最近处理基于OPC UA 协议服务kepserver发送的消息，目的是要消费这些消息，那符合发布订阅模式的MQTT就在合适不过了，MQTT是基于tcp的消息发送，目前JAVA方面有两种实现，分别是mqtt-fuse和mqtt-paho，优点或者适用性自行研究，MQTT也是一个轻量级的消息体架构，安装方便，易于管理，而且还支持主从节点配置。

MQTT服务搭建

mqtt-server支持win/linux等系统的安装，下文基于linux进行说明。
1.下载安装包
在官网（http://archive.apache.org/dist/activemq/activemq-apollo/1.7.1/）上下载apache-apollo-1.7.1-unix-distro.tar.gz，目前最新版就是1.7.1 ，上传并解压到指定目录下。
如果服务器有网可以直接下载
wget http://archive.apache.org/dist/activemq/activemq-apollo/1.7.1/apache-apollo-1.7.1-unix-distro.tar.gz
下载完后直接解压到对应文件夹，如图所示：

image.png

2.创建实例
进入到bin目录中，执行：./apollo create mybroker，创建成功后如下图。

image.png

3.完成后同目录下会看到创建的实例目录文件夹

image.png

cd 进入实例目录如下：

image.png
修改MQTT的访问保护模式，可以让其他客户端访问

<web_admin bind="http://127.0.0.1:61680"/> 
<web_admin bind="https://127.0.0.1:61681"/> 将其改成0.0.0.0

image.png

开启白名单（可以不开启，如果有安全考虑还是建议开启）

MQTT服务支持白名单，黑名单，设置步骤很简单，就是cd到实例的etc目录下
编辑login.config文件，去掉white_list_file注释，就打开白名单限制了，黑名单同理

image.png

同目录下，有个white-list.txt文件，添加受信任的IP就可以了
配置完成准备启动服务
进入到上截图的bin中，执行：./apollo-broker run，注意：每个实例需要单独启动，成功如下所示

image.png

启动后，我们发现可以通过http://IP:61680访问web页面，注意用户名和密码分别是：admin/password

image.png

注意在connectors选项卡中可以查看当前tcp、tls、ws等是否有对端的mqtt-client相连

MQTT客户端

下载org.eclipse.paho.ui.app-1.0.2-win32.win32.x86_64，解压直接运行Paho.exe就行，IP，输入用户名，密码（“选项”页中填写），点击连接，连接状态显示已连接表示连接成功。就可以发布主题，订阅主题了。

image.png

java代码通过MqttClient客户端订阅，推送服务

推送消息

package com.longi.controls.client;

import com.longi.controls.utils.CommonUtil;
import com.longi.controls.utils.ConnectUtil;
import org.apache.log4j.Logger;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;

/**
 * @version 1.0
 * @CalssName SendClient
 * @Author sunke5
 * @Date 2019-11-19 16:34
 */
public class SendClient {

    static final Logger log = Logger.getLogger(SendClient.class);

    private static String host = "tcp://ip:61613";
    private static String userName = "admin";
    private static String passWord = "password";
    private static MqttClient client;
    private static final String clientId = "client22";

    //本次测试订阅的主题：test
    private static String topicStr = "test";

    public static void initMQTT() throws MqttException {
        //MemoryPersistence设置clientId的保存形式，默认为以内存保存
        if(client==null){
            client = new MqttClient(host, clientId,new MqttDefaultFilePersistence());
        }
        MqttTopic topic = client.getTopic(topicStr);
        //连接MQTT服务
        client = ConnectUtil.initConnection(topic,userName,passWord,client);

        //订阅
        client.subscribe(topicStr, 1);
        //取消订阅
        //client.unsubscribe(topicStr);

    }

    public static void main(String[] args) throws MqttException{
        initMQTT();
        while (true){
            try {
                Thread.sleep(5000);
                ConnectUtil.PushMsg(client,"客户端发送："+System.currentTimeMillis());
            }catch (Exception e){
                CommonUtil.getExceptionDetail(e);
            }
        }
    }

}

订阅消息

package com.longi.controls.client;

import com.longi.controls.utils.CommonUtil;
import com.longi.controls.utils.ConnectUtil;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/**
 * @version 1.0
 * @CalssName ReceiveClient
 * @Author sunke5
 * @Date 2019-11-19 16:33
 */
public class ReceiveClient {

    private static String host = "tcp://ip:61613";
    private static String userName = "admin";
    private static String passWord = "password";

    private static MqttClient client;
    private static final String clientId = "client1111";
    private static String TOPIC = "BS";

    private void start() {
        try {
            // host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
            if(client==null){
                client = new MqttClient(host, clientId, new MemoryPersistence());
            }
            MqttTopic topic = client.getTopic(TOPIC);
            //连接MQTT服务
            client = ConnectUtil.initConnection(topic,userName,passWord,client);
            //订阅消息
            int[] Qos  = {1};
            String[] topic1 = {TOPIC};
            client.subscribe(topic1, Qos);

        } catch (Exception e) {
            CommonUtil.getExceptionDetail(e);
        }
    }

    public static void main(String[] args) throws MqttException {
        ReceiveClient client = new ReceiveClient();
        client.start();
    }
}

回调类

package com.longi.controls.service;

import com.longi.controls.utils.CommonUtil;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.fusesource.hawtbuf.UTF8Buffer;


/**
 * MQTT订阅回调类
 *
 * @version 1.0
 * @CalssName PushCallback
 * @Author sunke5
 * @Date 2019-11-18 16:53
 */
public class PushCallback implements MqttCallback {

    @Override
    public void connectionLost(Throwable cause) {
        // 连接丢失后，一般在这里面进行重连
        System.out.println("connectionLost----------");
        try {
            //initMQTT();
            System.out.println("连接断开，重连");
        }catch (Exception e){
            CommonUtil.getExceptionDetail(e);
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        System.out.println("deliveryComplete---------" + token.isComplete());
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        // subscribe后得到的消息会执行到这里面
        System.out.println("接收消息主题 : " + topic);
        System.out.println("接收消息Qos : " + message.getQos());
        System.out.println("接收消息内容 : " + new String(message.getPayload(),"UTF-8"));
    }
}

连接工具类

package com.longi.controls.utils;

import com.longi.controls.service.PushCallback;
import org.eclipse.paho.client.mqttv3.*;

/**
 * @version 1.0
 * @CalssName ConnectUtil
 * @Author sunke5
 * @Date 2019-11-19 11:36
 */
public class ConnectUtil {

    //自己写发布消息的方法，然后循环调用。
    public static MqttConnectOptions setOptions(MqttTopic topic,String userName, String passWord){
        MqttConnectOptions options = new MqttConnectOptions();
        //设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，
        //这里设置为true表示每次连接到服务器都以新的身份连接
        options.setCleanSession(false);
        //设置连接的用户名
        options.setUserName(userName);
        //设置连接的密码
        options.setPassword(passWord.toCharArray());
        // 设置超时时间 单位为秒
        options.setConnectionTimeout(10);
        // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
        options.setKeepAliveInterval(20);

        //setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息
        options.setWill(topic, "close".getBytes(), 2, true);

        return options;
    }

    public static MqttClient initConnection(MqttTopic topic,String userName, String passWord, MqttClient client) throws MqttException {

        MqttConnectOptions options = setOptions(topic,userName,passWord);
        //回调
        client.setCallback(new PushCallback());
        //链接
        client.connect(options);
        return client;
    }

    //自己写发布消息的方法，然后循环调用。
    public static void PushMsg(MqttClient client,String msg){
        MqttMessage m = new MqttMessage();
            /**  
             *  qos ＝ 0：最多一次的传输
             *  qos ＝ 1：至少一次的传输
             *  qos ＝ 2：只有一次的传输
             */
        m.setQos(1);
        m.setRetained(true);
        m.setPayload(msg.getBytes());
        try {
            client.publish("test", m);
        }catch(Exception e){
            System.out.println("发布消息失败-->"+msg);
            CommonUtil.getExceptionDetail(e);
        }
    }
}

服务端给客户端推送消息

package com.longi.controls.service;

import com.longi.controls.utils.CommonUtil;
import com.longi.controls.utils.ConnectUtil;
import org.apache.log4j.Logger;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/**
 * @version 1.0
 * @CalssName Server
 * @Author sunke5
 * @Date 2019-11-19 16:24
 */
public class Server {

    static final Logger log = Logger.getLogger(Server.class);

    private static String HOST = "tcp://ip:61613";
    private static String userName = "admin";
    private static String passWord = "password";
    public static final String TOPIC1 = "test";
    public static final String TOPIC2 = "test11";
    private static final String topicStr = "server";

    private MqttClient client;
    private MqttTopic topic11;
    private MqttTopic topic22;
    private MqttMessage message;

    public Server() throws MqttException {
        // MemoryPersistence设置clientid的保存形式，默认为以内存保存
        client = new MqttClient(HOST, topicStr, new MemoryPersistence());
        //client = new MqttClient(HOST, topicStr, new MqttDefaultFilePersistence());

        MqttTopic topic = client.getTopic(topicStr);
        MqttConnectOptions options = ConnectUtil.setOptions(topic,userName,passWord);
        try {
            client.setCallback(new PushCallback());
            client.connect(options);
            topic11 = client.getTopic(TOPIC1);
            topic22 = client.getTopic(TOPIC2);
        } catch (Exception e) {
            log.info(CommonUtil.getExceptionDetail(e));
        }
    }

    public void publish(MqttTopic topic , MqttMessage message) throws MqttPersistenceException,
            MqttException {
        MqttDeliveryToken token = topic.publish(message);
        token.waitForCompletion();
        log.info("message is published completely! "+ token.isComplete());
    }

    public static void main(String[] args) throws MqttException {
        Server server = new Server();

        server.message = new MqttMessage();
        server.message.setQos(2);
        server.message.setRetained(true);
        server.message.setPayload("给客户端11推送的信息".getBytes());
        server.publish(server.topic11 , server.message);

        server.message = new MqttMessage();
        server.message.setQos(2);
        server.message.setRetained(true);
        server.message.setPayload("给客户端22推送的信息".getBytes());
        server.publish(server.topic22 , server.message);

        log.info(server.message.isRetained() + "------ratained状态");
    }
}

测试没有发现什么问题，有2个待验证的点

主从节点配置
消息的上限

apollo.xml 配置详解参考https://blog.csdn.net/IamDaodaoLi/article/details/79824220

MQTT服务（Apollo）搭建和客户端使用

MQTT服务搭建

MQTT客户端

java代码 通过MqttClient客户端 订阅，推送服务

java代码通过MqttClient客户端订阅，推送服务