一RabbitMQ 概念详解和应用

Posted 2021-07-03 懒懒的佩奇

消息队列和同步请求的区别 

无论RabbitMQ还是Kafka，本质上都是提供了基于message或事件驱动异步处理业务的能力，相比于http和rpc的直接调用，它有着不可替代的优势：

1. 解耦，解耦的一个最常见做法就是在服务之间新增一层，使原来直接依赖的A,B service 松耦合，这在微服务架构中尤为重要。

2. 流量控制：通过消息队列意味着我们可以监控时间段内的需要处理的业务量，对于明显超出服务承受能力的请求，我们可以延迟处理或者拒绝处理，保证服务的稳定性与可用性。

3. 负载均衡：可以通过消息代理实现下游服务的负载均衡，有利于保证服务的高可用性；

 

消息队列协议介绍 AMQP

协议： AMQP，即Advanced Message Queuing Protocol，一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。

RabbitMQ模型：

1、Server：又称 broker，接受客户端连接，实现 AMQP 实体服务。
2、Connection：连接和具体 broker 网络连接。
3、Channel：网络信道，几乎所有操作都在 channel 中进行，channel 是消息读写的通道。客户端可以建立多个 channel，每个 channel 表示一个会话任务。
4、message：消息，服务器和应用程序之间传递的数据，由 properties 和 body 组成。properties 可以对消息进行修饰，比如消息的优先级，延迟等高级特性；body 是消息实体内容。
5、Virtual host：虚拟主机，用于逻辑隔离，最上层消息的路由。一个 Virtual host 可以若干个 Exchange 和 Queue，同一个 Virtual host 不能有同名的 Exchange 或 Queue。
6、Exchange：交换机，接受消息，根据路由键转发消息到绑定的队列上。
7、Binding：Exchange 和 Queue 之间的虚拟连接，binding 中可以包括 routing key。
8、Routing key：一个路由规则，虚拟机根据他来确定如何路由 一条消息。
9、Queue：消息队列，用来存放消息的队列。

 

Exchange 各种类型

交换机的类型，direct、topic、fanout、headers；

1、Direct Exchange，所有发送到 Direct Exchange 的消息被转发到 RouteKey  中指定的 Queue, Direct Exchange 可以使用默认的默认的 Exchange （default Exchange），默认的 Exchange 会绑定所有的队列，所以 Direct 可以直接使用 Queue 名（作为routing key ）绑定。或者消费者和生产者的 routing key 完全匹配。

2、Toptic Exchange,是指发送到 Topic Exchange 的消息被转发到所有关心的 Routing key 中指定 topic 的 Queue 上。Exchange 将 routing key 和某 Topic 进行模糊匹配，此时队列需要绑定一个 Topic。所谓模糊匹配就是可以使用通配符，“#”可以匹配一个或多个词，“*”只匹配一个词。比如“log.#”可以匹配“log.info.test”， "log.*"就只能匹配 log.error。

3、Fanout Exchange:不处理路由键，只需简单的将队列绑定到交换机上。发送到该交换机上的消息都会被发送到与该交换机绑定的队列上。Fanout 转发是最快的。

 总而言之：Direct change是严格意义上的匹配，routing key 与 binding key 完全一样； Fanout Exchange 是完全不关心你的routing key,向所有绑定的queue 全部发送；这是两个极端，那么Toptic Exchange 就相对Direct change宽松一些，它的消息投递取决于模糊匹配的结果。

 最后介绍headers模式：

4. Headers Exchange:

headers类型的Exchange不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。
在绑定Queue与Exchange时指定一组键值对；当消息发送到Exchange时，RabbitMQ会取到该消息的headers（也是一个键值对的形式），对比其中的键值对是否完全匹配Queue与Exchange绑定时指定的键值对；如果完全匹配则消息会路由到该Queue，否则不会路由到该Queue。

通过图来表示：

 

RabbitMQ 单机安装与使用

windows 系统下，安装教程传送门（https://www.cnblogs.com/saryli/p/9729591.html）

linux 系统下，安装教程传送门（https://www.linuxprobe.com/install-rabbitmq-on-centos-7.html）

（会在后面的系列介绍高可用集群的搭建）

这里以dotnet 为例，示例如何使用 rabbitmq。

 Producer End:

    public class MessageQueueService
    {
        private IConnection _connection;
        private IModel _channel;
        private const string TicketExchangeName = "tickets";
        public MessageQueueService()
        {
            var factory = new ConnectionFactory() { HostName = "localhost" };
            _connection = factory.CreateConnection();
            _channel = GetChannel();
            _channel.ExchangeDeclare(TicketExchangeName, ExchangeType.Fanout, false, false);
        }

        public void SendTicketMessage(Ticket message, string optType)
        {
            
            var messageBytes = ObjectToByteArray(message);
            _channel.BasicPublish(TicketExchangeName, optType, false, null, messageBytes);
        }
     }

 

Consumer End:

    public class TicketConsumerService : IHostedService
    {
        private IConnection _connection;
        private IModel _channel;
        private const string TicketExchangeName = "tickets";
        private string queueName = "";
        public TicketConsumerService()
        {
            var factory = new ConnectionFactory() { HostName = "localhost" };
            _connection = factory.CreateConnection();
            _channel = _connection.CreateModel();
            _channel.ExchangeDeclare(TicketExchangeName, ExchangeType.Fanout, false, false);
        }


        Task IHostedService.StartAsync(CancellationToken cancellationToken)
        {
            queueName =  _channel.QueueDeclare().QueueName;
            _channel.QueueBind(queueName, TicketExchangeName, "buy");
            var consumer = new EventingBasicConsumer(_channel);
            consumer.Received += (model, ea) =>
            {
                Ticket message = (Ticket)ByteArrayToObject(ea.Body.ToArray());
                var routingKey = ea.RoutingKey;
                Console.WriteLine(" [x] Received \'{0}\':\'{1}\'",
                    routingKey, message.Boarding);
            };
            _channel.BasicConsume(queue: queueName, autoAck: true, consumer: consumer);
            return Task.CompletedTask;
        }

        Task IHostedService.StopAsync(CancellationToken cancellationToken)
        {
            _connection.Dispose();
            _channel.Dispose();
            
            return Task.CompletedTask;
        }

        private Object ByteArrayToObject(byte[] arrBytes)
        {
            MemoryStream memStream = new MemoryStream();
            BinaryFormatter binForm = new BinaryFormatter();
            memStream.Write(arrBytes, 0, arrBytes.Length);
            memStream.Seek(0, SeekOrigin.Begin);
            Object obj = (Object)binForm.Deserialize(memStream);

            return obj;
        }
    }

这篇主要介绍了一些Message Queue的概念及如何在web 开发过程中使用。下一篇会介绍RabbitMQ的一些进阶特性和使用场景。

--------------------------------------------

欢迎大家留言讨论，指出错误和不当之处！