RabbitMQ实践——搭建单人聊天服务

devtools/2024/10/18 5:41:20/

大纲

  • 创建Core交换器
  • 用户登录
  • 发起聊天邀请
  • 接受邀请
  • 聊天
  • 实验过程
  • 总结
  • 代码工程

经过之前的若干节的学习，我们基本掌握了Rabbitmq各个组件和功能。本文我们将使用之前的知识搭建一个简单的单人聊天服务。
基本结构如下。为了避免Server有太多连线导致杂乱，下图将Server画成两个模块，实则是一个服务。
在这里插入图片描述
该服务由两个核心交换器构成。
Core交换器是服务启动时创建的，它主要是为了向不同用户传递“系统通知型”消息。比如Jerry向Tom发起聊天邀请，则是通过上面黑色字体6-10的流程发给了Core交换器。然后Core交换器将该条消息告知Tom。
Fanout交换器是用来消息传递的。Jerry和Tom都向其发送消息，然后路由到两个队列。它们两各自订阅一个队列，就可以看到彼此的聊天内容了。

创建Core交换器

package com.rabbitmq.chat.service;import java.util.Map;
import java.util.concurrent.locks.ReentrantLock;import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;import jakarta.annotation.PostConstruct;
import reactor.core.publisher.Flux;@Service
public class Core {@Autowiredprivate RabbitTemplate rabbitTemplate;private ConnectionFactory connectionFactory;final String exchangeName = "Core";@PostConstructpublic void init() {connectionFactory = rabbitTemplate.getConnectionFactory();createExchange(exchangeName);}private void createExchange(String exchangeName) {rabbitTemplate.execute(channel -> {channel.exchangeDeclare(exchangeName, "direct", false, true, null);return null;});}

用户登录

用户登录后，我们会创建一个“系统通知”队列。然后用户就会通过长连接形式，持续等待系统发出通知。

    private final ReentrantLock lock = new ReentrantLock();final private Map<String, SimpleMessageListenerContainer> listeners = new java.util.HashMap<>();public Flux<String> Login(String username) {createExclusiveQueue(username);createBanding(exchangeName, username, username);return Flux.create(emitter -> {SimpleMessageListenerContainer container = getListener(username, (Message message) -> {String msg = new String(message.getBody());System.out.println("Received message: " + msg);emitter.next(msg);});container.start();});}private void createExchange(String exchangeName) {rabbitTemplate.execute(channel -> {channel.exchangeDeclare(exchangeName, "direct", false, true, null);return null;});}private void createBanding(String exchangeName, String queueName, String routingKey) {rabbitTemplate.execute(channel -> {channel.queueBind(queueName, exchangeName, routingKey);return null;});}private SimpleMessageListenerContainer getListener(String queueName, MessageListener messageListener) {lock.lock();try {SimpleMessageListenerContainer listener = listeners.get(queueName);if (listener == null && messageListener != null) {listener = new SimpleMessageListenerContainer();listener.setConnectionFactory(connectionFactory);listener.setQueueNames(queueName);listener.setMessageListener(messageListener);listeners.put(queueName, listener);}return listener;} finally {lock.unlock();}}

Controller如下

package com.rabbitmq.chat.controller;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;import com.rabbitmq.chat.service.Core;import reactor.core.publisher.Flux;@RestController
@RequestMapping("/user")
public class UserController {@Autowiredprivate Core core;@PostMapping(value = "/login", produces = "text/event-stream")public Flux<String> login(@RequestParam String username) {return core.Login(username);}
}

发起聊天邀请

发起聊天邀请时，系统会预先创建一个聊天室（ChatRoomInfo ）。它包含上图中Fanout交换器、以及聊天双方需要订阅的消息队列。
这些创建完后，发起方就会等待对方发送的消息，也可以自己和自己聊天。因为消息队列已经创建好了，只是对方还没使用。

package com.rabbitmq.chat.service;import java.util.Map;
import java.util.concurrent.locks.ReentrantLock;import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;import jakarta.annotation.PostConstruct;
import lombok.Data;
import reactor.core.publisher.Flux;@Service
public class ChatRoom {@Autowiredprivate RabbitTemplate rabbitTemplate;private ConnectionFactory connectionFactory;@Dataprivate class ChatRoomInfo {private String exchange;private Map<String, String> usernameToQueuename;}private final Map<String, ChatRoomInfo> chatRooms = new java.util.HashMap<>();private final ReentrantLock lock = new ReentrantLock();   @PostConstructpublic void init() {connectionFactory = rabbitTemplate.getConnectionFactory();}public Flux<String> invite(String fromUsername, String toUsername) {String chatRoomName = getChatRoomName(fromUsername, toUsername);ChatRoomInfo chatRoomInfo = chatRooms.get(chatRoomName);if (chatRoomInfo == null) {createChatRoom(fromUsername, toUsername);}return talk(chatRoomName, fromUsername);}private void createChatRoom(String fromUsername, String toUsername) {String chatRoomName = getChatRoomName(fromUsername, toUsername);String exchangeName = chatRoomName;String fromQueueName = "queue-" + fromUsername + "-" + toUsername;String toQueueName = "queue-" + toUsername + "-" + fromUsername;rabbitTemplate.execute(action -> {action.exchangeDeclare(exchangeName, "fanout", false, true, null);action.queueDeclare(fromQueueName, false, true, false, null);action.queueDeclare(toQueueName, false, true, false, null);action.queueBind(fromQueueName, exchangeName, "");action.queueBind(toQueueName, exchangeName, "");return null;});lock.lock();try {ChatRoomInfo chatRoomInfo = new ChatRoomInfo();chatRoomInfo.setExchange(exchangeName);chatRoomInfo.setUsernameToQueuename(Map.of(fromUsername, fromQueueName, toUsername, toQueueName));chatRooms.put(chatRoomName, chatRoomInfo);} finally {lock.unlock();}}

接受邀请

被邀请方通过Core交换器得知有人要和它聊天。
然后接受邀请的请求会寻找聊天室信息，然后订阅聊天记录队列。

    public Flux<String> accept(String fromUsername, String toUsername) {String chatRoomName = getChatRoomName(fromUsername, toUsername);return talk(chatRoomName, toUsername);}private Flux<String> talk(String chatRoomName, String username) {ChatRoomInfo chatRoomInfo = chatRooms.get(chatRoomName);if (chatRoomInfo == null) {throw new IllegalArgumentException("Chat room not found");}String queueName = chatRoomInfo.getUsernameToQueuename().get(username);return Flux.create(emitter -> {SimpleMessageListenerContainer listener = new SimpleMessageListenerContainer();listener.setConnectionFactory(connectionFactory);listener.setQueueNames(queueName);listener.setMessageListener((Message message) -> {String msg = new String(message.getBody());System.out.println(username + " received message: " + msg);emitter.next(msg);});listener.start();});}

聊天

聊天的逻辑就是找到聊天室信息，然后向交换器发送消息。

    public void chat(String fromUsername, String toUsername, String message) {String chatRoomName = getChatRoomName(fromUsername, toUsername);ChatRoomInfo chatRoomInfo = chatRooms.get(chatRoomName);if (chatRoomInfo == null) {chatRoomName = getChatRoomName(toUsername, fromUsername);chatRoomInfo = chatRooms.get(chatRoomName);}if (chatRoomInfo == null) {throw new IllegalArgumentException("Chat room not found");}rabbitTemplate.convertAndSend(chatRoomInfo.getExchange(), "", fromUsername + ": " + message);}private String getChatRoomName(String fromUsername, String toUsername) {return fromUsername + "-" + toUsername + "-chat-room";}

Controller侧代码

package com.rabbitmq.chat.controller;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.PutMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;import com.rabbitmq.chat.service.ChatRoom;
import com.rabbitmq.chat.service.Core;import reactor.core.publisher.Flux;@RestController
@RequestMapping("/chat")
public class ChatController {@Autowiredprivate Core core;@Autowiredprivate ChatRoom chatRoom;@PutMapping(value = "/invite", produces = "text/event-stream")public Flux<String> invite(@RequestParam String fromUsername, @RequestParam String toUsername) {core.invite(fromUsername, toUsername);return chatRoom.invite(fromUsername, toUsername);}@PutMapping(value = "/accept", produces = "text/event-stream")public Flux<String> accept(@RequestParam String fromUsername, @RequestParam String toUsername) {core.accept(fromUsername, toUsername);return chatRoom.accept(fromUsername, toUsername);}@PostMapping("/send")public void send(@RequestParam String fromUsername, @RequestParam String toUsername, @RequestParam String message) {chatRoom.chat(fromUsername, toUsername, message);}
}

实验过程

在Postman中，我们先让tom登录，然后jerry登录。
在这里插入图片描述
在这里插入图片描述
在后台，我们看到创建两个队列
在这里插入图片描述
以及Core交换器的绑定关系也被更新
在这里插入图片描述
Jerry向Tom发起聊天邀请
在这里插入图片描述
可以看到Tom收到了邀请
在这里插入图片描述
同时新增了两个队列
在这里插入图片描述
以及一个交换器
在这里插入图片描述
在这里插入图片描述
Tom通过下面请求接受邀请
在这里插入图片描述
Jerry收到Tom接受了邀请的通知
在这里插入图片描述
后面它们就可以聊天了
在这里插入图片描述
在这里插入图片描述
它们的聊天窗口都收到了消息
在这里插入图片描述
在这里插入图片描述

总结

本文主要使用的知识点：

  • direct交换器以及其绑定规则
  • fanout交换器
  • 自动删除的交换器
  • 自动删除的队列
  • 只有一个消费者的队列
  • WebFlux响应式编程

代码工程

https://github.com/f304646673/RabbitMQDemo


http://www.ppmy.cn/devtools/56449.html

相关文章

C语言单链表的算法之插入节点

C语言单链表的算法之插入节点

一&#xff1a;访问各个节点中的数据 &#xff08;1&#xff09;访问链表中的各个节点的有效数据&#xff0c;这个访问必须注意不能使用p、p1、p2&#xff0c;而只能使用phead &#xff08;2&#xff09;只能用头指针不能用各个节点自己的指针。因为在实际当中我们保存链表的时…
阅读更多...
小脚本：文件保存后，自动上传到git

小脚本：文件保存后，自动上传到git

闲来无事写写小脚本 假设已经有一个脚本 echo off:START :: 获取当前脚本的路径 cd /d %~dp0:: 清空提交注释变量的值 set commit_msg:: 提示用户输入提交注释 set /p commit_msg请输入提交注释&#xff08;输入 n 退出&#xff09;::: 如果用户输入内容为 n&#xff0c;则退出…
阅读更多...
three.js - matcap材质（MeshMatcapMaterial）

three.js - matcap材质（MeshMatcapMaterial）

说一下matcap纹理 先总结&#xff1a;MeshMatcapMaterial材质&#xff0c;通过采样含有光照信息的贴图来模拟光照效果。这种材质特别适用于模拟静态光源下的光照&#xff0c;并且&#xff0c;因其简单性和快速性而被广泛应用于各种场景。但是&#xff0c;由于其性能考虑&#x…
阅读更多...
el-tree结构清空选中节点

el-tree结构清空选中节点

<el-tree:data"data"show-checkboxdefault-expand-allnode-key"id"ref"tree"highlight-current:props"defaultProps"> </el-tree>this.$refs.tree.setCheckedKeys(this.$refs.tree.getCheckedNodes(),false);
阅读更多...
【C++笔记整理—第1期】

【C++笔记整理—第1期】

C笔记整理 本系列是C语言基础语法的学习笔记整理。 文章目录 C笔记整理结构体定义及初始化结构体数组用指向结构体变量的指针构成链表函数传参的三种形式动态分配和撤销内存空间用typedef声明新的类型名 类和对象概述特点面向对象的软件工程 类的声明和对象的定义声明类类型 …
阅读更多...
如何使用 asyncio.log 启用 Debug 日志

如何使用 asyncio.log 启用 Debug 日志

如何使用 asyncio.log 启用 Debug 日志 1. 导入必要的模块2. 配置日志记录3. 启用 asyncio 的调试日志4. 示例代码总结 在 Python 中使用 asyncio 模块时&#xff0c;有时我们需要查看调试日志来了解程序的运行情况。本文将介绍如何使用 asyncio.log 启用调试日志。 1. 导入必…
阅读更多...
Python+requests+pytest+excel+allure 接口自动化测试实战

Python+requests+pytest+excel+allure 接口自动化测试实战

定义&#xff1a; Unittest是Python标准库中自带的单元测试框架&#xff0c;Unittest有时候也被称为PyUnit&#xff0c;就像JUnit是Java语言的标准单元测试框架一样&#xff0c;Unittest则是Python语言的标准单元测试框架。 Pytest是Python的另一个第三方单元测试库。它的目的…
阅读更多...
vue3 window.location 获取正在访问的地址，也可以通过useRoute来获取相关信息。

vue3 window.location 获取正在访问的地址，也可以通过useRoute来获取相关信息。

1、一般我们在开发的vue3项目的时候&#xff0c;地址是这样&#xff1a;http://192.168.1.101:3100/#/login 然后我们在布署完成以后一般是这样https://xxx.yyyyy.com/uusys/#/login 其实xxx可以是www&#xff0c;也可以是一个二级域名 yyyyy.com是域名&#xff0c;uusys一般…
阅读更多...
最新文章

Copyright @ 2022~2023