linux服务器centos7搭建mqtt服务java实现mqtt通讯
安装mqtt服务实现mqtt通讯文章目录安装mqtt服务实现mqtt通讯前言一、MQTT通讯简介1.结构模式图2.发布和订阅二、搭建mqtt服务器1.安装mqtt服务器2.运行mqtt服务器3.开启端口4.访问emqx可视化界面三、测试mqtt发送消息1.搭建两个mqtt客户端2.发送消息前言最近在开发mqtt相关的硬件通讯，自己用虚拟机在linux环境搭了一个mqtt服务器，实现简单的通讯演示一
·
安装mqtt服务实现mqtt通讯
文章目录
前言
最近在开发mqtt相关的硬件通讯，自己用虚拟机在linux环境搭了一个mqtt服务器，实现简单的通讯演示
一、MQTT通讯简介
1.结构模式图
mqtt通讯有服务端和客户端,服务端统一收集和分发信息。
2.发布和订阅
客户端有发布和订阅两种模式,如mqttClinet1发布的topic=“ASDFG”，mqttClinet2和mqttClient3的订阅topic=“ASDFG”，mqttClinet4的订阅topic=“ASDFGhjkl”，那么当mqttClient1发布一个消息msg=“哈喽!”，那么mqttClient2和mqttClient3的订阅topic和mqttClient1的发布topic一致，将接受到"哈喽!"的消息，mqttClient4就不会接收到任何消息，这就是mqtt的发布和订阅机制。
二、搭建mqtt服务器
1.安装mqtt服务器
可以下载我上传的mqtt安装包:mqtt安装包下载，也可以去官网找适合自己的mqtt安装包:官网emqx多版本mqtt安装包下载，下面我就一我已有的mqtt安装包安装为例进行演示。
进入服务器创建一个emqx文件夹并把emqx.rpm上传到此文件夹下，并安装
[root@localhost ~]# cd /opt/
[root@localhost opt]# mkdir emqx
[root@localhost opt]# cd ./emqx
[root@localhost emqx]# ls
emqx-centos7-4.2.7-x86_64.rpm
[root@localhost emqx]# rpm -ivh emqx-centos7-4.2.7-x86_64.rpm
2.运行mqtt服务器
[root@localhost emqx]# emqx start
上图代表运行成功!
3.开启端口
由于mqtt服务器使用的是1883端口,，同事emqx带可视化界面用到18083端口，确保防火墙将1883和18083这两端口开启,Centos7防火墙端口开启或关闭可以参考我的文章:Centos7防火墙端口开启和关闭
4.访问emqx可视化界面
打开浏览器访问http://主机名:18083
账号是admin，密码是public
进入之后就可以看到注册的mqtt客户端的一些详细信息了
三、测试mqtt发送消息
1.搭建两个mqtt客户端
mqttclient1代码如下
#pom依赖
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
<version>1.2.2</version>
</dependency>
public class ClientMQTT {
public static final String HOST = "tcp://192.168.31.74:1883";
public static final String TOPIC1 = "ASDFG";//订阅地址
private static final String clientid = "client1";
private MqttClient client;
private MqttConnectOptions options;
private String userName = "mqtt1"; //非必须
private String passWord = "mqtt1"; //非必须
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);
//设置断开后重新连接
options.setAutomaticReconnect(true);
// 设置回调
//client.setCallback(new PushCallback());
MqttTopic topic = client.getTopic(TOPIC1);
//遗嘱
options.setWill(topic, "close".getBytes(), 1, true);
client.connect(options);
//订阅消息
int[] Qos = {1};//0：最多一次 、1：最少一次 、2：只有一次
String[] topic1 = {TOPIC1};
client.subscribe(topic1, Qos);
//向client2发送消息
pubMessage("哈喽!","ASDFG");
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* 消息发送
* @param message
* @param topic
*/
public void pubMessage(String message,String topic){
MqttMessage mess = new MqttMessage();
mess.setQos(1);
mess.setRetained(true);
mess.setPayload(message.getBytes());
try {
client.publish(topic, mess);
} catch (Exception e) {
//LOGGER.error(e.getLocalizedMessage());
}
}
public static void main(String[] args) {
ClientMQTT client1 = new ClientMQTT();
client1.start();
}
}
mqttclient2代码如下
#pom依赖
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
<version>1.2.2</version>
</dependency>
public class ClientMQTT {
public static final String HOST = "tcp://192.168.31.74:1883";
public static final String TOPIC1 = "ASDFG";//请阅地址
private static final String clientid = "client2";
private MqttClient client;
private MqttConnectOptions options;
private String userName = "mqtt2"; //非必须
private String passWord = "mqtt2"; //非必须
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);
//设置断开后重新连接
options.setAutomaticReconnect(true);
// 设置回调
client.setCallback(new PushCallback());
MqttTopic topic = client.getTopic(TOPIC1);
//setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息
//遗嘱
options.setWill(topic, "close".getBytes(), 1, true);
client.connect(options);
//订阅消息
int[] Qos = {1};//0：最多一次 、1：最少一次 、2：只有一次
String[] topic1 = {TOPIC1};
client.subscribe(topic1, Qos);
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* 消息发送
* @param message
* @param topic
*/
public void pubMessage(String message,String topic){
MqttMessage mess = new MqttMessage();
mess.setQos(1);
mess.setRetained(true);
mess.setPayload(message.getBytes());
try {
client.publish(topic, mess);
} catch (Exception e) {
//LOGGER.error(e.getLocalizedMessage());
}
}
public static void main(String[] args) {
ClientMQTT client2 = new ClientMQTT();
client2.start();
}
}
//接收消息回调
class PushCallback implements MqttCallback {
public void connectionLost(Throwable cause) {
// 连接丢失后，一般在这里面进行重连
System.out.println("连接断开，可以做重连");
}
public void deliveryComplete(IMqttDeliveryToken token) {
System.out.println("deliveryComplete---------" + token.isComplete());
}
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()));
}
}
2.发送消息
首先启动mqttClient2服务准备接收，再启动mqttClient1服务直接发送，下图表示mqttclient2接收到mqttclient1发来的消息
华为开发者空间，是为全球开发者打造的专属开发空间，汇聚了华为优质开发资源及工具，致力于让每一位开发者拥有一台云主机，基于华为根生态开发、创新。
更多推荐
1
0- 0
已为社区贡献2条内容
所有评论(0)