RabbitMQ 客户端 连接、发送、接收处理消息

news/2025/1/19 16:57:08/

RabbitMQ 客户端 连接、发送、接收处理消息

一. RabbitMQ 的机制跟 Tcp、Udp、Http 这种还不太一样

RabbitMQ 服务，不是像其他服务器一样，负责逻辑处理，然后转发给客户端
而是所有客户端想要向 RabbitMQ服务发送消息，

第一步：创建一个链接 RabbitMQ 服务的连接

需要传入 RabbitMQ服务地址、用户名、密码，然后在连接代码中传入一个 queue 的字符串作为 标志
连接成功后，RabbitMQ服务上就可以看到这个链接了
如下图，可以看到有一个 Name = queueL1 的连接，后边有链接状态、消息数
Ready 和 Total 都是 0
在这里插入图片描述

向 RabbitMQ 发送消息的：

(1) 如果没有建立连接，执行第一步，建立一个链接
(2) 通过 发送消息接口向 RabbitMQ 服务 发消息
(3) RabbitMQ 服务接收到消息，只是按照连接的 queue 分别把消息放在自己名字的 queue 下， RabbitMQ 服务只是存着客户端发送的消息，服务什么都不处理

向 RabbitMQ 服务发送几条消息
下图可以看到 queueL1 的队列已经接收了 5 条消息，这五条消息如果没有客户端接收处理，就一直在这存着
在这里插入图片描述

接收 RabbitMQ 服务消息：

(1) 如果没有建立连接，执行第一步，建立一个链接
(2) 注册接收消息接口，在 RabbitMQ 中叫 消费消息，可以标记消费消息后是否将 RabbitMQ 的数据删除
(3) 如果 RabbitMQ 服务收到消息，就转发给 注册接收消息接口的 连接，如果接收的连接标记了 AutoDelete，那么发送给客户端后，RabbitMQ 就会将消息从消息队列中删除

注册接收消息，我的客户端就会收到 RabbitMQ 发送过来的消息，消息中包含发送上来的消息内容，还有发送消息的 queue 名字

此时再看，就会发现 Ready 和 Total 又变成 0 了
在这里插入图片描述

为什么上面讲解中将 接收 RabbitMQ 服务消息、向 RabbitMQ 发送消息的 分开说
是因为 RabbitMQ 发送消息就仅仅是发消息，发送完就不管了
而 RabbitMQ 的消费消息(接收消息) 也仅仅是接收消息，它不管是谁发的消息，只要是发送的 RabbitMQ 服务的消息，它都能接收，

(3.1) 比如我创建了 一个 连接，queue名为 xxxA，
它发送了消息 “Hello World”，
xxxA 连接自己又注册了 消费消息(接收消息)，那么xxxA 自己就会接收到 xxxA 队列发送的 Hello World 信息

(3.2) 我又创建了 新的连接，queue 名还是 xxxA
那么新的连接也可以收到 (3.1) 发的 消息 HelloWorld

二. 客户端连接服务器

  1. 实例化一个 连接 RabbitMQ 服务的客户端连接
    实例化需要传入 服务地址、端口、用户名、密码

    using RabbitMQ.Client;
    using System;
    using System.Threading.Tasks;
    using RabbitMQ.Client.Exceptions;
    using UnityEngine;
    using System.Text;
    using RabbitMQ.Client.Events;

    namespace Network
    {
    ///
    /// RabbitMQ 创建一个链接
    /// 供 RabbitMQReceive、RabbitMQSend 使用
    ///
    public class RabbitMQConnect
    {
    private RabbitMQConnectData connectData;

        private ConnectionFactory factory;private IChannel channel;private IConnection connection;private NetWorkState state;private Action<string, byte[]> receivedCallBack;private const int TimeOut = 10; //连接超时 10 秒private bool dispose = false;public RabbitMQConnect(RabbitMQConnectData connectData){this.connectData = connectData;State = NetWorkState.Disconnected;dispose = false;}public string Queue{get { return connectData.queue; }}public NetWorkState State{get { return state; }private set { state = value; }}public IChannel Channel{get { return channel; }}/// <summary>/// 网络是否连接中/// </summary>public bool IsConnect{get{if (null == channel || null == connection){return false;}return channel.IsOpen && connection.IsOpen;}}public async Task StartConnect(){if (State == NetWorkState.Connecting){await Task.Delay(TimeOut * 1000);}if (State == NetWorkState.Connected){return;}// 创建连接工厂// 如果初始化失败，不会启动恢复连接//factory = new ConnectionFactory()//{//    HostName = hostName, // 替换为你的 RabbitMQ 服务器地址//    UserName = userName, // 替换为用户名//    Password = password  // 替换为密码//};string url = $"amqp://{connectData.userName}:{connectData.password}@{connectData.hostName}:{connectData.port}"; //string.Format("amqp://unity:unity@139.9.137.14:5672");factory = new ConnectionFactory(){Uri = new Uri(url)};// 自动恢复连接factory.AutomaticRecoveryEnabled = true;// 如果由于异常导致恢复失败（例如RabbitMQ节点仍然不可达），它将在固定的时间间隔（默认为5秒）后重试。间隔时间可配置如下// Connection.CloseAsync 关闭的连接不会启动自动恢复连接factory.NetworkRecoveryInterval = TimeSpan.FromSeconds(10);factory.TopologyRecoveryEnabled = true;while (State != NetWorkState.Connected){if (!dispose){await Connect();}}await Task.Delay(1);if (!string.IsNullOrEmpty(connectData.receiveQueeu)){await BasicConsumer();}}private async Task Connect(){try{State = NetWorkState.Connecting;// 异步创建连接connection = await factory.CreateConnectionAsync();channel = await connection.CreateChannelAsync();// 声明队列QueueDeclareOk queueDeclareOk = await channel.QueueDeclareAsync(queue: connectData.queue,durable: false,exclusive: false,autoDelete: false,arguments: null);/*autoDelete = true：没有消费者时队列自动删除，通常用于临时或一次性的队列。autoDelete = false：队列不会自动删除，通常用于需要长期存在的队列。选择是否设置 autoDelete = true 取决于你是否希望队列在没有消费者时自动删除。如果你的队列是临时的、一次性的，那么使用 autoDelete = true 会更适合；如果队列是长期需要使用的，则设置为 autoDelete = false 会更为合适 */State = NetWorkState.Connected;// 设置消费者的预取计数为10，允许同时处理10条消息await channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 10, global: false);Debug.Log("RabbitMQ Connect Success");GameNotifycation.GetInstance().Notify<NetWorkState>(ENUM_MSG_TYPE.MSG_NETWORK_STATE_CHANGE, State);}catch (BrokerUnreachableException e){await Task.Delay(5000);State = NetWorkState.ConnectFailed;Debug.LogError("ConnectError:" + e.ToString());// apply retry logic}await Task.Delay(1);}/// <summary>/// 发送消息/// exchange:   要发布消息的交换机名称。/// routingKey: 路由键，决定消息应该路由到哪个队列。/// mandatory:  如果设置为 true，RabbitMQ 会确保消息至少被投递到一个队列。如果没有队列接收该消息，RabbitMQ 会触发 basic.return。/// immediate:  如果设置为 true，RabbitMQ 会在消息无法立即被消费时丢弃消息。/// basicProperties: 消息的属性，类型为 IBasicProperties。这些属性可以设置消息的优先级、持久性等。/// body: 消息体的字节数组。/// /// BasicPublishAsync 方法 没有返回消息投递的结果。它仅仅表示“请求已经被成功发送到 RabbitMQ 的交换机”。如果发布操作成功，Task 会正常完成，不会抛出异常。你可以通过异常处理来捕获潜在的错误。/// </summary>/// <param name="msg"></param>public async Task SendAsync(string message){if (!IsConnect){UnityEngine.Debug.Log("Send not IsConnect");await StartConnect();}try{IChannel channel = Channel;var body = Encoding.UTF8.GetBytes(message);var props = new BasicProperties();props.ContentType = "text/plain";props.DeliveryMode = DeliveryModes.Transient;await channel.BasicPublishAsync(exchange: "",routingKey: Queue,mandatory: false,basicProperties: props,body: body).ConfigureAwait(false);//Debug.Log($"[x] Sent: Complete");}catch (Exception ex){UnityEngine.Debug.LogError($"Error publishing message: {ex.Message}");}}/// <summary>/// 设置接收消息回调/// </summary>/// <param name="receivedCallBack"></param>public void SetReceive(Action<string, byte[]> receivedCallBack){this.receivedCallBack = receivedCallBack;}/// <summary>/// 创建异步消费者/// </summary>/// <returns></returns>public async Task<string> BasicConsumer(){if (!IsConnect){await StartConnect();}var consumer = new AsyncEventingBasicConsumer(Channel);// 处理消息的异步回调逻辑consumer.ReceivedAsync += ReceivedAsync;// 开始消费string result = await Channel.BasicConsumeAsync(queue: connectData.receiveQueeu,  // 指定消费者要监听的队列名称autoAck: false,        // 决定是否自动确认消息。如果 true，消息在交付时会自动确认。如果 false，则需要手动调用 BasicAck 确认消息consumer: consumer);  // 指定消息的处理方式，通过实现 IBasicConsumer 接口来定义如何处理从队列中接收到的消息/*autoAck = true：消息一旦传递给消费者，RabbitMQ 就认为该消息被成功处理，无需再确认。autoAck = false：消费者需要显式地调用 channel.BasicAck 来确认消息的处理，通常用于消息处理失败时能够重试消息。*/return result;}/// <summary>/// 异步接收消息/// 如果 Channel.BasicConsumeAsync 方法中 autoAck 设置为 true，那么 channel.BasicAckAsync 调用是不允许的/// 想在  Channel.BasicConsumeAsync 消费消息收到消息时 调用 channel.BasicAckAsync，必须将 Channel.BasicConsumeAsync 方法中 autoAck 设置为 false/// </summary>/// <param name="sender"></param>/// <param name="eventArgs"></param>/// <returns></returns>private async Task ReceivedAsync(object sender, BasicDeliverEventArgs eventArgs){try{//Debug.Log("ReceivedAsync");AsyncEventingBasicConsumer consumer = sender as AsyncEventingBasicConsumer;string queue = consumer.Channel.CurrentQueue;var body = eventArgs.Body.ToArray();receivedCallBack?.Invoke(queue, body);// 模拟异步任务处理（比如访问数据库或调用其他服务）await channel.BasicAckAsync(eventArgs.DeliveryTag, false);}catch (Exception ex){Debug.LogError($"Error processing message: {ex.Message}");// 如果处理失败，可以拒绝并重新入队（可选）//await Channel.BasicNackAsync(eventArgs.DeliveryTag, multiple: false, requeue: true);}await Task.Delay(1);}/// <summary>/// 关闭连接/// </summary>public async void Dispose(){dispose = true;// 先关闭通道、再关闭连接if (channel != null)  // 通道关闭{await channel.CloseAsync();channel = null;}if (connection != null)  // 连接关闭{UnityEngine.Debug.Log("ConnectDispose");await connection.CloseAsync();connection = null;}await Task.Delay(1);}
}

    }

RibbitMQ 服务通过 queue 来区分每一个连接的客户端，代码部分如下

QueueDeclareOk queueDeclareOk = await channel.QueueDeclareAsync(queue: queue,durable: false,exclusive: false,autoDelete: false,arguments: null);

  1. 客户端实例

  2. 测试用例

    using UnityEngine;
    using Network;
    using LitJson;
    using System.Text;
    using System.Collections;
    using System.Collections.Generic;

    public class RabbitMQDemo : MonoBehaviour
    {
    // 客户端
    private RabbitMQConnect rabbitMQConnect;
    private Queue receiveQueue = new Queue();

    void Start()
{RabbitMQConnectData connectData = new RabbitMQConnectData();connectData.queue = "TestA";connectData.receiveQueeu = "TestA";connectData.hostName = "XXX.XXX.XXX.XXX";connectData.port = "5672";connectData.userName = "unity";connectData.password = "unity";// 实例化rabbitMQConnect = new RabbitMQConnect(connectData);rabbitMQConnect.SetReceive(Receive);StartConnect();
}private async void StartConnect()
{await rabbitMQConnect.StartConnect();
}private async void Send(string meg)
{await rabbitMQConnect.SendAsync(meg);
}private void Receive(string queue, byte[] byteData)
{var json = Encoding.UTF8.GetString(byteData);UnityEngine.Debug.Log($"[x] ReceivedAsync: {json}");receiveQueue.Enqueue(netWorkData);
}private int number = 1000;
// Update is called once per frame
void Update()
{if (Input.GetKeyDown(KeyCode.A)){++number;Send("Hello RabbitMQ:" + number);}DispatchMessage();
}private void DispatchMessage()
{if (receiveQueue.Count <= 0){return;}string json = receiveQueue.Dequeue();
}private void OnDestroy()
{Debug.LogError("OnDestroy");rabbitMQConnect.Dispose();
}

    }

    /// <summary>
/// 网络连接状态
/// </summary>
public enum NetWorkState
{// init/// <summary>/// 关闭/断开连接/// </summary>Closed,// client/// <summary>/// 已经建立连接/// </summary>Connected,/// <summary>/// 正在请求连接/// </summary>Connecting,/// <summary>/// 连接失败/// </summary>ConnectFailed,// both/// <summary>/// 连接超时/// </summary>Timeout,/// <summary>/// 断开连接/// </summary>Disconnected,
}

扩展
可以在 网页上 Overview 页面，找到 Ports and contexts 部分
可以看到每种协议对应的端口是不一样的
每种协议都有一种独立的连接方式
需要根据自己选择的协议拼接路径

比如 我上面代码使用的 http 方式

    string localHost = "localhost"; // ip如 xxx.xxx.xxx.xxxstring userName = "用户名";string password = "密码";// 创建连接工厂// 如果初始化失败，不会启动恢复连接factory = new ConnectionFactory(){HostName = hostName, // 替换为你的 RabbitMQ 服务器地址UserName = userName, // 替换为用户名Password = password  // 替换为密码};

amqp 协议连接方式如下

	string url = $"amqp://{userName}:{password}@{hostName}:{port}"; factory = new ConnectionFactory(){Uri = new Uri(url)};

在这里插入图片描述


http://www.ppmy.cn/news/1564462.html

相关文章

正则表达式 匹配特定字符后的所有字符

正则表达式 匹配特定字符后的所有字符

在处理文本数据时&#xff0c;正则表达式&#xff08;RegularExpressions&#xff0c;简称Regex&#xff09;是一种非常强大的工具&#xff0c;它可以用来搜索、匹配和替换文本中符合特定模式的字符串。 首先&#xff0c;明确我们的目标是匹配完整的URL&#xff0c;并获取它之…
阅读更多...
[BrainShadow-V1] VR头戴设备统计报告

[BrainShadow-V1] VR头戴设备统计报告

Brain-Shadow-V1 EventVR headsetsReported byXiao enDate2025/01/15Version1.0 HTC Vive Pro 2 Pro HTC Vive Pro 2 是一款高端虚拟现实头显&#xff0c;配备双 2.5K 显示屏&#xff0c;组合分辨率达到 48962448&#xff0c;提供 120 的视场角和 120Hz 的刷新率。该设备支持…
阅读更多...
前端小知识 鼠标穿透 pointer-events: none；

前端小知识 鼠标穿透 pointer-events: none；

为什么会说到这个呢&#xff1f;是我觉得没有识别出来&#xff0c;然后就导致了这样的问题&#xff0c;这种情况不应该发生。我写了如下这样一段代码&#xff0c;但是发现当自己选择时间的时候无法选择。然后就发现变成了光标在闪烁。这样其实就是因为我选择到了这个input框的鼠…
阅读更多...
Spark Streaming的核心功能及其示例PySpark代码

Spark Streaming的核心功能及其示例PySpark代码

Spark Streaming是Apache Spark中用于实时流数据处理的模块。以下是一些常见功能的实用PySpark代码示例&#xff1a; 基础流处理&#xff1a;从TCP套接字读取数据并统计单词数量 from pyspark import SparkContext from pyspark.streaming import StreamingContext# 创建Spar…
阅读更多...
Docker 搭建mysql 连接超时问题，xxl-job启动mysql连接报错，禁用dns

Docker 搭建mysql 连接超时问题，xxl-job启动mysql连接报错，禁用dns

1.本地连接Navicat报错信息&#xff0c;猜测是navicat默认连接超时导致的&#xff0c;后面换成idea一个插件虽然慢但连接上了 2013 - Lost connection to MySQL server at reading initial communication packet 2.启动xxl-job会报错&#xff0c;网上有人mysql驱动与数据库不匹…
阅读更多...
PyTorch使用教程(6)一文讲清楚torch.nn和torch.nn.functional的区别

PyTorch使用教程(6)一文讲清楚torch.nn和torch.nn.functional的区别

torch.nn 和 torch.nn.functional 在 PyTorch 中都是用于构建神经网络的重要组件&#xff0c;但它们在设计理念、使用方式和功能上存在一些显著的区别。以下是关于这两个模块的详细区别&#xff1a; 1. 继承方式与结构 torch.nn torch.nn 中的模块大多数是通过继承 torch.nn…
阅读更多...
使用jupyter notebook没有正常打开浏览器的几种情况解决

使用jupyter notebook没有正常打开浏览器的几种情况解决

迅速记录前期 1.下载 https://www.anaconda.com/products/individual 2.安装 直接默认安装就行 3.打开jupyter notebook 在开始菜单里面可以找到 4.遇到的问题解决 1.运行jupyter notebook&#xff0c;黑窗口自动关了 每次黑窗口迅速的加载完就自己关掉了 也没有打开新…
阅读更多...
麒麟kylin YUM配置管理

麒麟kylin YUM配置管理

一、概述 YUM&#xff08;Yellowdog Updater Modified&#xff09;是一个用于RPM包管理系统的自动化更新工具&#xff0c;最初由Duke University的Linux实验室开发。它简化了基于RPM的Linux发行版&#xff08;如Red Hat Enterprise Linux, CentOS, Fedora等&#xff09;上的软…
阅读更多...
最新文章

Copyright @ 2022~2023