有时候我们的项目中会用到 即时通讯功能，比如电商系统中的客服聊天、支付成功后的异步回调通知等。最近发现RabbitMQ可以很方便的实现 即时通讯功能，如果你没有特殊的业务需求，甚至可以不写后端代码，今天给大家介绍下如何使用RabbitMQ来实现 即时通讯！

MQTT协议

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的 轻量级通讯协议，该协议构建于 TCP/IP协议上。MQTT最大优点在于，可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。

RabbitMQ启用MQTT功能

RabbitMQ默认不会启用启用MQTT功能，需要手动开启。

我们需要先安装并启动RabbitMQ，这里以Docker环境安装为例，运行命令如下;

docker run -p 5672:5672 -p 15672:15672 -p 1883:1883 -p 15675:15675 --name rabbitmq-mqtt \
-v /mydata/rabbitmq-mqtt/data:/var/lib/rabbitmq \
-d rabbitmq:3.9.11-management

接下来就是启用RabbitMQ的MQTT WEB插件了，进入rabbitmq容器后，使用如下命令开启；


# 先进入rabbitmq容器
docker 
exec -it rabbitmq-mqtt /bin/bash

# 再启用mqtt web插件，会同时启用rabbitmq_mqtt插件
rabbitmq-plugins 
enable rabbitmq_web_mqtt

开启成功后，查看管理控制台，我们可以发现MQTT的web服务运行在 15675端口上了，访问地址：http://192.168.3.101:15672

这或许是一个对你有用的开源项目，mall项目是一套基于 SpringBoot3 + Vue 的电商系统（Github标星60K），后端支持多模块和 最新微服务架构 ，采用Docker和K8S部署。包括前台商城项目和后台管理系统，能支持完整的订单流程！涵盖商品、订单、购物车、权限、优惠券、会员、支付等功能！

Boot项目：github.com/macrozheng/…

Cloud项目：github.com/macrozheng/…

教程网站：

项目演示：

MQTT客户端

我们可以使用MQTT客户端来测试MQTT的即时通讯功能，这里使用的是 MQTTX这个客户端工具。

这里使用Docker环境为例，运行命令如下;

docker run -p 80:80 --name mqttx-web -d emqx/mqttx-web

运行成功后可以访问MQTTX的控制台，访问地址：http://192.168.3.101

点击左侧的 加号按钮来创建一个MQTT连接，配置好连接信息，注意MQTT版本选择 3.1.1；

再添加一个订阅，订阅 testTopicA这个主题，我们会向这个主题发送消息；

发布者向主题中发布消息，订阅者可以实时接收到。

前端直接实现即时通讯

既然 MQTTX客户端可以直接通过RabbitMQ实现即时通讯，那我们直接使用前端技术是否也能实现即时通讯？答案是肯定的！下面我们将通过 html+javascript实现一个简单的聊天功能，真正不写一行后端代码实现即时通讯！

WEB端与MQTT服务进行通讯需要使用一个叫 MQTT.js的库，项目地址：github.com/mqttjs/MQTT…

实现的功能非常简单，一个单聊功能，需要注意的是配置好MQTT服务的访问地址为：ws://192.168.3.101:15675/ws


<!DOCTYPE 
html>

<
html 
lang=
"en">

<
head>
    
<
meta 
charset=
"UTF-8">
    
<
title>Title
</
title>

</
head>

<
body>

<
div>
    
<
label>目标Topic：
<
input 
id=
"targetTopicInput" 
type=
"text">
</
label>
<
br>
    
<
label>发送消息：
<
input 
id=
"messageInput" 
type=
"text">
</
label>
<
br>
    
<
button 
onclick=
"sendMessage()">发送
</
button>
    
<
button 
onclick=
"clearMessage()">清空
</
button>
    
<
div 
id=
"messageDiv">
</
div>

</
div>

</
body>

<
script 
src=
"https://unpkg.com/mqtt/dist/mqtt.min.js">
</
script>

<
script>

    
//RabbitMQ的web-mqtt连接地址
    
const url = 
'ws://192.168.3.101:15675/ws';
    
//获取订阅的topic
    
const topic = 
getQueryString(
"topic");
    
//连接到消息队列
    
let client = mqtt.
connect(url);
    client.
on(
'connect', 
function (
) {
        
//连接成功后订阅topic
        client.
subscribe(topic, 
function (
err) {
            
if (!err) {
                
showMessage(
"订阅topic：" + topic + 
"成功！");
            }
        });
    });
    
//获取订阅topic中的消息
    client.
on(
'message', 
function (
topic, message) {
        
showMessage(
"收到消息：" + message.
toString());
    });

    
//发送消息
    
function 
sendMessage(
) {
        
let targetTopic = 
document.
getElementById(
"targetTopicInput").
value;
        
let message = 
document.
getElementById(
"messageInput").
value;
        
//向目标topic中发送消息
        client.
publish(targetTopic, message);
        
showMessage(
"发送消息给" + targetTopic + 
"的消息：" + message);
    }

    
//从URL中获取参数
    
function 
getQueryString(
name) {
        
let reg = 
new 
RegExp(
"(^|&)" + name + 
"=([^&]*)(&|$)", 
"i");
        
let r = 
window.
location.
search.
substr(
1).
match(reg);
        
if (r != 
null) {
            
return 
decodeURIComponent(r[
2]);
        }
        
return 
null;
    }

    
//在消息列表中展示消息
    
function 
showMessage(
message) {
        
let messageDiv = 
document.
getElementById(
"messageDiv");
        
let messageEle = 
document.
createElement(
"div");
        messageEle.
innerText = message;
        messageDiv.
appendChild(messageEle);
    }

    
//清空消息列表
    
function 
clearMessage(
) {
        
let messageDiv = 
document.
getElementById(
"messageDiv");
        messageDiv.
innerHTML = 
"";
    }

</
script>

</
html>

接下来我们订阅不同的主题开启两个页面测试下功能（页面已经放在SpringBoot项目的resources目录下了，需要先启动应用再访问）：
- 第一个订阅主题 testTopicA，访问地址：http://localhost:8088/page/index?topic=testTopicA
- 第二个订阅主题 testTopicB，访问地址：http://localhost:8088/page/index?topic=testTopicB
之后互相发送消息，让我们来看看效果吧！

在SpringBoot中使用

没有特殊业务需求的时候，前端可以直接和RabbitMQ对接实现即时通讯。但有时候我们需要通过服务端去通知前端，此时就需要在应用中集成MQTT了，接下来我们来讲讲如何在SpringBoot应用中使用MQTT。

首先我们需要在项目的 pom.xml中添加MQTT相关依赖；


<!--Spring集成MQTT-->

<
dependency>
    
<
groupId>org.springframework.integration
</
groupId>
    
<
artifactId>spring-integration-mqtt
</
artifactId>

</
dependency>

在 application.yml中添加MQTT相关配置，主要是访问地址、用户名密码、默认主题信息；


rabbitmq:
  
mqtt:
    
url: 
tcp://192.168.3.101:1883
    
username: 
guest
    
password: 
guest
    
defaultTopic: 
testTopic

编写一个Java配置类从配置文件中读取上面的配置便于使用；


/**
 * 
@auther macrozheng
 * 
@description MQTT相关配置
 * 
@date 2025/8/1
 * 
@github https://github.com/macrozheng
 */

@Data

@EqualsAndHashCode(callSuper = false)

@Component

@ConfigurationProperties(prefix = "rabbitmq.mqtt")

public 
class 
MqttConfig {
    
/**
     * RabbitMQ连接用户名
     */
    
private String username;
    
/**
     * RabbitMQ连接密码
     */
    
private String password;
    
/**
     * RabbitMQ的MQTT默认topic
     */
    
private String defaultTopic;
    
/**
     * RabbitMQ的MQTT连接地址
     */
    
private String url;
}

添加MQTT消息订阅者相关配置，使用 @ServiceActivator注解声明一个服务激活器，通过 MessageHandler来处理订阅消息；


/**
 * 
@auther macrozheng
 * 
@description MQTT消息订阅者相关配置
 * 
@date 2025/8/1
 * 
@github https://github.com/macrozheng
 */

@Slf4j

@Configuration

public 
class 
MqttInboundConfig {
    
@Autowired
    
private MqttConfig mqttConfig;

    
@Bean
    
public MessageChannel 
mqttInputChannel
() {
        
return 
new 
DirectChannel();
    }

    
@Bean
    
public MessageProducer 
inbound
() {
        
MqttPahoMessageDrivenChannelAdapter 
adapter 
=
                
new 
MqttPahoMessageDrivenChannelAdapter(mqttConfig.getUrl(), 
"subscriberClient",
                        mqttConfig.getDefaultTopic());
        adapter.setCompletionTimeout(
5000);
        adapter.setConverter(
new 
DefaultPahoMessageConverter());
        
//设置消息质量：0->至多一次；1->至少一次；2->只有一次
        adapter.setQos(
1);
        adapter.setOutputChannel(mqttInputChannel());
        
return adapter;
    }

    
@Bean
    
@ServiceActivator(inputChannel = "mqttInputChannel")
    
public MessageHandler 
handler
() {
        
return 
new 
MessageHandler() {

            
@Override
            
public 
void 
handleMessage
(Message<?> message) 
throws MessagingException {
                
//处理订阅消息
                log.info(
"handleMessage : {}",message.getPayload());
            }

        };
    }
}

添加MQTT消息发布者相关配置；


/**
 * 
@auther macrozheng
 * 
@description MQTT消息发布者相关配置
 * 
@date 2025/8/1
 * 
@github https://github.com/macrozheng
 */

@Configuration

public 
class 
MqttOutboundConfig {

    
@Autowired
    
private MqttConfig mqttConfig;

    
@Bean
    
public MqttPahoClientFactory 
mqttClientFactory
() {
        
DefaultMqttPahoClientFactory 
factory 
= 
new 
DefaultMqttPahoClientFactory();
        
MqttConnectOptions 
options 
= 
new 
MqttConnectOptions();
        options.setServerURIs(
new 
String[] { mqttConfig.getUrl()});
        options.setUserName(mqttConfig.getUsername());
        options.setPassword(mqttConfig.getPassword().toCharArray());
        factory.setConnectionOptions(options);
        
return factory;
    }

    
@Bean
    
@ServiceActivator(inputChannel = "mqttOutboundChannel")
    
public MessageHandler 
mqttOutbound
() {
        
MqttPahoMessageHandler 
messageHandler 
=
                
new 
MqttPahoMessageHandler(
"publisherClient", mqttClientFactory());
        messageHandler.setAsync(
true);
        messageHandler.setDefaultTopic(mqttConfig.getDefaultTopic());
        
return messageHandler;
    }

    
@Bean
    
public MessageChannel 
mqttOutboundChannel
() {
        
return 
new 
DirectChannel();
    }
}

添加MQTT网关，用于向主题中发送消息；


/**
 * 
@auther macrozheng
 * 
@description MQTT网关，通过接口将数据传递到集成流
 * 
@date 2025/8/1
 * 
@github https://github.com/macrozheng
 */

@Component

@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")

public 
interface 
MqttGateway {
    
/**
     * 发送消息到默认topic
     */
    
void 
sendToMqtt
(String payload);

    
/**
     * 发送消息到指定topic
     */
    
void 
sendToMqtt
(String payload, 
@Header(MqttHeaders.TOPIC) String topic);

    
/**
     * 发送消息到指定topic并设置QOS
     */
    
void 
sendToMqtt
(
@Header(MqttHeaders.TOPIC) String topic, 
@Header(MqttHeaders.QOS) 
int qos, String payload);
}

添加MQTT测试接口，使用MQTT网关向特定主题中发送消息；


/**
 * 
@auther macrozheng
 * 
@description MQTT测试接口
 * 
@date 2025/8/1
 * 
@github https://github.com/macrozheng
 */

@Slf4j

@RestController

@Tag(name = "MqttController", description = "MQTT测试接口")

@RequestMapping("/mqtt")

public 
class 
MqttController {

    
@Autowired
    
private MqttGateway mqttGateway;

    
@PostMapping("/sendToDefaultTopic")
    
@Operation(summary = "向默认主题发送消息")
    
public CommonResult 
sendToDefaultTopic
(String payload) {
        mqttGateway.sendToMqtt(payload);
        
return CommonResult.success(
null);
    }

    
@PostMapping("/sendToTopic")
    
@Operation(summary = "向指定主题发送消息")
    
public CommonResult 
sendToTopic
(String payload, String topic) {
        mqttGateway.sendToMqtt(payload, topic);
        
return CommonResult.success(
null);
    }
}

调用接口向主题中发送消息进行测试，访问地址：http://localhost:8088/swagger-ui.html

后台成功接收到消息并进行打印。

2025-08-01T16:25:48.503+08:00  INFO 35516 --- [spring-mqtt] [ubscriberClient] c.m.blog.mqtt.config.MqttInboundConfig   : handleMessage : 来自网页上的消息
2025-08-01T16:25:49.329+08:00  INFO 35516 --- [spring-mqtt] [ubscriberClient] c.m.blog.mqtt.config.MqttInboundConfig   : handleMessage : 来自网页上的消息
2025-08-01T16:25:50.134+08:00  INFO 35516 --- [spring-mqtt] [ubscriberClient] c.m.blog.mqtt.config.MqttInboundConfig   : handleMessage : 来自网页上的消息

总结

消息中间件应用越来越广泛，不仅可以实现可靠的异步通信，还可以实现即时通讯，掌握一个消息中间件还是很有必要的。如果没有特殊业务需求，客户端或者前端直接使用MQTT对接消息中间件即可实现即时通讯，有特殊需求的时候也可以使用SpringBoot集成MQTT的方式来实现，总之消息中间件是实现即时通讯的一个好选择！

项目源码地址

github.com/macrozheng/…

ITPub博客

还在用WebSocket实现即时通讯？试试MQTT吧，真香！