netcore之RabbitMQ入门

简介

 RabbitMQ是实现了高级消息队列协议（AMQP）的开源消息中间件，基于Erlang语言编写。

AMQP是什么

AMQP 0-9-1（高级消息队列协议）是一种消息传递协议，它允许一致的客户端应用程序与一致的消息传递中间件代理进行通信。

消息传递代理接收来自发布者（发布它们的应用程序，也称为生产者）的消息，并将它们路由到消费者（处理它们的应用程序）。

 由于它是一个网络协议，发布者、消费者和代理都可以驻留在不同的机器上。

 AMQP 0-9-1模型简介

AMQP 0-9-1模型具有以下世界视图：消息发布到交换，这通常与邮局或邮箱进行比较。交换然后使用名为绑定的规则将消息副本分发到队列。然后，代理将消息传递给订阅队列的消费者，或者消费者根据需要从队列获取/拉取消息。

发布消息时，发布者可以指定各种消息属性（消息元数据）。有些元数据可以由代理使用，但是，其余的元数据对代理是完全不透明的，只能由接收消息的应用程序使用。

 网络不可靠，应用程序可能无法处理消息，因此AMQP 0-9-1模型具有消息确认的概念：当消息传递给消费者时，消费者会自动或在应用程序开发人员选择时立即通知代理。当消息确认正在使用时，代理将仅在收到消息（或消息组）通知时从队列中完全删除消息。

 例如，在某些情况下，当消息无法路由时，消息可能会返回给发布者、丢弃，或者，如果代理实现扩展，则将消息放入所谓的“死信队列”。发布者通过使用某些参数发布消息来选择如何处理这种情况。

 队列、交换和绑定统称为AMQP实体。

交换和交换类型

Rabbitmq中的核心思想，生产者不会把消息直接发送到队列当中，一般生产者是向交换机发送消息，交换机把消息推送到队列当中。

交换机是发送消息的实体，交换机接收消息并将消息路由到零个或者多个队列当中，使用的路由算法取决于绑定的交换类型和规则，因此AMQP 0-9-1提供了以下四种交换类型：

• Direct exchange    直接交换
• Fanout exchange  主题交换
• Topic exchange      头交换
• Headers exchang    扇形交换

除了交换类型之外，还使用许多属性声明交换其中最重要的是：

• 耐久性（Durability）       ：交易所在经纪人重启后仍能存活
• 自动删除（Auto-delete）：当最后一个队列与其解除绑定时，将删除Exchange
• 参数（arguments）         ：可选，由插件和特定于代理的功能使用

交换可以是持久的，也可以是暂时的。持久性交易所能在经纪重启后存活下来，而短暂性交易所则不能（当经纪重新上线时，必须重新申报）。并非所有的场景和用例都需要持久的交换。

本文主要记录了学习RabbitMQ

开发准备

• RabbitMQHelper   ：该工程主要封装了RabbitMQ的公用方法
• RabbitMQClient    ：该工程为生产者
• RabbitMQServer  ：该工程为消费者
• 环境准备                ：RabbitMQ 安装还请自行百度

1.创建RabbitMQHelper工程文件

 

2.安装依赖

   通过程序包管理器控制台或者nuget安装RabbitMQ.Client,默认项目选择RabbitMQHelper

install-pack RabbitMQ.client

　　

2.1 添加帮助类所需实体

 添加交换机实体

using System;
using System.Collections.Generic;
using System.Text;

namespace RabbitMQHelper.Model
{
    /// <summary>
    /// 交换机实体
    /// </summary>
    public class ExchangeModel
    {
        /// <summary>
        /// 交换机名称
        /// </summary>
        public string ExchangeName { get; set; }
        /// <summary>
        /// 交换机类型
        /// </summary>
        public string ExchangeType { get; set; }
        /// <summary>
        /// 路由key
        /// </summary>
        public string RouteKey { get; set; }
        /// <summary>
        /// 是否持久化
        /// </summary>
        public bool Durable { get; set; }
    }
}

添加链接实体

using System;
using System.Collections.Generic;
using System.Text;

namespace RabbitMQHelper.Model
{
    /// <summary>
    /// 连接实体
    /// </summary>
    public class HostModel
    {
        /// <summary>
        /// 客户端账号
        /// </summary>
        public string UserName { get; set; }
        /// <summary>
        /// 客户端密码
        /// </summary>
        public string PassWord { get; set; }

        /// <summary>
        /// 连接地址
        /// </summary>
        public string Host { get; set; }
        /// <summary>
        /// 端口号
        /// </summary>

        public int Port { get; set; }
        public ExChangeModel ExChangeModel { get; set; }
        /// <summary>
        /// 虚拟路径
        /// </summary>
        public string VirtualHost { get; set; }
    }
    /// <summary>
    /// RabbitMq实体
    /// </summary>
    public class ExChangeModel
    {
        /// <summary>
        /// 队列名称
        /// </summary>
        public string QueueName { get; set; }
        /// <summary>
        /// 路由名称
        /// </summary>
        public string RouteKey { get; set; }
        /// <summary>
        /// 交换机名称
        /// </summary>

        public string ExChangeName { get; set; }
    }
}

2.3 添加持久化链接单例

  此处主要是考虑到如果每一次请求都去创建一个连接的话，比较耗时，而且rabbitMQ官方也建议使用长连接的方式进行通信，所以此处用一个单例进行存储连接信息

using RabbitMQ.Client;
using System;
using System.Collections.Generic;
using System.Text;

namespace RabbitMQHelper.core
{
    public class RabbitMQSingleton
    {
        private static RabbitMQSingleton rabbitServicesSingleton;
        static RabbitMQSingleton()
        {
            rabbitServicesSingleton = new RabbitMQSingleton();
        }
        private Dictionary<string, IConnection> RabbitMQconn = new Dictionary<string, IConnection>();
        public static RabbitMQSingleton GetInstance()
        {
            return rabbitServicesSingleton;
        }
        /// <summary>
        /// 添加MQ连接
        /// </summary>
        /// <param name="key">连接名</param>
        /// <param name="value">内容</param>
        public void SetRabbitMqConn(string key, IConnection value)
        {
            if (!RabbitMQconn.ContainsKey(key))
            {
                RabbitMQconn.Add(key, value);
            }
        }
        /// <summary>
        /// 获取rabbitmq所有连接信息
        /// </summary>
        /// <returns></returns>
        public Dictionary<string, IConnection> GetRabbitMQconn()
        {
            return RabbitMQconn;
        }
        /// <summary>
        /// 移除连接
        /// </summary>
        /// <param name="key"></param>
        /// <returns></returns>
        public bool RemoveMQconn(string key)
        {
            bool sflag = true;
            try
            {
                if (RabbitMQconn.ContainsKey(key))
                {
                    RabbitMQconn.Remove(key);
                }
            }
            catch (Exception)
            {
                sflag = false;
                throw;
            }
            return sflag;
        }
    }
}

2.4.添加辅助类

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQHelper.Model;
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;

namespace RabbitMQHelper
{
    public class RabbitHelper
    {
        private readonly ConnectionFactory factory = null;
        public RabbitHelper(HostModel hostModel)
        {
            // 创建连接工厂
            factory = new ConnectionFactory
            {
                UserName = hostModel.UserName,//连接用户名
                Password = hostModel.PassWord,//连接密码
                HostName = "localhost",//连接地址
                Port = hostModel.Port,//端口号
                //VirtualHost = hostModel.VirtualHost
            };
        }
        /// <summary>
        /// 创建连接
        /// </summary>
        /// <returns></returns>
        public IConnection Connection()
        {

            IConnection connection = null;
            try
            {
                factory.AutomaticRecoveryEnabled = true;//自动重连
                connection = factory.CreateConnection();
            }
            catch (Exception)
            {
                throw new Exception("连接失败!~~~~~");
            }
            return connection;
        }
      /// <summary>
      /// 
      /// </summary>
      /// <param name="conn"></param>
      /// <param name="queueName"></param>
      /// <param name="msg"></param>
      /// <param name="exchangeModel"></param>
      /// <returns></returns>
        public bool sendMsg(IConnection conn, string queueName, string msg, ExchangeModel exchangeModel)
        {
            bool sflag = true;
            try
            {
                //var channel = conn.CreateModel();
                using (var channel = conn.CreateModel())
                {

                    //1交换机，交换机类型
                    channel.ExchangeDeclare(exchangeModel.ExchangeName, exchangeModel.ExchangeType);
                    //队列名称，是否持久化，独占的队列，不使用时是否自动删除，
                    channel.QueueDeclare(queueName, exchangeModel.Durable,false,false,null);
                    //转换成byte数组
                    var sendBytes = Encoding.UTF8.GetBytes(msg);
                    //设置持久化参数
                    var properties = channel.CreateBasicProperties();
                    properties.DeliveryMode = 2;//1表示不持久，2表示持久化
                    if (!exchangeModel.Durable)
                    {
                        properties = null;
                    }
                    //发送消息：交换机名称，路由，持久化参数，消息内容
                    channel.BasicPublish(exchangeModel.ExchangeName, exchangeModel.RouteKey, properties, sendBytes);
                }
            }
            catch (Exception)
            {

                sflag = true;
                throw;
            }
            return sflag;
        }
        /// <summary>
        /// 接收消息
        /// </summary>
        /// <param name="connection"></param>
        /// <param name="queueName"></param>
        /// <param name="durable"></param>
        /// <returns></returns>
        public string ConsumMsg(IConnection connection,string queueName, ExchangeModel exchangeModel)
        {
            string msg = string.Empty;
            var channel = connection.CreateModel();
         
                //队列绑定：队列名称，交换机名称，路由
                channel.QueueBind(queueName, exchangeModel.ExchangeName, exchangeModel.RouteKey, null);
                var consumer = new EventingBasicConsumer(channel);       
                //接收到消息事件
                consumer.Received += (ch, ea) =>
                {
                    var message = Encoding.UTF8.GetString(ea.Body);
                    msg = message;
                    Console.WriteLine($"收到消息： {message}");
                    //确认该消息已被消费
                    channel.BasicAck(ea.DeliveryTag, false);
                };
                //启动消费者 设置为手动应答消息
                channel.BasicConsume(queueName, false, consumer);
                return msg;

        }

    }
}

2.5添加生产者类

using RabbitMQ.Client;
using RabbitMQHelper.core;
using RabbitMQHelper.IServer;
using RabbitMQHelper.Model;
using System;
using System.Collections.Generic;
using System.Text;

namespace RabbitMQHelper.Server
{
    public class SendService : ISendService
    {
        IConnection connection;//rabbitmq连接地址
        private object obj=new object();//对象
        RabbitHelper rabbitHelper;//rabbitHelper类

        public SendService()
        {
            RabbitMQSingleton rabbitMQSingleton = RabbitMQSingleton.GetInstance();
            Dictionary<string, IConnection> connDictionary = rabbitMQSingleton.GetRabbitMQconn();
            if (connDictionary != null && connDictionary.Count > 0)
            {
                connection = connDictionary["test"];
            }
            else
            {
                HostModel hostModel = new HostModel();
                hostModel.UserName = "admin";
                hostModel.PassWord = "admin";
                hostModel.Host = "127.0.0.1";
                hostModel.Port = 5672;
                //hostModel.VirtualHost = "/";
              
                lock (obj)
                {
                    rabbitHelper = new RabbitHelper(hostModel);
                    connection = rabbitHelper.Connection();
                    rabbitMQSingleton.SetRabbitMqConn("test", connection);
                }
            }
        }
        public bool CloseConnection()
        {
            throw new NotImplementedException();
        }
        public void SendMsg(string queueName, string msg)
        {
            ExchangeModel exchangeModel = new ExchangeModel();
            exchangeModel.Durable = false;
            exchangeModel.ExchangeName = "ClentName";
            exchangeModel.ExchangeType = ExchangeType.Direct;
            exchangeModel.RouteKey = "ClentRoute";
            rabbitHelper.sendMsg(connection, queueName, msg, exchangeModel);
        }

        //public void SendMsg(string msg)
        //{

        //    throw new NotImplementedException();
        //}
    }
}

2.6添加消费者

using RabbitMQ.Client;
using RabbitMQHelper.core;
using RabbitMQHelper.Model;
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;

namespace RabbitMQHelper.Server
{
    public class ConsumerServer
    {
        IConnection connection;//rabbitmq连接地址
        private object obj = new object();//对象
        RabbitHelper rabbitHelper;//rabbitHelper类

        public ConsumerServer()
        {
            RabbitMQSingleton rabbitMQSingleton = RabbitMQSingleton.GetInstance();
            Dictionary<string, IConnection> connDictionary = rabbitMQSingleton.GetRabbitMQconn();
            if (connDictionary != null && connDictionary.Count > 0)
            {
                connection = connDictionary["test"];
            }
            else
            {
                HostModel hostModel = new HostModel();
                hostModel.UserName = "admin";
                hostModel.PassWord = "admin";
                hostModel.Host = "127.0.0.1";
                hostModel.Port = 5672;
                //hostModel.VirtualHost = "/";

                lock (obj)
                {
                    rabbitHelper = new RabbitHelper(hostModel);
                    connection = rabbitHelper.Connection();
                    rabbitMQSingleton.SetRabbitMqConn("test", connection);
                }
            }
        }
        public string GetMsg(string queueName)
        {
            ExchangeModel exchangeModel = new ExchangeModel();
       
            exchangeModel.ExchangeName = "ClentName";
            exchangeModel.ExchangeType = ExchangeType.Direct;
            exchangeModel.RouteKey = "ClentRoute";
            return  rabbitHelper.ConsumMsg(connection, queueName, exchangeModel);
        }
    }
}

3.添加RibbitMQClient项目（消息生产者）

创建控制台程序

using RabbitMQ.Client;
using RabbitMQHelper.IServer;
using RabbitMQHelper.Server;
using System;

namespace RibbitMQClient
{
    class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("消息生产者开始生产数据！");
            Console.WriteLine("输入exit退出!");


            ISendService sendService = new SendService();

            string input;

            do
            {

                input = Console.ReadLine();

                sendService.SendMsg("Clent1", input);
            } while (input.Trim().ToLower() != "exit");


        }
    }
}

4.创建RibbitMQServer 消费者模式

using RabbitMQHelper.Server;
using System;
using System.Text;

namespace RibbitMQServer
{
    class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("Hello World!");
            ConsumerServer consumerServer = new ConsumerServer();
            consumerServer.GetMsg("Clent1");
            Console.WriteLine("消费者已启动");
                Console.ReadKey();
        

        }
    }
}

5.运行程序

  可正常接收数据

标注:如有理解不对的还请多指教，或者有好的实现方式也可一起讨论。