小5聊C#控制台使用RabbitMQ进行简单的消息生产和消费

Posted 2022-07-28 小5聊

消息队列MessageQueue，简称MQ

处理非主业务功能，同时也能在高并发下异步处理的功能，使用MQ就是最佳选择

虽然开发了很多年，现在回想起来，还真没有自己主导框架在实际项目中运用MQ，都是现有框架往里面加功能。

所以此处，也只是在简单的RabbitMQ安装和使用

1、RabbitMQ概念

1）实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件）

2）RabbitMQ服务器是用Erlang语言编写

3）RabbitMQ，支持大多操作系统和编程语言

4）exchange，交换器，TA接收生产者消息，TA把消息推送到队列中

5）4种消息类型，direct、topic、headers、fanout

2、RabbitMQ官网下载

看到这些官网都是英文，有些感慨，为啥我们就没能开发这么流行的技术呢

1）点击跳转页面

2）点击-Chocolatey or Installer (recommended=推荐)

3）点击，rabbitmq-server-3.10.6

跳转页面，页面滚动到中间即可看到如下内容，目前版本是-3.10.6

 

3、安装Erl开发语言环境软件

1）如果没有先安装Erlang，那么会出现以下提示

 

 2）跳转到页面

3）根据自己系统下面

不过一般windows都是64位，需要点开那个箭头，otp_win64_25.0.3

4）之后就是下一步

默认安装目录：C:\\Program Files\\Erlang OTP

5）查看环境变量

安装的时候，已经自动添加了ERLANG_HOME变量对应的安装路径

%ERLANG_HOME%\\bin

6）cmd命令查看erlang版本

 

4、安装RabbitMQ

1）运行rabbitmq-server-3.10.6.exe

 2）下一步，无需其他操作

默认安装路径：C:\\Program Files\\RabbitMQ Server

3）找到RabbitMQ的输出目录

C:\\Users\\自己操作系统名称\\AppData\\Roaming\\RabbitMQ

•  默认没有配置文件，需要自己添加，文件

[ 
  rabbit,
  [ 
    tcp_listeners,
    [ "0.0.0.0" ,5672]
  ]
,
  rabbitmq_management,
  [ 
    listener,
    [
      port , 15672,
      ip , "0.0.0.0",
      ssl,false
    ]
  ]
].

• 文件命名如下

rabbitmq.config

4）cmd命令重新启动RabbitMQ服务

cmd定位到安装目录，如下

C:\\Program Files\\RabbitMQ Server\\rabbitmq_server-3.10.6\\sbin

分别执行如下4条语句：服务》停止》移除》安装》开启

  rabbitmq-service stop

  rabbitmq-service remove

  rabbitmq-service install

  rabbitmq-service start

• 同样可以在《服务》里找到它并重启

 

5）访问RabbitMQ的Web端管理工具

默认的账号密码就是guest

 5、设置账号和权限

1）登录进入界面，打开Admin和Users，操作如下

可能不同版本稍微有点区别，记得一定要设置用户角色分类，就是Set区域，可以选择Admin管理员

2）用户角色分类

编号	角色名称	备注
1	Admin/Administrator-超级管理员	可登录管理控制台，拥有所有权限，可查看用户所有信息，以及操作用户和策略（policy），
2	Monitoring-监控者	可登录，可查看节点信息，比如进程数、内存‘磁盘使用情况
3	Policymaker-策略制定者	可登录，对policy策略进行管理，但不可查看节点西悉尼
4	Management-普通管理者	仅登录，无查看策略和节点权限
5	Impersonator-模拟者
6	None-其他	不可登录，普通得生产者和消费者

 3）添加没有角色得用户进行MQ连接

• 添加none角色用户

•  出现如下提示

• 点击进入user_none界面

This user does not have permission to access any virtual hosts.
Use "Set Permission" below to grant permission to access virtual hosts.
此用户没有访问任何虚拟主机的权限。
使用下面的“设置权限”授予访问虚拟主机的权限。

• 设置权限

 点击Set permission按钮即可

•  再次访问

• 消息队列信息

6、消息生产端控制台代码

生产发布消息到RabbitMQ消息中心，等待订阅者消费消息

• 发送了3条消息到队列

•  代码

/*
    NuGet
    1、RabbitMQ.Client - 6.0.0
*/

using RabbitMQ.Client;
using System;
using System.Text;

namespace rabbitMQ.Client.producer

    class Program
    
        /// <summary>
        /// 消息生产者（消费用户）
        /// </summary>
        /// <param name="args"></param>
        static void Main(string[] args)
        
            //创建连接工厂
            ConnectionFactory factory = new ConnectionFactory
            
                UserName = "user_none",
                Password = "123123",
                HostName = "192.168.1.102",
                Port = 5672
            ;

            //创建连接
            var connection = factory.CreateConnection();

            //创建通道
            var channel = connection.CreateModel();

            //声明一个队列
            channel.QueueDeclare("My.First.Queue2", false, false, false, null);

            string msg;
            do
            
                msg = Console.ReadLine();
                Console.WriteLine($"发送的消息：msg");

                var sendbytes = Encoding.UTF8.GetBytes(msg);

                channel.BasicPublish("", "My.First.Queue2", null, sendbytes); //发布
            
            while (!msg.Trim().ToLower().Equals("exit"));

            channel.Close();
            connection.Close();
        
    

7、消息消费端控制台代码

消息订阅者，从消息中心获取未消费得消息

• 消费完成后，队列里得消息就变为0 了，表示已经确认消费完成了 

• 代码如下

 

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Text;

namespace rabbitMQ.Client.consumer

    class Program
    
        /// <summary>
        /// 消息消费者（后台处理方法）
        /// </summary>
        /// <param name="args"></param>
        static void Main(string[] args)
        
            //创建连接工厂
            ConnectionFactory factory = new ConnectionFactory
            
                UserName = "user_none",
                Password = "123123",
                HostName = "192.168.1.102",
                Port = 5672
            ;

            //创建连接
            var connection = factory.CreateConnection();

            //创建通道
            var channel = connection.CreateModel();

            channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
            //接收到的消息处理事件
            EventingBasicConsumer Recipient = new EventingBasicConsumer(channel);
            channel.BasicConsume("My.First.Queue2", false, Recipient);

            Recipient.Received += (ch, ea) =>
            
                var RecipientMsg = Encoding.UTF8.GetString(ea.Body.ToArray());
                Console.WriteLine($"收到消息：RecipientMsg");

                //确认该消息已被处理
                channel.BasicAck(ea.DeliveryTag, false);
                Console.WriteLine($"消息已确认处理【ea.DeliveryTag】");
            ;

            try
            

            
            catch
            
                Console.WriteLine($"服务中心还未有队列消息！");
            

            Console.ReadKey();
            channel.Dispose();
            connection.Close();


        
    

【遇到的情况】

为了梳理安装过程，先后把Erl和RabbitMQ软件卸载了，但是RabbitMQ.Client依然是可以运行的，这是不是说明，开启RabbitMQ这个服务，其实已经有独立进程运行了

1）带着这个疑问查看了任务管理器，Erl卸载了，居然还在

2）结束上面进程后，再次运行，发现已经无法连接到目标了

 3）消费端，找不到队列会报错

解决方法：

1）如果是同时运行，那么可在消费端添加一个Sleep休眠，确保消费端在生产端之后运行，这样生产端就能在消费端获取队列时生成队列。

2）先在rabbitmq的web端先手动添加队列名称（基于目前理解，有不对的地方望大家指出）