RabbitMQ笔记

2022-07-04 01:34:05 阅读：171 来源： 互联网

标签：队列 Channel 笔记交换机线程 RabbitMQ channel 消息

RabbitMQ channel 频道，exchange 交换机和 queue队列

connection是指物理的连接，一个client与一个server之间有一个连接；

一个连接上可以建立多个channel，可以理解为逻辑上的连接。

一般应用的情况下，有一个channel就够用了，不需要创建更多的channel？？
（ Channel 起了什么作用，它实质上是屏蔽了 Connection 的细节，让开发者不用去管 TCP 层面上的事儿，同时基于 NIO 可以使得 Connection 的 TCP 能够被复用，减少了 TCP 连接建立的开销。 1 channel per thread 因为 Channel 本身不是线程安全的, 1 consumer per channel.是因为如果一个 Consumer 在一个 Channel 中正在监听某一个 Queue的消息，那么这个 Consumer 是不能在这个 Channel 中同时去处理另一个 Queue 的，出于消费速度的考虑所以需要开辟多个 Channel，如果你本身没那么大消息吞吐量，也可以共用一个 Channel；而 Producer 是不存在这个问题的。）

为了将不同类型的消息进行区分，设置了交换机与路由两个概念。
比如，将A类型的消息发送到名为‘C1’的交换机，将类型为B的发送到'C2'的交换机。当客户端连接C1处理队列消息时，取到的就只是A类型消息。

发送消息时，只要有“交换机”就够了。
至于交换机后面有没有对应的处理队列，发送方是不用管的。routingkey可以是空的字符串（表示发送到所有的队列中）。在示例中，我使用了两个key交替发送消息，是为了下面更便于理解routingkey的作用。

交换机设置参数中有两个重要的概念：

A，类型。有三种类型： Fanout类型最简单，这种模型忽略routingkey；Direct类型是使用最多的，使用确定的routingkey。这种模型下，接收消息时绑定'key_1'则只接收key_1的消息；最后一种是Topic，这种模式与Direct类似，但是支持通配符进行匹配，比如： 'key_*'，就会接受key_1和key_2。Topic貌似美好，但是有可能导致不严谨，所以还是推荐使用Direct。

B，持久化。指定了持久化的交换机，在重新启动时才能重建，否则需要客户端重新声明生成才行。
需要特别明确的概念：交换机的持久化，并不等于消息的持久化。只有在持久化队列中的消息，才能持久化；如果没有队列，消息是没有地方存储的；消息本身

queue：队列
队列仅是针对接收方（consumer）的，由接收方根据需求创建的。只有队列创建了，交换机才会将新接受到的消息送到队列中，交换机是不会在队列创建之前的消息放进来的。即在建立队列之前，发出的所有消息都被丢弃了。

net客户端代码,发送接收

class RabbitMQClient
    {
        public static IConnection GetConnection(string hostName, string userName, string password)
        {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.HostName = hostName;
            connectionFactory.UserName = userName;
            connectionFactory.Password = password;
            connectionFactory.Port = 5672;
            return connectionFactory.CreateConnection();
        }

        public static void Send(string queue, string data)
        {
            using (IConnection connection = GetConnection("localhost", "guest", "guest"))
            {
                // //这里Rabbit的玩法就是一个通道channel下包含多个队列Queue
                using (IModel channel = connection.CreateModel())
                {
                    channel.QueueDeclare(queue, false, false, false, null);
                    channel.BasicPublish(string.Empty, queue, null, Encoding.UTF8.GetBytes(data));
                }
            }
        }
        public static void Receive(string queue)
        {
            using (IConnection connection = GetConnection("localhost", "guest", "guest"))
            {
                using (IModel channel = connection.CreateModel())
                {
                    channel.QueueDeclare(queue, false, false, false, null);
                    var consumer = new EventingBasicConsumer(channel);

                    BasicGetResult result = channel.BasicGet(queue, true);
                    if (result != null)
                    {
                        string data =
                        Encoding.UTF8.GetString(result.Body.ToArray());
                        Console.WriteLine(data);
                    }
                }
            }
        }
    }

假如发送/接送方单线程,每次都连接,收发,断开, 每个动作30ms, 在MQ的管理界面看到收发的Message Rate最大是36 /s, 跟这个时间对得上

假如开多个线程,同时打开10个以上连接,会出现下面的异常, RabbitMQ.Client.Exceptions.BrokerUnreachableException:“None of the specified endpoints were reachable”

所以应该采用单例模式,生产者,消费者都只打开一个连接. 这样改造之后Message Rate 可以达到300/s

把线程数加到40, using (IModel channel = _conn.CreateModel()) //引发的异常:“System.TimeoutException”

再改成每个线程用单独一个Channel来发,Message Rate 可以达到850/s

 RabbitMQClient mQClient = new RabbitMQClient("localhost");
           

            //1个线程Msg Rate = 125/s, 10个线程300 /s,20个线程400/s,40个线程850/s
            for (int k = 0; k < 40; k++)
            {
                Task.Run(() =>
                {
                    IModel Channel = mQClient.GetChannel(); //每个线程建立一个通道
                    for (int i = 0; i < 1000; i++)
                    {
                        mQClient.Send("IDG", "Hello World!" + i,Channel);
                        mQClient.Receive("IDG", Channel);

                    }
                    Channel.Close();

                });
            }

            
            Console.ReadLine();

开源软件成熟度评测报告-分布式消息中间件_对比Kafka和RabbitMQ

https://www.confluent.io/blog/kafka-fastest-messaging-system/#throughput-results 这篇文章说kafka是RabbitMQ的吞吐量的15倍, 但在吞吐量30MB/s 以下, RabbitMQ的反应时间是最好的,只有1ms, 假如每条msg是1K,就是每秒3万条消息

在我的Mac Air测试, 启用持久化队列和持久化消息, 不开启publish Confirm, 单线程发1000条信息发完1000条,再接收完用时3.74秒;

Task GetChannel =0.21s

改成2个线程对应2个Channel,每个通道发500条,用时2.52秒; (我的CPU是双核的,2个线程是最快的,多了线程反而多了切换的时间)

改成3个线程对应3个Channel,每个通道发333条,用时3.3秒;

改成5个线程对应5个Channel,每个通道发200条,用时5.0秒; 其中

task GetChannel =1.87s (5线程建立连接最耗时)
task QueueDeclare =0.02s
task BasicGet =10ms/次
task BasicPublish=1ms/次

关闭掉持久化消息,速度反而变慢了.订阅消息的速度是发送消息的1/10到1/20, 慢太多了.

在消息消费模型方面，Kafka使用Pull模型，消费者根据需要的消息，从服务端Pull数据；RabbitMQ采用Push模型，消费者与服务端队列保持长连接，队列中有消息则会Push至消费者。

在消息确认方面，Kafka提供消息至少发送一次(“at least once”），确保消息被可靠消费; RabbitMQ提供消息确认机制（Ack），直到消费者显式返回Ack才移除消息，确保消息到达消费者。此外，RabbitMQ还提供消息过滤，预读取计数等功能。

消息中间件的性能效率主要指集群接收和发送消息的吞吐率。吞吐率指标有两个，一是生产者/消费者每秒发送/接收的消息条数，单位为条/秒（Record/s）；二是生产者/消费者每秒发送/接收的消息量，单位为MB/s。我们通过消息的发送吞吐率和接收吞吐率两方面评估消息队列中间件的性能，主要测试场景如下：

（1）客户端并发数：不断增加生产者/消费者数量，测试发送/接收吞吐率的变化情况；
（2）消息大小：不断增加消息大小，测试发送吞吐率的变化情况；
（3）系统可靠性：通过增加副本数、改变应答模式，测试发送吞吐率的变化情况；

Direct Exchange

见文知意，直连交换机意思是此交换机需要绑定一个队列，要求该消息与一个特定的路由键完全匹配。简单点说就是一对一的，点对点的发送。

Fanout exchange

这种类型的交换机需要将队列绑定到交换机上。一个发送到交换机的消息都会被转发到与该交换机绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。简单点说就是发布订阅。发送到多个Queue

标签：队列,Channel,笔记,交换机,线程,RabbitMQ,channel,消息
来源： https://www.cnblogs.com/zitjubiz/p/RabbitMq_Compare_Kafka_Tuning.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

RabbitMQ笔记

Direct Exchange

Fanout exchange