基于mqtt协议的消息推送服务器,基于 MQTT 协议的推送服务

一、简述

MQTT(Message Queuing Telemetry Transport，消息队列遥测传输协议)，是一种基于发布/订阅(publish/subscribe)模式的“轻量级”通讯协议，该协议构建于TCP/IP协议上，由IBM在1999年发布。MQTT最大优点在于，可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。作为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有较广泛的应用。

MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器(M2M)通信和物联网(IoT)。其在，通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。

二、apache-Apollo 服务器搭建

1.将下载后的 压缩包进行解压:

apollo.png

2.打开 cmd 运行 bin/apollo.cmd

apollo.png

3.创建一个服务器实例

image.png

4.在服务中打开 apollo 服务

image.png

初始账号: admin 密码:password

6.进入apollo控制台

image.png

进入此页面说明服务器成功搭建

三、服务端推送和客户端监听

SeverClass

package com.mqtt;

import org.eclipse.paho.client.mqttv3.MqttClient;

import org.eclipse.paho.client.mqttv3.MqttConnectOptions;

import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;

import org.eclipse.paho.client.mqttv3.MqttException;

import org.eclipse.paho.client.mqttv3.MqttMessage;

import org.eclipse.paho.client.mqttv3.MqttPersistenceException;

import org.eclipse.paho.client.mqttv3.MqttTopic;

import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class ServerMQTT {

//tcp://MQTT安装的服务器地址:MQTT定义的端口号

public static final String HOST = "tcp://0.0.0.0:61613";

//定义一个主题

public static final String TOPIC = "message";

//定义MQTT的ID，可以在MQTT服务配置中指定

private static final String clientid = "server11";

private MqttClient client;

private MqttTopic topic11;

private String userName = "admin"; //非必须

private String passWord = "password"; //非必须

private MqttMessage message;

/**

* 构造函数

* @throws MqttException

*/

public ServerMQTT() throws MqttException {

// MemoryPersistence设置clientid的保存形式，默认为以内存保存

client = new MqttClient(HOST, clientid, new MemoryPersistence());

connect();

}

/**

* 用来连接服务器

*/

private void connect() {

MqttConnectOptions options = new MqttConnectOptions();

options.setCleanSession(false);

options.setUserName(userName);

options.setPassword(passWord.toCharArray());

// 设置超时时间

options.setConnectionTimeout(10);

// 设置会话心跳时间

options.setKeepAliveInterval(20);

try {

client.setCallback(new PushCallback());

client.connect(options);

topic11 = client.getTopic(TOPIC);

} 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);

token.waitForCompletion();

System.out.println("message is published completely! "

+ token.isComplete());

}

/**

* 启动入口

* @param args

* @throws MqttException

*/

public static void main(String[] args) throws MqttException {

ServerMQTT server = new ServerMQTT();

server.message = new MqttMessage();

server.message.setQos(1); //保证消息能到达一次

server.message.setRetained(true);

server.message.setPayload("这是服务器发出的信息的内容".getBytes());

server.publish(server.topic11 , server.message);

System.out.println(server.message.isRetained() + "------ratained状态");

}

}

ClientClass

package com.mqtt;

import java.util.concurrent.ScheduledExecutorService;

import org.eclipse.paho.client.mqttv3.MqttClient;

import org.eclipse.paho.client.mqttv3.MqttConnectOptions;

import org.eclipse.paho.client.mqttv3.MqttException;

import org.eclipse.paho.client.mqttv3.MqttMessage;

import org.eclipse.paho.client.mqttv3.MqttTopic;

import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class ClientMQTT {

public static final String HOST = "tcp://0.0.0.0:61613";

public static final String TOPIC1 = "message";

private static final String clientid = "client11";

private MqttClient client;

private MqttConnectOptions options;

private String userName = "admin"; //非必须

private String passWord = "password"; //非必须

@SuppressWarnings("unused")

private ScheduledExecutorService scheduler;

private void start() {

try {

// host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存

client = new MqttClient(HOST, clientid, new MemoryPersistence());

// MQTT的连接设置

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);

// 设置回调

client.setCallback(new PushCallback());

MqttTopic topic = client.getTopic(TOPIC1);

//setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息

//options.setWill(topic, "close".getBytes(), 2, true);

client.connect(options);

//订阅消息

//订阅消息

client.subscribe(TOPIC1, 1);

} catch (Exception e) {

e.printStackTrace();

}

}

public static void main(String[] args) throws MqttException {

ClientMQTT client = new ClientMQTT();

client.start();

}

}

server

server

client

client