1.emqtt下载地址：http://emqtt.com/downloads，找到自己要下载的版本信息，注意开发版、稳定版；

2.下载后放到硬盘根目录，进入下载路径目录，cmd进入dos窗口，如下图：

3.输入.binemqttd consloe,之后弹出启动状态页面，代表启动成功；

用户名为：admin   密码为：public

登录上去就可以看服务器的运行状态了。

测试

依赖

org.eclipse.paho

org.eclipse.paho.client.mqttv3

1.0.2

Server

package com.mymqtt.myemqtt;

import java.util.Scanner;

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

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

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

public class Server {

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

String host = "tcp://127.0.0.1:1883";

String topic = "hello";

String clientId = "server";// clientId不能重复

MqttConnectOptions options = new MqttConnectOptions();

options.setCleanSession(true);

MqttClient client = new MqttClient(host, clientId);

client.connect(options);

MqttMessage message = new MqttMessage();

@SuppressWarnings("resource")

Scanner scanner = new Scanner(System.in);

System.out.println("请输入要发送的内容：");

while (true) {

String line = scanner.nextLine();

message.setPayload(line.getBytes());

client.publish(topic, message);

}

}

}

Client

package com.mymqtt.myemqtt;

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

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

public class Client {

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

String host = "tcp://127.0.0.1:1883";

String topic = "hello";

String clientId = "12345";// clientId不能重复

// 1.设置mqtt连接属性

MqttConnectOptions options = new MqttConnectOptions();

options.setCleanSession(true);

// 2.实例化mqtt客户端

MqttClient client = new MqttClient(host, clientId);

// 3.连接

client.connect(options);

client.setCallback(new PushCallback());

while (true) {

client.subscribe(topic, 2);

}

// client.disconnect();

}

}

PushCallback

package com.mymqtt.myemqtt;

import java.util.Date;

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

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

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

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

public class PushCallback implements MqttCallback {

private final Logger logger = LoggerFactory.getLogger(this.getClass());

public void connectionLost(Throwable cause) {

// 连接丢失后，一般在这里面进行重连

System.out.println("连接断开，可以做重连");

logger.info("掉线时间:{}", new Date());

}

public void deliveryComplete(IMqttDeliveryToken token) {

System.out.println("deliveryComplete---------" + token.isComplete());

}

public void messageArrived(String topic, MqttMessage message) throws Exception {

// subscribe后得到的消息会执行到这里面

// System.out.println(message);

System.out.println("接收消息主题 : " + topic);

System.out.println("接收消息Qos : " + message.getQos());

System.out.println("接收消息内容 : " + new String(message.getPayload()));

}

}

结果：

控制台显示

两个连接