消息中间件RabbitMQ02:账号的注册、点对点推送信息

发布于:2025-05-01 ⋅ 阅读:(28) ⋅ 点赞:(0)

一、默认用户登录和账号注册

1.登录

        安装好了RMQ之后,我们可以访问如下地址: 

RabbitMQ Management

        输入默认的管理员密码,4.1.0的管理员账号和密码是:

guest
guest

 2.添加账号

consumer
consumer

添加成功后:

角色的对比: 

标签 (中文) 标签 (英文) 描述
管理 Management 用户可以访问管理插件。
政策制定者 Policymaker 用户可以访问管理插件,并管理他们有权访问的虚拟主机(vhosts)的策略和参数。
监听 Monitoring 用户可以访问管理插件,查看所有连接、通道以及节点相关信息。
管理员 Administrator 用户可以做监控可以做的所有事情,管理用户、虚拟主机和权限,关闭其他用户的连接,管理所有虚拟主机的策略和参数。

3.添加用户权限

         我们刚刚添加了用户consumer,但是他不具备任何权限,我们可以为他配置权限,先点击进去

 

 二、HelloWorld入门Demo

1.编写消费者代码并运行

class Program
{
    static async Task Main(string[] args)
    {

        // 01 RabbitMQ 配置
        string hostname = "localhost";  // RabbitMQ 主机名
        string username = "consumer";   // RabbitMQ 账号
        string password = "consumer";   // RabbitMQ 密码
        string queueName = "hello";     // 队列名称,和生产者保持一致
        var factory = new ConnectionFactory() { HostName = hostname, UserName = username, Password = password };

        // 02 创建连接
        using (var connection = await factory.CreateConnectionAsync())
        using (var channel = await connection.CreateChannelAsync())
        {

            // 03 声明一个队列(队列名称,MQ持久化 重启不丢失=true,队列私有化 仅供第一个消费者使用=true,无人使用 自动删除=true ,附加参数)
            await channel.QueueDeclareAsync(queueName, false, false, false, null);

            // 04 创建消费者实例
            var consumer = new AsyncEventingBasicConsumer(channel);

            // 05 设置消息到达时的回调函数(相当于接收消息后,需要执行的方法)
            consumer.ReceivedAsync +=  async (model, ea) =>
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);//将消息转换为字符串
                Console.WriteLine($"消费者接受了信息:{message}");

                // 处理完消息后手动签收确认消息
                await channel.BasicAckAsync(ea.DeliveryTag, false);
                Console.WriteLine("消息已确认: " + message);
            };

            // 06 启动消费者,开始接收消息(队列名称,需要手动签收消息=false,回调函数)
            await channel.BasicConsumeAsync(queueName, false, consumer);

            Console.WriteLine("消费者可点击Enter退出...");
            Console.ReadLine();
        }
    }
}

此时我们可以观察到,管理网站的变化:

2.编写生产者代码并运行 

class Program
{
    static async Task Main(string[] args)  // 将Main方法修改为异步方法
    {
        // 01 RabbitMQ 配置
        string hostname = "localhost";  // RabbitMQ 主机名
        string username = "guest";      // RabbitMQ 账号
        string password = "guest";      // RabbitMQ 密码
        string queueName = "hello";     // 队列名称
        var factory = new ConnectionFactory() { HostName = hostname, UserName = username, Password = password };

        // 02 创建连接
        using (var connection = await factory.CreateConnectionAsync())  // 使用CreateConnectionAsync方法
        using (var channel = await connection.CreateChannelAsync())
        {
            // 03 声明一个队列(队列名称,MQ持久化 重启不丢失=true,队列私有化 仅供第一个消费者使用=true,无人使用 自动删除=true ,附加参数)
            await channel.QueueDeclareAsync(queueName, false, false, false, null);

            // 04 编辑消息内容(存成字节数组形式)
            string message = "Hello, World!";
            byte[] body = Encoding.UTF8.GetBytes(message);

            // 05 发送消息到队列(交换机名称 没用到可为空,队列名称,字节数组形式的消息内容)
            await channel.BasicPublishAsync(string.Empty, queueName, body);
            Console.WriteLine($"生产者发送了消息: {message}");
        }

        Console.WriteLine("生产者可点击Enter退出...");
        Console.ReadLine();
    }
}

此时观察控制台程序,说明已经完成“生产者-MQ-消费者”流程:

3.查看队列消费情况

三、知识补充

1.连接问题

上述代码未指定端口,是因为ConnectionFactory默认指定了5672端口

var factory = new ConnectionFactory() { HostName = hostname, UserName = username, Password = password };

2.重复消费问题

        RabbitMQ 使用轮询机制将消息分发给多个消费者,每个消息只会被一个消费者处理,确保同一条消息不会被重复消费。消费者通过手动确认消息来保证消息的可靠消费。

3.未确认消费问题

        如果消费者1未确认消息,生产者不会将该消息发送给消费者2,直到消费者1确认或消息超时。RabbitMQ 会将消息保留在消费者1的队列中,直到它确认消息

4.拒绝消费问题

        如果消费者1拒绝消费并且要求重新入队,RabbitMQ 会将消息发送给其余消费者(不含消费者1)消费,可以用BasicRejectAsync方法实现。

consumer.ReceivedAsync +=  async (model, ea) =>
{
    if(true)
    {
        // 处理完消息后手动签收确认消息
        await channel.BasicAckAsync(ea.DeliveryTag, false);
    }
    else
    {
        //拒绝签收,并且重新入队给其余消费者消费
        await channel.BasicRejectAsync(ea.DeliveryTag, requeue: true);
    }

};

5.消息内容的类型问题

        RabbitMQ 本身只处理字节数据,因此如果要发送一个对象,需要先把对象序列化为字符串,字符串再转化为字节流

        生产者代码可以参照这种写法:

// 创建一个对象
var person = new Person { Name = "Alice", Age = 30 };
// 序列化对象为 JSON 字符串
string jsonMessage = JsonSerializer.Serialize(person);
// 转换为字节数组
byte[] body = Encoding.UTF8.GetBytes(jsonMessage);

         消费者代码可以参考这种写法:

consumer.Received += async (model, ea) =>
{
    // 获取消息的字节数组
    var body = ea.Body.ToArray();
    // 转换字节数组为 JSON 字符串
    string jsonMessage = Encoding.UTF8.GetString(body);
    // 反序列化为对象
    var person = JsonSerializer.Deserialize<Person>(jsonMessage);

};