消息的发送与接收

news/2024/2/23 14:48:54/文章来源:https://blog.csdn.net/weixin_73195042/article/details/135581622

消息的发送与接收

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

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;}}}//建立连接成功调用@OnOpenpublic 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));}//关闭连接时调用@OnClosepublic 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群发消息@OnMessagepublic 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));}}//错误时调用@OnErrorpublic 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 {@AutowiredLoginService loginservice;@RequestMapping("/onlineusers")@ResponseBodypublic 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")@ResponseBodypublic 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的对比

RabbitMQActiveMQRocketMQKafka
公司/社区RabbitApache阿里Apache
开发语言ErlangJavaJavaScala&Java
协议支持AMQP,XMPP,SMTP,STOMPOpenWire,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使用流程总结

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

  1. 建立connection

  2. 创建channel

  3. 利用channel声明队列

  4. 利用channel向队列发送消息

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

  1. 建立connection

  2. 创建channel

  3. 利用channel声明队列

  4. 定义consumer的消费行为handleDelivery()

  5. 利用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 {@Testpublic 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.创建通道ChannelChannel 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.创建通道ChannelChannel channel = connection.createChannel();// 3.创建队列String queueName = "simple.queue";channel.queueDeclare(queueName, false, false, false, null);// 4.订阅消息channel.basicConsume(queueName, true, new DefaultConsumer(channel){@Overridepublic 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';//获取订阅的topicconst topic = getQueryString("topic");//连接到消息队列let client = mqtt.connect(url);client.on('connect', function () {//连接成功后订阅topicclient.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:1883username: guestpassword: guestdefaultTopic: 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 {@Autowiredprivate MqttConfig mqttConfig;@Beanpublic MessageChannel mqttInputChannel() {return new DirectChannel();}@Beanpublic 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() {@Overridepublic void handleMessage(Message<?> message) throws MessagingException {//处理订阅消息log.info("handleMessage : {}",message.getPayload());}};}
}

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

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

/*** MQTT消息发布者相关配置*/
@Configuration
public class MqttOutboundConfig {@Autowiredprivate MqttConfig mqttConfig;@Beanpublic 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;}@Beanpublic 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 {@Autowiredprivate 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>() {@Overrideprotected 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;}@Overrideprotected 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));}}@Overridepublic void handlerAdded(ChannelHandlerContext ctx) {// 新客户端连接时添加到ChannelGroupchannelGroup.add(ctx.channel());}@Overridepublic void handlerRemoved(ChannelHandlerContext ctx) {// 客户端断开连接时从ChannelGroup中移除channelGroup.remove(ctx.channel());}@Overridepublic 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();}
});

先启动nettyServer 再启动Application,访问:简单聊天室

我们似乎能基本实现聊天。(发送文件需要服务器存储未实现)。但是还是缺了太多东西。这里netty我们还是接触了一点点罢了。

netty更多内容可以参考官方文档Netty: Home或者黑马的教程:https://www.bilibili.com/video/BV1py4y1E7oA

第三方平台实现

环信 - 中国IM即时通讯云服务开创者! (easemob.com)

每个免费用户最多可以注册100个能通讯的用户。

基于项目的实现

简便且功能要求不太多的:websocket

搭建难度高,但是稳定且专业的:netty

哪都沾一点的:rabbitmq

直接上手但是有限制需要花钱的:第三方

参考文档

Spring-websocket: Spring boot整合websocket实现即时通讯 (gitee.com)

RabbitMQ实现即时通讯_rabbitmq 聊天-CSDN博客

https://github.com/macrozheng/mall-learning/tree/master/mall-tiny-mqtt

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.luyixian.cn/news_show_926196.aspx

如若内容造成侵权/违法违规/事实不符,请联系dt猫网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

(C语言)冒泡排序

一、运行结果&#xff1b; 二、源代码&#xff1b; # define _CRT_SECURE_NO_WARNINGS # include <stdio.h>//实现buble_sort函数&#xff1b; void buble_sort(int arr[], int sz) {//初始化变量值&#xff1b;int i 0;//嵌套循环冒泡排序&#xff1b;//外层循环&…

【REST2SQL】10 REST2SQL操作指南

【REST2SQL】01RDB关系型数据库REST初设计 【REST2SQL】02 GO连接Oracle数据库 【REST2SQL】03 GO读取JSON文件 【REST2SQL】04 REST2SQL第一版Oracle版实现 【REST2SQL】05 GO 操作 达梦 数据库 【REST2SQL】06 GO 跨包接口重构代码 【REST2SQL】07 GO 操作 Mysql 数据库 【RE…

微信小程序-----全局配置与页面配置

目录 前言 全局配置文件 一、window 1. 小程序窗口的组成部分 2. window 节点常用的配置项 3. 设置导航栏的标题 4. 设置导航栏的背景色 5. 设置导航栏的标题颜色 6. 全局开启下拉刷新功能 7. 设置下拉刷新时窗口的背景色 8. 设置下拉刷新时 loading 的样式 9. 设置…

两步解决宝塔面板无法访问(无法访问或拒绝链接)

宝塔面板&#xff0c;突然无法进入&#xff0c;显示“IP拒绝链接”。 使用SSH工具登录服务器 /etc/init.d/bt defaultbt default 命令 宝塔获取登录的默认地址、用户名和登录密码&#xff1b; 重启面板服务 sudo /etc/init.d/bt初始化宝塔选项 漏刻有时

装饰者模式:打破继承限制,实现灵活的功能扩展

欢迎来到我的博客&#xff0c;代码的世界里&#xff0c;每一行都是一个故事 装饰者模式&#xff1a;打破继承限制&#xff0c;实现灵活的功能扩展 前言装饰者模式简介装饰者模式的工作原理实际应用java代码实现结语 前言 在软件开发中&#xff0c;我们经常面临着需求的变化和新…

rsync全面讲解

rsync 是一个常用的 Linux 应用程序&#xff0c;用于文件同步。 它可以在本地计算机与远程计算机之间&#xff0c;或者两个本地目录之间同步文件&#xff08;但不支持两台远程计算机之间的同步&#xff09;。它也可以当作文件复制工具&#xff0c;替代cp和mv命令。 它名称里面…

基础面试题整理4

1.mybatis的#{}和${}区别 #{}是预编译处理&#xff0c;${}是字符串替换#{}可以防止SQL注入&#xff0c;提高安全性 2.mybatis隔离级别 读未提交 READ UNCOMMITED&#xff1a;读到了其他事务中未提交的数据&#xff0c;造成"脏读","不可重复读","幻读&…

Python进程池multiprocessing.Pool

环境&#xff1a; 鲲鹏920:192核心 内存&#xff1a;756G python&#xff1a;3.9 python单进程的耗时 在做单纯的cpu计算的场景&#xff0c;使用单进程核多进程的耗时做如下测试&#xff1a; 单进程情况下cpu的占用了如下&#xff0c;占用一半的核心数&#xff1a; 每一步…

git 提炼笔记

1、设置用户名和邮箱&#xff08;邮箱可以不是真的&#xff09; git config --global user.name test101 // 设置用户名为 test101git config --global user.email test101test101.cn // 设置邮箱为test101test101.cn2、查看用户名和邮箱 git config --global user.name git…

【SpringBoot框架篇】35.kafka环境搭建和收发消息

kafka环境搭建 kafka依赖java环境,如果没有则需要安装jdk yum install java-1.8.0-openjdk* -y1.下载安装kafka kafka3.0版本后默认自带了zookeeper&#xff0c;3.0之前的版本需要单独再安装zookeeper,我使用的最新的3.6.1版本。 cd /usr/local wget https://dlcdn.apache.…

Redis主从架构、哨兵集群原理实战

1.主从架构简介 背景 单机部署简单&#xff0c;但是可靠性低&#xff0c;且不能很好利用CPU多核处理能力生产环境必须要保证高可用&#xff0c;一般不可能单机部署读写分离是可用性要求不高、性能要求较高、数据规模小的情况 目标 读写分离&#xff0c;扩展主节点的读能力&…

canvas绘制美队盾牌

查看专栏目录 canvas示例教程100专栏&#xff0c;提供canvas的基础知识&#xff0c;高级动画&#xff0c;相关应用扩展等信息。canvas作为html的一部分&#xff0c;是图像图标地图可视化的一个重要的基础&#xff0c;学好了canvas&#xff0c;在其他的一些应用上将会起到非常重…

STC8H8K蓝牙智能巡线小车——1. 环境搭建(基于RTX51操作系统)

1. 基本介绍 开发环境准备&#xff1a;Keil uVision5 烧录软件&#xff1a;STC-ISP&#xff08;V6.92A&#xff09; 芯片&#xff1a; STC8H8K64U-45I-LQFP64 芯片引脚&#xff1a; 2.创建项目 打开Keil&#xff0c;点击【Project】&#xff0c;选择【new uVersion proje…

快乐学Python,如何使用爬虫从网页中提取感兴趣的内容?

前面的内容&#xff0c;我们了解了使用urllib3和selenium来下载网页&#xff0c;但下载下来的是整个网页的内容&#xff0c;那我们又怎么从下载下来的网页中提取我们自己感兴趣的内容呢&#xff1f;这里就需要Python的另一个库来实现-BeautifulSoup。 BeautifulSoup 是一个 Py…

数据仓库(2)-认识数仓

1、数据仓库是什么 数据仓库 &#xff0c;由数据仓库之父比尔恩门&#xff08;Bill Inmon&#xff09;于1990年提出&#xff0c;主要功能仍是将组织透过资讯系统之联机事务处理(OLTP)经年累月所累积的大量资料&#xff0c;透过数据仓库理论所特有的资料储存架构&#xff0c;做…

可以在微信群里使用midjourney,gpt4,gemini,文心一言4.0,且免费

免费使用gpt4和midjourney 免费使用 参考链接&#xff1a; https://chat.xutongbao.top/

【银行测试】银行项目,信用卡业务测试+常问面试(三)

目录&#xff1a;导读 前言一、Python编程入门到精通二、接口自动化项目实战三、Web自动化项目实战四、App自动化项目实战五、一线大厂简历六、测试开发DevOps体系七、常用自动化测试工具八、JMeter性能测试九、总结&#xff08;尾部小惊喜&#xff09; 前言 银行测试-信用卡业…

轻松识别Midjourney等AI生成图片,开源GenImage

AIGC时代&#xff0c;人人都可以使用Midjourney、Stable Diffusion等AI产品生成高质量图片&#xff0c;其逼真程度肉眼难以区分真假。这种虚假照片有时会对社会产生不良影响&#xff0c;例如&#xff0c;生成公众人物不雅图片用于散播谣言&#xff1b;合成虚假图片用于金融欺诈…

Angular系列教程之DOM操作

文章目录 引言1. ElementRef2. Renderer23. ViewChild结论 引言 在Angular中&#xff0c;DOM操作是开发Web应用程序的一个重要方面。通过对DOM进行操作&#xff0c;我们可以动态地修改页面内容、样式和元素行为。本文将详细介绍如何在Angular中进行DOM操作&#xff0c;并提供相…

从数据可视化到场景渲染:山海鲸的创新与实践

作为山海鲸的开发者&#xff0c;我们深知可视化模型场景渲染在数据分析和决策支持中的重要作用。因此在保证山海鲸可视化软件免费编辑、分享、部署的同时也在场景渲染方面不断优化&#xff0c;本文将介绍山海鲸在可视化模型场景渲染方面的技术革新与实践探索。 首先&#xff0…