消息的发送与接收

消息的发送与接收不仅仅是在于聊天功能的实现。其实还有很多种情况也算"消息的发送与接收"。而且我们还可以通过多种方法去实现。我们可以基于实际情况来选择。

WebSocket实现

node做后端。找了好多，前端页面总是用到了jQuery，包括底下的java做后端的前端代码等。我们先用最简单的代码来帮助我们吧！

首先，配好package.json。注意依赖要引入ws。运行项目前先要运行一下：npm i

{
  "name": "chat-server",
  "version": "1.0.0",
  "description": "",
  "main": "client.js",
  "scripts": {
    "test": "echo \"Error: no test specified\" && exit 1",
    "start": "node server.js"
  },
  "dependencies": {
    "debug": "~2.6.9",
    "ejs": "~2.6.1",
    "express": "~4.16.1",
    "express-session": "^1.17.2",
    "http-errors": "~1.6.3",
    "jsonwebtoken": "^8.5.1",
    "ws": "^8.5.0"
  },
  "keywords": [],
  "author": "",
  "license": "ISC"
}

创建server.js，这是服务端

const  WebSocket = require("ws")
WebSocketServer = WebSocket.WebSocketServer
const wss = new WebSocketServer({ port: 8080 });
wss.on('connection', function connection(ws) {
    ws.on('message', function message(data, isBinary) {
        wss.clients.forEach(function each(client) {
            if (client !== ws && client.readyState === WebSocket.OPEN) {
                client.send(data, { binary: isBinary });
            }
        });

    });

    ws.send('欢迎加入聊天室');
});

创建client.js，这是客户端

const  WebSocket = require("ws")
var ws = new WebSocket("ws://localhost:8080")
ws.onopen = ()=>{
    console.log("open")
}
ws.onmessage = (evt)=>{
    console.log(evt.data)
}

创建test.html，此处用vscode插件live-server打开。

<!DOCTYPE html>
<html lang="en">

<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <meta http-equiv="X-UA-Compatible" content="ie=edge">
    <title>Document</title>
</head>

<body>
    <h1>websockets简单示例</h1><br>
    <div id="message"></div>
    <div>
        <input type="text" id="sendText">
        <button id="connect" onclick="connect()">建立连接</button>
        <button id="sendData" onclick="sendData()">发送数据</button>
        <button id="closeConnect" onclick="closeConnect()">关闭连接</button>
    </div>
</body>
<script type="text/javascript">
    let websockets;
    //创建一个数组对象用于存放当前的连接的状态，以便在页面上实时展示出来当前的状态
    let statusArr = [
        { state: 0, value: '正在连接' },
        { state: 1, value: '已建立连接' },
        { state: 2, value: '正在关闭连接' },
        { state: 3, value: '已关闭连接' },
    ]
    /**
    *   建立连接
    *
    */
    function connect() {
        // 1. 创建websockets对象，参数为服务器websockets地址
        websockets = new WebSocket("ws:127.0.0.1:8080");

        // 2.监听websocket的状态变化，接收的信息，关闭时的状态

        //监听连接状态的变化
        websockets.onopen = (event) => socketChange();

        //监听接收消息的情况
        websockets.onmessage = (res) => {
            document.querySelector("#message").innerHTML += `<p>接收数据: ${res.data}</p>`
        }

        //监听关闭时的状态变化
        websockets.onclose = (event) => socketChange();
    }
    /**
    *   socket状态变化
    *
    */
    function socketChange() {
        let state = websockets.readyState;
        let val = statusArr.map((item) => {
            if (item.state == state) {
                return item.value
            }
        });

        //实时显示状态的变化
        document.querySelector("#message").innerHTML += `<p>当前的socket连接状态是: ${val}</p>`
    }
    /**
    *   发送数据
    *
    */
    function sendData() {
        //1. 首先获取输入的信息，判断信息是否可以发送
        let val = document.querySelector("#sendText").value;

        if (val == "" || val == undefined) {
            document.querySelector("#message").innerHTML += "<p>发送数据为空，请填写完成后再发送！</p>";
            return;
        }

        websockets.send(val);
        document.querySelector("#message").innerHTML += `<p>发送数据:${val}</p>`;
    }
    /**
    *   关闭连接
    *
    */
    function closeConnect() {
        websockets.close();
    }
</script>
</html>

在本文件夹内的控制台输入 node .\server.js node .\client.js 启动服务器端和客户端。

我们打开Document两个窗口，尝试进行消息交流。

在这里插入图片描述

我们发现消息是能正常发送出去并能被正常接收到。

Java使用Socket实现

java做后端。此处使用了Spring-websocket: Spring boot整合websocket实现即时通讯 (gitee.com)的代码。

引入依赖：

 <dependencies>
		<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-websocket</artifactId>
            <version>2.1.6.RELEASE</version>
        </dependency>
        
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-thymeleaf</artifactId>
			<version>2.1.6.RELEASE</version>
		</dependency>

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter</artifactId>
			<version>2.1.6.RELEASE</version>
			<exclusions><!-- 去掉springboot默认配置 -->  
		        <exclusion>  
		            <groupId>org.springframework.boot</groupId>  
		            <artifactId>spring-boot-starter-logging</artifactId>  
		        </exclusion>  
		    </exclusions>  
		</dependency>

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<version>2.1.6.RELEASE</version>
			<scope>test</scope>
		</dependency>

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-web</artifactId>
			<version>2.1.6.RELEASE</version>
			<exclusions><!-- 去掉springboot默认配置 -->  
		        <exclusion>  
		            <groupId>org.springframework.boot</groupId>  
		            <artifactId>spring-boot-starter-logging</artifactId>  
		        </exclusion>  
		    </exclusions>  
		</dependency>
	  <dependency>
		  <groupId>mysql</groupId>
		  <artifactId>mysql-connector-java</artifactId>
		  <version>8.0.25</version>
	  </dependency>
        <dependency>
			<groupId>org.mybatis.spring.boot</groupId>
			<artifactId>mybatis-spring-boot-starter</artifactId>
			<version>2.0.1</version>
		</dependency>
		<dependency>
		    <groupId>com.github.pagehelper</groupId>
		    <artifactId>pagehelper</artifactId>
		    <version>4.1.6</version>
		</dependency>
		<dependency>
		    <groupId>org.apache.poi</groupId>
		    <artifactId>poi</artifactId>
		    <version>3.14</version>
		</dependency>

		<dependency>
		    <groupId>com.alibaba</groupId>
		    <artifactId>druid</artifactId>
		    <version>1.1.20</version>
		</dependency>
		<dependency>
		   <groupId>org.postgresql</groupId>
		   <artifactId>postgresql</artifactId>
		   <version>42.2.5</version>
		</dependency>
		<dependency>
		    <groupId>com.alibaba</groupId>
		    <artifactId>fastjson</artifactId>
		    <version>1.2.68</version>
		</dependency>
		<dependency>
            <groupId>commons-net</groupId>
            <artifactId>commons-net</artifactId>
            <version>3.1</version>
        </dependency>
        <dependency>
		    <groupId>org.springframework.boot</groupId>  
		    <artifactId>spring-boot-starter-log4j2</artifactId>
		    <version>2.1.6.RELEASE</version>
		</dependency>

核心方法：

webSocketServer类

package boot.spring.service;

import java.io.IOException;
import java.util.Date;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;

import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;

import boot.spring.po.Message;

@ServerEndpoint("/webSocket/{username}")
@Component
public class WebSocketServer {
	 //静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
    private static AtomicInteger onlineNum = new AtomicInteger();
    //concurrent包的线程安全Set，用来存放每个客户端对应的WebSocketServer对象。
    private static ConcurrentHashMap<String, Session> sessionPools = new ConcurrentHashMap<>();
    //发送消息
    public void sendMessage(Session session, String message) throws IOException {
        if(session != null){
            synchronized (session) {
                System.out.println("发送数据：" + message);
                session.getBasicRemote().sendText(message);
            }
        }
    }
    //给指定用户发送信息
    public void sendInfo(String userName, String message){
        Session session = sessionPools.get(userName);
        try {
            sendMessage(session, message);
        }catch (Exception e){
            e.printStackTrace();
        }
    }
    // 群发消息
    public void broadcast(String message){
    	for (Session session: sessionPools.values()) {
            try {
                sendMessage(session, message);
            } catch(Exception e){
                e.printStackTrace();
                continue;
            }
        }
    }
    //建立连接成功调用
    @OnOpen
    public void onOpen(Session session, @PathParam(value = "username") String userName){
        sessionPools.put(userName, session);
        addOnlineCount();
        System.out.println(userName + "加入webSocket！当前人数为" + onlineNum);
        // 广播上线消息
        Message msg = new Message();
        msg.setDate(new Date());
        msg.setTo("0");
        msg.setText(userName);
        broadcast(JSON.toJSONString(msg,true));
    }
    //关闭连接时调用
    @OnClose
    public void onClose(@PathParam(value = "username") String userName){
        sessionPools.remove(userName);
        subOnlineCount();
        System.out.println(userName + "断开webSocket连接！当前人数为" + onlineNum);
        // 广播下线消息
        Message msg = new Message();
        msg.setDate(new Date());
        msg.setTo("-2");
        msg.setText(userName);
        broadcast(JSON.toJSONString(msg,true));
    }

    //收到客户端信息后，根据接收人的username把消息推下去或者群发
    // to=-1群发消息
    @OnMessage
    public void onMessage(String message) throws IOException{
        System.out.println("server get" + message);
        Message msg=JSON.parseObject(message, Message.class);
		msg.setDate(new Date());
		if (msg.getTo().equals("-1")) {
			broadcast(JSON.toJSONString(msg,true));
		} else {
			sendInfo(msg.getTo(), JSON.toJSONString(msg,true));
		}
    }

    //错误时调用
    @OnError
    public void onError(Session session, Throwable throwable){
        System.out.println("发生错误");
        throwable.printStackTrace();
    }

    public static void addOnlineCount(){
        onlineNum.incrementAndGet();
    }

    public static void subOnlineCount() {
        onlineNum.decrementAndGet();
    }
    
    public static AtomicInteger getOnlineNumber() {
        return onlineNum;
    }
    
    public static ConcurrentHashMap<String, Session> getSessionPools() {
        return sessionPools;
    }
}

controller层

@Controller
public class ChatController {

	@Autowired
	LoginService loginservice;
    
	@RequestMapping("/onlineusers")
	@ResponseBody
	public Set<String> onlineusers(@RequestParam("currentuser") String currentuser) {
		ConcurrentHashMap<String, Session> map = WebSocketServer.getSessionPools();
		Set<String> set = map.keySet();
		Iterator<String> it = set.iterator();
		Set<String> nameset = new HashSet<String>();
		while (it.hasNext()) {
			String entry = it.next();
			if (!entry.equals(currentuser))
				nameset.add(entry);
		}
		return nameset;
	}


	@RequestMapping("getuid")
	@ResponseBody
	public User getuid(@RequestParam("username") String username) {
		Long a = loginservice.getUidbyname(username);
		User u = new User();
		u.setUid(a);
		return u;
	}
}

启动项目，访问：http://localhost:8080/login。此前需要先运行项目里的stuff.sql文件，并配好其数据库密码等，点击登录即可。

在这里插入图片描述

这个项目包含了：即时通信，查看当前在线的其他用户，用户的上线提醒，群发消息等功能。

消息中间件实现

消息中间件市面上常见的四种：ActiveMQ 、RabbitMQ 、 RocketMQ、Kafka等。

几种常见MQ的对比

	RabbitMQ	ActiveMQ	RocketMQ	Kafka
公司/社区	Rabbit	Apache	阿里	Apache
开发语言	Erlang	Java	Java	Scala&Java
协议支持	AMQP，XMPP，SMTP，STOMP	OpenWire,STOMP，REST,XMPP,AMQP	自定义协议	自定义协议
可用性	高	一般	高	高
单机吞吐量	一般	差	高	非常高
消息延迟	微秒级	毫秒级	毫秒级	毫秒以内
消息可靠性	高	一般	高	一般

追求可用性：Kafka、 RocketMQ 、RabbitMQ

追求可靠性：RabbitMQ、RocketMQ

追求吞吐能力：RocketMQ、Kafka

追求消息低延迟：RabbitMQ、Kafka

综合看来，我们选择比较流行的rabbitmq来帮助我们。

部署

在安装rabbitmq之前需要先安装erlang，并配置好其环境变量。（跟java环境变量配置一样）

安装rabbitmq部署并启动服务：安装rabbitmq(解压版教程)_rabbitmq-server 压缩包版本如何使用-CSDN博客

启动有可能会报毒，我们选择允许。

访问：RabbitMQ Management，显示出rabbitmq的管理界面就算部署成功！

默认账号密码都是guest。

RabbitMQ的简要概述

RabbitMQ中的一些角色：

publisher：生产者
consumer：消费者
exchange个：交换机，负责消息路由
queue：队列，存储消息
virtualHost：虚拟主机，隔离不同租户的exchange、queue、消息的隔离

RabbitMQ使用流程总结

基本消息队列的消息发送流程：

建立connection
创建channel
利用channel声明队列
利用channel向队列发送消息

基本消息队列的消息接收流程：

建立connection
创建channel
利用channel声明队列
定义consumer的消费行为handleDelivery()
利用channel将消费者与队列绑定

SpringAMQP

SpringAMQP是基于RabbitMQ封装的一套模板，并且还利用SpringBoot对其实现了自动装配，使用起来非常方便。

SpringAmqp的官方地址：https://spring.io/projects/spring-amqp

SpringAMQP提供了三个功能：

自动声明队列、交换机及其绑定关系
基于注解的监听器模式，异步接收消息
封装了RabbitTemplate工具，用于发送消息

  <dependencies>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <!--AMQP依赖，包含RabbitMQ-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <!--单元测试-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
        </dependency>
    </dependencies>

创建2个模块，publisher和consumer，完善一下项目的启动类，yaml配置文件等。每个模块写一个测试类：

PublisherTest:

public class PublisherTest {
    @Test
    public void testSendMessage() throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        // 1.2.建立连接
        Connection connection = factory.newConnection();

        // 2.创建通道Channel
        Channel channel = connection.createChannel();

        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

        // 4.发送消息
        String message = "hello, rabbitmq!";
        channel.basicPublish("", queueName, null, message.getBytes());
        System.out.println("发送消息成功：【" + message + "】");

        // 5.关闭通道和连接
        channel.close();
        connection.close();

    }
}

ConsumerTest:

public class ConsumerTest {

    public static void main(String[] args) throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        // 1.2.建立连接
        Connection connection = factory.newConnection();

        // 2.创建通道Channel
        Channel channel = connection.createChannel();

        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

        // 4.订阅消息
        channel.basicConsume(queueName, true, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                // 5.处理消息
                String message = new String(body);
                System.out.println("接收到消息：【" + message + "】");
            }
        });
        System.out.println("等待接收消息。。。。");
    }
}

先启动publisher：

在这里插入图片描述

再看看rabbitmq的控制台：
在这里插入图片描述

再启动consumer
在这里插入图片描述

此时再看看rabbitmq控制台：

在这里插入图片描述

消息已经被成功消费了！

rabbitmq实现即时通信

具体可以参考：RabbitMQ实现即时通讯_rabbitmq 聊天-CSDN博客

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

RabbitMQ启用MQTT功能，需要先安装然RabbitMQ然后再启用MQTT插件。

接下来就是启用RabbitMQ的MQTT插件了，默认是不启用的，使用命令开启即可；

需要进到目录\rabbitmq_server-3.9.13\sbin里，执行如下命令：

.\rabbitmq-plugins.bat enable rabbitmq_mqtt

在这里插入图片描述

MQTTX

我们可以使用MQTT客户端来测试MQTT的即时通讯功能，这里使用的是MQTTX这个客户端工具。下载地址：MQTTX: Your All-in-one MQTT Client Toolbox

点击新建连接按钮或者左边的加号来创建一个MQTT客户端；

接下来对MQTT客户端进行配置，主要是配置好协议端口、连接用户名密码和QoS即可。

再配置一个订阅者，订阅者订阅testTopicA这个主题，我们会向这个主题发送消息；

在这里插入图片描述

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

在这里插入图片描述

前端直接实现即时通讯

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

由于RabbitMQ与Web端交互底层使用的是WebSocket，所以我们需要开启RabbitMQ的MQTT WEB支持，使用如下命令开启即可；

rabbitmq-plugins enable rabbitmq_web_mqtt

开启成功后，查看管理控制台，我们可以发现MQTT的WEB服务运行在15675端口上了。

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

<!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://localhost: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>

之后在界面输入对应的topic地址，发送消息，发现是可以实现的！

在SpringBoot中使用

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

此处项目源码地址：https://github.com/macrozheng/mall-learning/tree/master/mall-tiny-mqtt

首先，项目需要引入mqtt依赖

<!--Spring集成MQTT-->
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-mqtt</artifactId>
</dependency>

在application.yml添加配置：

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

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

/**
 * MQTT相关配置
 */
@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来处理订阅消息；

/**
 * MQTT消息订阅者相关配置
 */
@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());
            }
 
        };
    }
}

注意：messageHandler导包路径：import org.springframework.messaging.MessageHandler;

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

/**
 * MQTT消息发布者相关配置
 */
@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网关，用于向主题中发送消息；

/**
 * MQTT网关，通过接口将数据传递到集成流
 */
@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网关向特定主题中发送消息；

/**
 * MQTT测试接口
 */
@RestController
@RequestMapping("/mqtt")
public class MqttController {
    @Autowired
    private MqttGateway mqttGateway;
    @PostMapping("/sendToDefaultTopic")
    public CommonResult sendToDefaultTopic(String payload) {
        mqttGateway.sendToMqtt(payload);
        return CommonResult.success(null);
    }
    @PostMapping("/sendToTopic")
    public CommonResult sendToTopic(String payload, String topic) {
        mqttGateway.sendToMqtt(payload, topic);
        return CommonResult.success(null);
    }
}

别忘声明一个返回类

package com.itcast.mq.model;

import lombok.Data;

import java.io.Serializable;

@Data
public class CommonResult<T> {
    /*返回体*/
    private  Integer code;
    private String msg;
    private T data;

    /*成功，且返回体有数据*/
    public static CommonResult success(Object object) {
        CommonResult r = new CommonResult();
        r.setCode(200);
        r.setMsg("成功");
        r.setData(object);
        return r;
    }
    //成功，但返回体没数据
    public static CommonResult success(){
        return success(null);
    }
    //失败返回信息
    public static CommonResult Err(Integer code,String msg){
        CommonResult r = new CommonResult();
        r.setCode(code);
        r.setMsg(msg);
        return r;
    }

}

配置一下启动类

@SpringBootApplication
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

启动该项目，打开postman输入链接：http://localhost:8080/mqtt//sendToTopic?payload=&topic=。填好输入内容和目标topic的id即可！
在这里插入图片描述

回到mqtt客户端看一下

消息发送成功！

netty实现

Netty是一个异步事件驱动的网络应用程序框架，用于快速开发可维护的高性能协议服务器和客户端。

超详细Netty入门，看这篇就够了！ - 知乎 (zhihu.com)

具体实现可以参考：https://blog.csdn.net/weixin_44814270/article/details/132947704

依赖：

 <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
            <version>2.1.6.RELEASE</version>
            <exclusions><!-- 去掉springboot默认配置 -->
                <exclusion>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-starter-logging</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
            <version>2.1.6.RELEASE</version>
            <exclusions><!-- 去掉springboot默认配置 -->
                <exclusion>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-starter-logging</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <!--我这里使用的是jfinal-enjoy模板引擎-->
        <dependency>
            <groupId>com.jfinal</groupId>
            <artifactId>enjoy</artifactId>
            <version>5.1.2</version>
        </dependency>
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.65.Final</version> <!-- 使用最新版本 -->
        </dependency>
    </dependencies>

配置NettyChatServer：

public class NettyChatServer {

    private final int port;
    private final EventExecutorGroup eventExecutorGroup;
    private final ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    public NettyChatServer(int port) {
        this.port = port;
        this.eventExecutorGroup = new DefaultEventExecutorGroup(4); // 用于在handler中处理耗时任务
    }

    public void start() throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            // 字符串编解码器，用于将消息编码成字符串和解码成字符串
                            pipeline.addLast(new HttpServerCodec());
                            pipeline.addLast(new HttpObjectAggregator(65536));
                            pipeline.addLast(new WebSocketServerProtocolHandler("/websocket"));
                            pipeline.addLast(eventExecutorGroup, new ChatServerHandler(channelGroup));
                            // 添加自定义的聊天处理器
                            //  pipeline.addLast(eventExecutorGroup, new ChatServerHandler(channelGroup));
                        }
                    });

            ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
            System.out.println("Chat Server started on port " + port);

            channelFuture.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public void stop() {
        // 停止服务器
    }

    public static void main(String[] args) {
        int port = 8888;
        NettyChatServer chatServer = new NettyChatServer(port);
        try {
            chatServer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

配置ChatServerHandler

public class ChatServerHandler extends SimpleChannelInboundHandler<WebSocketFrame> {

    private final ChannelGroup channelGroup;

    public ChatServerHandler(ChannelGroup channelGroup) {
        this.channelGroup = channelGroup;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) throws Exception {
        // 处理WebSocket消息
        if (frame instanceof TextWebSocketFrame) {
            TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
            String message = textFrame.text();

            // 在服务器控制台上输出消息
            System.out.println("Received message: " + message);

            // 将消息广播给所有连接的客户端
            channelGroup.writeAndFlush(new TextWebSocketFrame(message));
        }
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        // 新客户端连接时添加到ChannelGroup
        channelGroup.add(ctx.channel());
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        // 客户端断开连接时从ChannelGroup中移除
        channelGroup.remove(ctx.channel());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // 异常处理
        cause.printStackTrace();
        ctx.close();
    }

因为项目用到了，所以我们需要jfinal-enjoy模板引擎文件：

package com.zd.config;

import com.jfinal.template.Engine;
import com.jfinal.template.ext.spring.JFinalViewResolver;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class EnjoyConfig {


    @Bean(name = "jfinalViewResolver")
    public JFinalViewResolver getJFinalViewResolver() {

        // 创建用于整合 spring boot 的 ViewResolver 扩展对象
        JFinalViewResolver jfr = new JFinalViewResolver();

        // 对 spring boot 进行配置
        jfr.setSuffix(".html");
        jfr.setContentType("text/html;charset=UTF-8");
        jfr.setOrder(0);

        // 设置在模板中可通过 #(session.value) 访问 session 中的数据
        jfr.setSessionInView(true);

        // 获取 engine 对象，对 enjoy 模板引擎进行配置，配置方式与前面章节完全一样
        Engine engine = JFinalViewResolver.engine;

        // 热加载配置能对后续配置产生影响，需要放在最前面
        engine.setDevMode(true);

        // 使用 ClassPathSourceFactory 从 class path 与 jar 包中加载模板文件
        engine.setToClassPathSourceFactory();

        // 在使用 ClassPathSourceFactory 时要使用 setBaseTemplatePath
        // 设置静态资源路径在 /static 下
        engine.setBaseTemplatePath("/static/");

        return jfr;
    }
}

配置一下controller层：

@Controller
public class ChatController {

    @RequestMapping("/chat")
    public String main() {
        return "main";
    }
}

配置html、css、js

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>简单聊天室</title>
    <!-- 引入Bootstrap CSS文件 -->
    <link href="./bootstarp/css/bootstrap.min.css" rel="stylesheet">
    <link href="main.css" rel="stylesheet">
</head>
<body>
<div class="container">
    <div class="row">
        <div class="col-md-8 offset-md-2">
            <div id="chat-container">
                <div id="chat-header">
                    <h2>简单聊天室</h2>
                </div>
                <div id="chat-box">
                    <!-- 示例聊天消息 -->
                    <!--  <div class="message">
                          <div class="avatar">A</div>
                          <div class="message-content">
                              <div class="sender-name">User 1</div>
                              <div class="message-text">Hello, how are you?</div>
                          </div>
                      </div>-->
                    <!-- 示例聊天消息结束 -->
                </div>
                <div id="message-buttons">
                    <input type="text" id="message-input" placeholder="输入消息...">
                    <input type="file" id="file-input">
                    <button class="btn btn-primary" onclick="sendMessage()">发送文本</button>
                    <button class="btn btn-primary" onclick="sendFile()">发送文件</button>
                    <button class="btn btn-danger" id="clear-button" onclick="clearChat()">清空聊天</button>
                </div>
            </div>
        </div>
    </div>
</div>

<!-- 引入jQuery -->
<script src="https://code.jquery.com/jquery-3.6.0.min.js"></script>
<!-- 引入Bootstrap JS文件 -->
<script src="./bootstarp/js/bootstrap.min.js"></script>
<script src="main.js"></script>
</body>
</html>

body {
    background-color: #f2f2f2;
    font-family: Arial, Helvetica, sans-serif;
    margin: 0;
    padding: 0;
}

#chat-container {
    max-width: 600px;
    margin: 20px auto;
    background-color: #fff;
    border-radius: 5px;
    box-shadow: 0px 0px 10px rgba(0, 0, 0, 0.1);
    overflow: hidden;
}

#chat-header {
    background-color: #007BFF;
    color: #fff;
    padding: 10px;
    text-align: center;
    border-top-left-radius: 5px;
    border-top-right-radius: 5px;
}

#chat-box {
    max-height: 100vh;
    width: 100%;
    overflow-y: scroll;
    height: 600px;
}

.message {
    display: flex;
    margin: 10px;
    padding: 10px;
    border-bottom: 1px solid #ccc;
}

.avatar {
    width: 50px;
    height: 50px;
    background-color: #007BFF;
    color: #fff;
    border-radius: 50%;
    text-align: center;
    line-height: 50px;
    margin-right: 10px;
}

.message-content {
    flex-grow: 3;
}

.sender-name {
    font-weight: bold;
    margin-bottom: 5px;
}

.message-text {
    word-wrap: break-word;
}

#message-input, #file-input {
    width: 100%;
    padding: 10px;
    border: 1px solid #ccc;
}

#message-buttons {
    padding: 10px;
    text-align: center;
}

button {
    padding: 10px 20px;
    background-color: #007BFF;
    color: #fff;
    border: none;
    cursor: pointer;
    margin-right: 10px;
}

button:hover {
    background-color: #0056b3;
}

#clear-button {
    background-color: #dc3545;
}

@media (max-width: 768px) {
    #chat-container {
        margin-top: 10px;
    }

    #chat-box {
        max-height: 200px;
    }
}

// WebSocket连接
const socket = new WebSocket('ws://localhost:8888/websocket'); // 请将 your_netty_server_address 替换为实际的Netty WebSocket服务器地址

socket.addEventListener('open', (event) => {
    console.log('WebSocket连接已建立');
});

socket.addEventListener('message', (event) => {
    // 解析消息
    console.log(event)
    const data = JSON.parse(event.data);

    const chatBox = document.getElementById('chat-box');

    if (data.type === 'text') {
        // 接收到文本消息
        const messageDiv = document.createElement('div');
        messageDiv.classList.add('message');

        const avatarDiv = document.createElement('div');
        avatarDiv.classList.add('avatar');
        avatarDiv.textContent = data.sender.charAt(0); // 使用发送者的首字母作为头像内容

        const messageContentDiv = document.createElement('div');
        messageContentDiv.classList.add('message-content');

        const senderNameDiv = document.createElement('div');
        senderNameDiv.classList.add('sender-name');
        senderNameDiv.textContent = data.sender;

        const messageTextDiv = document.createElement('div');
        messageTextDiv.classList.add('message-text');
        messageTextDiv.textContent = data.message;

        messageContentDiv.appendChild(senderNameDiv);
        messageContentDiv.appendChild(messageTextDiv);

        messageDiv.appendChild(avatarDiv);
        messageDiv.appendChild(messageContentDiv);

        chatBox.appendChild(messageDiv);
    } else if (data.type === 'file') {
        // 接收到文件消息
        const fileURL = URL.createObjectURL(data.file);
        const messageDiv = document.createElement('div');
        messageDiv.classList.add('message');

        const avatarDiv = document.createElement('div');
        avatarDiv.classList.add('avatar');
        avatarDiv.textContent = data.sender.charAt(0); // 使用发送者的首字母作为头像内容

        const messageContentDiv = document.createElement('div');
        messageContentDiv.classList.add('message-content');

        const senderNameDiv = document.createElement('div');
        senderNameDiv.classList.add('sender-name');
        senderNameDiv.textContent = data.sender;

        const fileLink = document.createElement('a');
        fileLink.href = fileURL;
        fileLink.textContent = '下载文件';
        fileLink.download = data.fileName;

        messageContentDiv.appendChild(senderNameDiv);
        messageContentDiv.appendChild(fileLink);

        messageDiv.appendChild(avatarDiv);
        messageDiv.appendChild(messageContentDiv);

        chatBox.appendChild(messageDiv);
    }

    // 滚动到最新消息
    chatBox.scrollTop = chatBox.scrollHeight;
});

function sendMessage() {
    const messageInput = document.getElementById('message-input');
    const message = messageInput.value.trim();

    if (message !== '') {
        // 发送文本消息到服务器
        const data = {
            sender: 'YSK',
            type: 'text',
            message: message
        };
        socket.send(JSON.stringify(data));
        // 清空输入框
        messageInput.value = '';
    }
}

function sendFile() {
    const fileInput = document.getElementById('file-input');
    const file = fileInput.files[0];

    if (file) {
        // 发送文件到服务器
        const reader = new FileReader();

        reader.onload = function (event) {
            const data = {
                type: 'file',
                fileName: file.name,
                file: event.target.result
            };
            socket.send(JSON.stringify(data));
        };

        reader.readAsArrayBuffer(file);

        // 清空文件选择框
        fileInput.value = '';
    }
}

function clearChat() {
    const chatBox = document.getElementById('chat-box');
    chatBox.innerHTML = '';
}

// 监听Enter键，发送文本消息
const messageInput = document.getElementById('message-input');
messageInput.addEventListener('keyup', function (event) {
    if (event.key === 'Enter') {
        sendMessage();
    }
});