收发消息编写代码非常简单，不需要提前创建主题，直接编码向主题发送消息或订阅接收消息即可。

在本篇博客最后会提供源代码的下载。

客户端

官网地址：https://docs.emqx.com/zh/emqx/v5.8/connect-emqx/introduction.html

Github

源码地址：https://github.com/eclipse-paho/paho.mqtt.java

客户端使用

编码示例官网地址：https://www.emqx.com/zh/blog/how-to-use-mqtt-in-java

Github

示例地址：https://github.com/emqx/MQTT-Client-Examples/tree/master/mqtt-client-Java

id="一springboot-示例程序">一、SpringBoot

示例程序

data-src="https://img2024.cnblogs.com/blog/2502715/202602/2502715-20260216195642516-1296579410.png"

首先我们需要在

xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0

http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.jobs</groupId><artifactId>springboot_emqx</artifactId><version>1.0-SNAPSHOT</version><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.4.5</version></parent><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></dependency><dependency><groupId>org.apache.commons</groupId><artifactId>commons-lang3</artifactId></dependency><!--引入Paho

MQTT

--><dependency><groupId>org.eclipse.paho</groupId><artifactId>org.eclipse.paho.client.mqttv3</artifactId><version>1.2.5</version></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId><version>2.4.5</version></plugin></plugins></build>

</project>

class="language-yaml">server:port:

8080

如果运行多个实例，可以考虑在程序代码中增加随机字符串作为后缀clientId:

emqx-client#

订阅主题表达式（如果为空，表示不订阅；如果订阅多个主题表达式，需要以英文逗号分隔）#

主题表达式可以使用通配符，+号是单层通配符，#号是多层通配符，使用方法参看以下官网链接：#

https://docs.emqx.com/zh/emqx/v5.8/messaging/mqtt-wildcard-subscription.html#

订阅主题默认消息质量等级为

容器。

客户端发送消息使用该类实例对象中的方法即可，具体内容如下：

org.apache.commons.lang3.StringUtils;
import
org.eclipse.paho.client.mqttv3.MqttCallback;
import
org.eclipse.paho.client.mqttv3.MqttClient;
import
org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import
org.eclipse.paho.client.mqttv3.MqttMessage;
import
org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import
org.springframework.beans.factory.annotation.Autowired;
import
org.springframework.beans.factory.annotation.Value;
import
org.springframework.stereotype.Component;import
javax.annotation.PostConstruct;
import
host;@Value("${emqx.clientId}")private
String
clientId;@Value("${emqx.username}")private
String
username;@Value("${emqx.password}")private
String
password;@Value("${emqx.topics}")private
String
topics;//客户端连接收发消息的回调对象，本Demo中由Spring容器注入的是EmqxCallback对象实例@Autowiredprivate
MqttCallback
mqttClient;//当对象实例化后，所需要的其它依赖注入完成，并且即将初始化完成时，执行该方法@PostConstructprivate
void
MqttConnectOptions();options.setAutomaticReconnect(true);
//异常断开后自动重连options.setUserName(username);
//用户账号options.setPassword(password.toCharArray());
//用户密码options.setCleanSession(true);
//当前连接是临时会话，每次连接都创建新的会话//实例化连接对象mqttClient
new
MemoryPersistence());mqttClient.setCallback(mqttCallback);mqttClient.connect(options);
//连接Emqx服务器//如果配置的主题表达式不为空，则进行订阅if
(StringUtils.isNotEmpty(topics))
{//如果配置了多个主题表达式，则以英文逗号分隔String[]
topicsArr
topics.split(",");mqttClient.subscribe(topicsArr);
catch
{log.error("客户端连接Emqx服务器失败：{}",
new
e.getMessage();log.error(errMsg);return
errMsg;}return
MqttMessage();mqttMessage.setPayload(msg);mqttMessage.setQos(qos.getValue());mqttMessage.setRetained(retain);//发送消息mqttClient.publish(topic,
mqttMessage);return
e.getMessage();log.error(errMsg);return
errMsg;}}/***
{mqttClient.subscribe(topicFilter,
"";}
e.getMessage();log.error(errMsg);return
errMsg;}}/***
订阅多个主题表达式（传入的列表中，Qos的顺序需要与主题表达式的顺序保持一致）**
@param
qoslist.stream().mapToInt(QosEnum::getValue).toArray();mqttClient.subscribe(topicArr,
qosArr);return
e.getMessage();log.error(errMsg);return
errMsg;}}/***
{mqttClient.unsubscribe(topicFilter);return
"";}
e.getMessage();log.error(errMsg);return
errMsg;}}/***
String[0]);mqttClient.unsubscribe(topicArr);return
"";}
e.getMessage();log.error(errMsg);return
errMsg;}}
是客户端作为消费者接收发过来的消息，deliveryComplete
org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import
org.eclipse.paho.client.mqttv3.MqttCallback;
import
org.eclipse.paho.client.mqttv3.MqttMessage;
import
org.springframework.stereotype.Component;@Slf4j
@Component
{//由于我们在初始化客户端连接对象时，已经设置了自动重连，所以这里只记录日志log.error("客户端工具意外断开连接："
需要注意的是：该方法由客户端同步调用，在此方法未正确执行前，不会给Emqx服务器发送Ack确认消息*
当客户端重连时，所有Qos1、Qos2且客户端未进行Ack确认的消息，都将由Emqx服务器再次发送给客户端**
@param
StringBuilder();sb.append("消息主题：").append(topic).append(System.lineSeparator());sb.append("消息Id：").append(mqttMessage.getId()).append(System.lineSeparator());//这里假设消息发送者，发送的消息采用的是系统默认字符集编码将字符串转换成字节数组的sb.append("消息内容：").append(new
String(mqttMessage.getPayload())).append(System.lineSeparator());sb.append("消息质量等级Qos：").append(mqttMessage.getQos()).append(System.lineSeparator());//是否是保留消息（当发送者发送消息时，接收客户端处于离线状态。
然后接收客户端再上线，才会收到保留消息）sb.append("是否是保留消息：").append(mqttMessage.isRetained()).append(System.lineSeparator());sb.append("是否是重发的消息：").append(mqttMessage.isDuplicate());log.info(sb.toString());}
catch
{log.error(e.getMessage());}}/***
@param
deliveryComplete(IMqttDeliveryToken
token)
token.getTopics();log.info("向主题（{}）发送消息（id为{}）成功",
String.join(",",
发送消息和订阅主题时，为了规范消息质量等级值的传入，这里专门编写了一个枚举类：
{Qos0(0),Qos1(1),Qos2(2);private
final
org.apache.commons.lang3.StringUtils;
import
org.springframework.beans.factory.annotation.Autowired;
import
org.springframework.boot.test.context.SpringBootTest;@SpringBootTest
public
发消息测试，可以使用【MQTT客户端工具】验证接收情况*/@Testpublic
void
(StringUtils.isNotEmpty(result))
{//如果发送失败，打印
org.apache.commons.lang3.StringUtils;
import
org.springframework.beans.factory.annotation.Autowired;
import
org.springframework.web.bind.annotation.*;@Slf4j
@RestController
发送消息*/@PostMapping("/publish")public
AjaxResult
emqxService.publish(me.getTopic(),
me.getMsg(),
(StringUtils.isNotEmpty(result))
{return
data;//---------返回成功的相关方法public
static
AjaxResult.success("success");}public
static
data);}//---------返回错误的相关方法public
static
AjaxResult.error("error");}public
static
功能介绍：https://docs.emqx.com/zh/emqx/v5.8/data-integration/webhook.html
Web
data-src="https://img2024.cnblogs.com/blog/2502715/202602/2502715-20260216195717191-89019490.png"
触发器选择【消息发布】，请求方式选择【POST】，填写好你提供的
URL
org.springframework.web.bind.annotation.*;import
java.util.Map;@Slf4j
订阅主题中接收的消息数据*/@PostMapping("/webhook")public
AjaxResult
调用发来的消息：{publish_received_at=1771242437998,
qos=1,
metadata={rule_id=webhooktest_WH_D},
id=00064AEF8453F73D599000004CFD0002,
flags={retain=false,
根据接口调用后在控制台上打印出的内容，我们可以通过具体的
key
本篇博客的源代码下载地址为：https://files.cnblogs.com/files/blogs/699532/springboot_emqx.zip