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

消息队列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）用户角色分类

 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
    {
        /// 
        /// 消息生产者（消费用户）
        /// 
        /// 
        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
    {
        /// 
        /// 消息消费者（后台处理方法）
        /// 
        /// 
        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端先手动添加队列名称（基于目前理解，有不对的地方望大家指出）