rabbitMQ 五种工作模式

rabbitMQ 五种工作模式

目录

1.rabbitmq是什么？

2.什么是消息队列？

3.消息队列应用的场景

4.rabbitmq的工作原理和组成部分 

5. 消息模型

5.1 正常消息模型

5.2工作消息模型

5.3 Publish/subscribe（交换机类型：Fanout，也称为广播 ）

5.4 Routing 路由模型（交换机类型：direct）

5.5 Topics 通配符模式（交换机类型：topics）

---

1.rabbitmq是什么？

就是用Erlang语言开发的基于AMQP（高级消息队列协议）一种消息队列，中文学习地址：http://rabbitmq.mr-ping.com/RabbitMQ官网下载地址：Downloading and Installing RabbitMQ — RabbitMQ

2.什么是消息队列？

MQ全称为Message Queue，即消息队列。“消息队列”是在消息的传输过程中保存消息的容器。它是典型的：生产者、消费者模型。生产者不断向消息队列中生产消息，消费者不断的从队列中获取消息。这样生产者和消费者都不用知道对方的存在。因为消息的生产和消费都是异步的，而且只关心消息的发送和接收，没有业务逻辑的侵入，这样就实现了生产者和消费者的解耦。

3.消息队列应用的场景

1.异步和高并发

在高并发的场景下，服务很难同时执行大量数据，这时就需要异步，来防止阻塞。比如当订单非常多的时候，通过消息队列就可以缓解系统压力

2.解耦

因为MQ相当于一个中介，生产者和消费者不用知道对方的存在，从而实现解耦。

3.削峰填谷

4.rabbitmq的工作原理和组成部分 

下图是RabbitMQ的基本结构：

组成部分说明：

Broker：消息队列服务进程，此进程包括两个部分：Exchange和Queue
Exchange：消息队列交换机，按一定的规则将消息路由转发到某个队列，对消息进行过虑。
Queue：消息队列，存储消息的队列，消息到达队列并转发给指定的
Producer：消息生产者，即生产方客户端，生产方客户端将消息发送
Consumer：消息消费者，即消费方客户端，接收MQ转发的消息。
生产者发送消息流程：

1、生产者和Broker建立TCP连接。

2、生产者和Broker建立通道。

3、生产者通过通道消息发送给Broker，由Exchange将消息进行转发。

4、Exchange将消息转发到指定的Queue（队列）

消费者接收消息流程：

1、消费者和Broker建立TCP连接

2、消费者和Broker建立通道

3、消费者监听指定的Queue（队列）

4、当有消息到达Queue时Broker默认将消息推送给消费者。

5、消费者接收到消息。

6、ack回复

5. 消息模型 5.1 正常消息模型

        （此时使用默认交换机，这时routingkey就等于队列名称）

  连接rabbitmq工具类： 

        /// 

        /// 获取RabbitMQ连接
        /// 
        /// 
        public static IConnection GetConnection()
        {
            var factory = new ConnectionFactory
            {
                HostName = "192.168.3.22", //ip
                Port = 5672, // 端口
                UserName = "zkz", // 账户 注意不要用guest账号
                Password = "zkz", // 密码 
                VirtualHost = "/"   // 虚拟主机 一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mq
            };

            return factory.CreateConnection();
        }

生产者发送消息：

        public static void SendMessage()
        {
            //队列名称
            string queueName = "normal";
            // 1.获取到连接
            using (var connection = RabbitMQHelper.GetConnection())
            {
                //2. 创建信道 队列的声明，消息的发送等都在信道类中
                using (var channel = connection.CreateModel())
                {
                    // 3.创建队列 
                    //参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments
                    
                    channel.QueueDeclare(queueName, false, false, false, null);
                    while (true)
                    {

                        string message = "Hello RabbitMQ Message";
                        var body = Encoding.UTF8.GetBytes(message);
                        // 4.发送消息到rabbitmq,使用rabbitmq中默认提供交换机路由,默认的路由Key和队列名称完全一致

                        //参数：String exchange, String routingKey, BasicProperties props, byte[] body
                        
                        channel.BasicPublish(exchange: "", routingKey: queueName, null, body);
                        Thread.Sleep(1000);
                        Console.WriteLine("send normal message");
                    }
                }
            }
        }

消费者接收消息

        public static void ReceiveMessage()
        {
            // 消费者消费是队列中消息
            string queueName = "normal";
            //1.创建连接
            var connection = RabbitMQHelper.GetConnection();
            {
                //2.创建信道
                var channel = connection.CreateModel();
                {
                    //参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments
                    
                     //注意：参数要与生成者设定一样，如果生产这第二个参数设置为true,而消费者在声明队列时设定为false,这样就消费不到消息了
                    channel.QueueDeclare(queueName, false, false, false, null);
                    //通过事件委托的方法来获取消息
                    var consumer = new EventingBasicConsumer(channel);
                    consumer.Received += (model, ea) =>
                    {
                        var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                        Console.WriteLine(" Normal Received => {0}", message);
                    };
                    // 监听队列，第二个参数：是否自动进行消息确认。
                    //参数：String queue, boolean autoAck, Consumer callback
                    
                    channel.BasicConsume(queueName, true, consumer);
                }

            }

        }

 如何避免当RabbitMQ退出或奔溃时避免队列和队列中的消息丢失

//第二个参数durable设置成true 这样是持久化，mq重启后队列还在
channel.QueueDeclare(queueName, true, false, false, null);


//同时想在队列中消息在mq推出或重启后还在
        public bool SendMQMessage(string queueName, string msg)
        {
            try
            {

                //创建属性
                IBasicProperties basicProperties = channel.CreateBasicProperties();
                //设置消息为持久性 避免当RabbitMQ退出或奔溃时
                //消息丢失 配合 QueueDeclare 第二个参数durable:true使用 
                basicProperties.Persistent = true;
                var payload = Encoding.UTF8.GetBytes(msg);
                channel.BasicPublish("", queueName, basicProperties, payload);
                return true;
            }
            catch (Exception ex)
            {
                return false;
            }
        }

消息确认机制（ACK）
通过刚才的案例可以看出，消息一旦被消费者接收，队列中的消息就会被删除。

那么问题来了：RabbitMQ怎么知道消息被接收了呢？

如果消费者领取消息后，还没执行 *** 作就挂掉了呢？或者抛出了异常？消息消费失败，但是RabbitMQ无从得知，这样消息就丢失了！

因此，RabbitMQ有一个ACK机制。当消费者获取消息后，会向RabbitMQ发送回执ACK，告知消息已经被接收。不过这种回执ACK分两种情况：

自动ACK：消息一旦被接收，消费者自动发送ACK

手动ACK：消息接收后，不会发送ACK，需要手动调用

大家觉得哪种更好呢？

这需要看消息的重要性：

如果消息不太重要，丢失也没有影响，那么自动ACK会比较方便

如果消息非常重要，不容丢失。那么最好在消费完成后手动ACK，否则接收消息后就自动ACK，RabbitMQ就会把消息从队列中删除。如果此时消费者宕机，那么消息就丢失了。

这时消费者方法：

            // 消费者消费是队列中消息
            string queueName = "normal";
            //1.创建连接
            var connection = RabbitMQHelper.GetConnection();
            {
                //2.创建信道
                var channel = connection.CreateModel();
                {
                    //参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments
                    
                     //注意：参数要与生成者设定一样，如果生产这第二个参数设置为true,而消费者在声明队列时设定为false,这样就消费不到消息了
                    channel.QueueDeclare(queueName, false, false, false, null);
                    //通过事件委托的方法来获取消息
                    var consumer = new EventingBasicConsumer(channel);
                    consumer.Received += (model, ea) =>
                    {
                        var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                        Console.WriteLine(" Normal Received => {0}", message);
                        channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);//手动发送ACK应答

                    };
                    // 监听队列，第二个参数：是否自动进行消息确认。
                    //参数：String queue, boolean autoAck, Consumer callback
                    
                    channel.BasicConsume(queueName, false, consumer);
                }

            }

        }

//把第二个参数 autoAck改成false             
channel.BasicConsume(queueName, false, consumer);


//接收消息的委托事件中加手动确认
channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);

特别提醒：当你需要使用多个队列，比如你需要向多个队列中发送消息，还有监听多个队列中的消息，你只要创建一个信道就可以： 

        private void btnConnect_Click(object sender, EventArgs e)
        {
            try
            {
                //1.1.实例化连接工厂
                factory = new ConnectionFactory()
                {
                    HostName = "127.0.0.1",
                    UserName = tbUserName.Text,
                    Password = tbPassword.Text
                };
                //2. 建立连接
                connection = factory.CreateConnection();
                //3. 创建信道
                channel = connection.CreateModel();
                consumer = new EventingBasicConsumer(channel);
                consumer.Received += Receive_Excute;


                //声明多个队列
                channel.QueueDeclare(MQQueueParameter.ReceiveTaskQueue, true, false, false, null);
                channel.QueueDeclare(MQQueueParameter.ReceiveLicense, true, false, false, null);
                channel.QueueDeclare(MQQueueParameter.Handcmd, true, false, false, null);
                channel.QueueDeclare(MQQueueParameter.Taskmanage, true, false, false, null);


                //监听多个队列
                channel.BasicConsume(queue: MQQueueParameter.ReceiveTaskQueue, //接收来自WMS的工单信息,xml字符串格式
                    noAck: true, //和tcp协议的ack一样，为false则服务端必须在收到客户端的回执（ack）后才能删除本条消息
                    consumer: consumer);

                channel.BasicConsume(queue: MQQueueParameter.ReceiveLicense, //接收来自WMS的工单信息,xml字符串格式
                    noAck: true, //和tcp协议的ack一样，为false则服务端必须在收到客户端的回执（ack）后才能删除本条消息
                    consumer: consumer);
                

            }
            catch (Exception ex)
            {
                //MessageBox.Show("连接异常{ex.Message}");
                MessageBox.Show(ex.Message.ToString());
            }
        }
        /// 

        /// 监听队列返回的消息
        /// 
        /// 
        /// 
        private void Receive_Excute(object sender, BasicDeliverEventArgs e)
        {
            var body = e.Body;
            string message = Encoding.UTF8.GetString(body);
            JObject jo = JObject.Parse(message);

            this.Invoke(new Action(() =>
            {
                switch (e.RoutingKey)
                {
                    case MQQueueParameter.ReceiveTaskQueue: //
                        taskNo = jo["TASKID"].ToString(); //获取TASKID的值
                        SRCPOS = jo["SRCPOS"].ToString();
                        DESTPOS = jo["DESTPOS"].ToString();
                        this.rTBReceive.Text = "接收到新任务：" + message;
                        gettaskporcess();
                        break;
                    case MQQueueParameter.ReceiveLicense:
                        string PICKLICENSE = jo["PICKLICENSE"].ToString(); //获取TASKID的值
                        if ("1".Equals(PICKLICENSE))
                        {
                            request();
                        }
                        if ("3".Equals(PICKLICENSE))
                        {
                            requestClose();
                        }
                        this.rTBReceive.Text = "接收到取放货许可信息：" + message;
                        break;
                    case MQQueueParameter.Handcmd:  //

                        this.rTBReceive.Text = "接收到人工指令：" + message;
                        break;
                    case MQQueueParameter.Taskmanage:

                        this.rTBReceive.Text = "接收异常任务处理反馈：" + message;
                        break;
                    default:
                        break;
                }
            }));
        }

        /// 

        /// 给队列发送消息
        /// 
        /// 
        /// 
        /// 
        public bool SendMQMessage(string queueName, string msg)
        {
            try
            {

                //创建属性
                IBasicProperties basicProperties = channel.CreateBasicProperties();
                //设置消息为持久性 避免当RabbitMQ退出或奔溃时
                //消息丢失 配合 QueueDeclare 第二个参数durable:true使用
                basicProperties.Persistent = true;
                var payload = Encoding.UTF8.GetBytes(msg);
                channel.BasicPublish("", queueName, basicProperties, payload);
                return true;
            }
            catch (Exception ex)
            {
                return false;
            }
        }

5.2工作消息模型

        

work queues与入门程序相比，多了一个消费端，两个消费端共同消费同一个队列中的消息，但是一个消息只能被一个消费者获取。

生产者：

        public static void SendMessage()
        {
            string queueName = "Worker_Queue";

            using (var connection = RabbitMQHelper.GetConnection())
            {
                using(var channel = connection.CreateModel())
                {
                    channel.QueueDeclare(queueName, false, false, false, null);
                    for (int i = 0; i < 30; i++)
                    {
                        string message = $"RabbitMQ Worker {i + 1} Message";
                        var body = Encoding.UTF8.GetBytes(message);
                        channel.BasicPublish("", queueName, null, body);
                        Console.WriteLine("send Task {0} message",i + 1);
                    }
                   
                }
            }
            
        }

 消费者（你可以启动2个消费者的exe程序，这时你会发现rabbitmq是公平分发的模式，当然你也可以根据需要设定多劳多得的模式）：

        public static void ReceiveMessage()
        {
            string queueName = "Worker_Queue";
            var connection = RabbitMQHelper.GetConnection();
            {
                var channel = connection.CreateModel();
                {
                    channel.QueueDeclare(queueName, false, false, false, null);
                    var consumer = new EventingBasicConsumer(channel);
                    //设置prefetchCount : 1来告知RabbitMQ，在未收到消费端的消息确认时，
                    //不再分发消息，也就确保了当消费端处于忙碌状态时，不再分配任务。这就是多劳多得模式。如果是公平分发注释该语句即可
                    channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
                    consumer.Received +=(model, ea) => {
                        var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                        Console.WriteLine(" Worker Queue Received => {0}", message);
                    }; 
                    channel.BasicConsume(queueName,true, consumer);
                }
               
            }
          
        } 

注意：5.1正常模式和5.2消息模式都是使用默认交换机，routingkey就是自己的队列名。下面将结束不使用默认交换机的消息模式，即订阅模式。

说明下：

1、一个生产者多个消费者
2、每个消费者都有一个自己的队列
3、生产者没有将消息直接发送给队列，而是发送给exchange(交换机、转发器)
4、每个队列都需要绑定到交换机上
5、生产者发送的消息，经过交换机到达队列，实现一个消息被多个消费者消费
例子：注册->发邮件、发短信

X（Exchanges）：交换机一方面：接收生产者发送的消息。另一方面：知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何 *** 作，取决于Exchange的类型。

介绍Exchange的3种类型：

Fanout：广播，将消息交给所有绑定到交换机的队列

Direct：定向，把消息交给符合指定routing key 的队列

Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列

Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

5.3 Publish/subscribe（交换机类型：Fanout，也称为广播 ）

Publish/subscribe模型示意图 ：x代表exchange

和前面两种模式不同：

• 1） 声明Exchange

• 2） 发送消息到Exchange，不再发送到Queue

生产者:

        public static void SendMessage()
        {
            using (var connection = RabbitMQHelper.GetConnection())
            {
                using(var channel = connection.CreateModel())
                {
                    // 声明交换机对象
                    channel.ExchangeDeclare("fanout_exchange", "fanout");
                    // 创建队列
                    string queueName1 = "fanout_queue1";
                    channel.QueueDeclare(queueName1, false, false, false, null);
                    string queueName2 = "fanout_queue2";
                    channel.QueueDeclare(queueName2, false, false, false, null);
                    string queueName3 = "fanout_queue3";
                    channel.QueueDeclare(queueName3, false, false, false, null);
                    // 绑定到交互机
                    // fanout_exchange 绑定了 3个队列 
                    channel.QueueBind(queue: queueName1, exchange: "fanout_exchange", routingKey: "");
                    channel.QueueBind(queue: queueName2, exchange: "fanout_exchange", routingKey: "");
                    channel.QueueBind(queue: queueName3, exchange: "fanout_exchange", routingKey: "");

                    for (int i = 0; i < 10; i++)
                    {
                        string message = $"RabbitMQ Fanout {i + 1} Message";
                        var body = Encoding.UTF8.GetBytes(message);
                        channel.BasicPublish("fanout_exchange", "", null, body);
                        Console.WriteLine("Send Fanout {0} message",i + 1);
                    }
                }
            }
            
        } 

消费者：

        public static void ConsumerMessage()
        {
            var connection = RabbitMQHelper.GetConnection();
            {
                var channel = connection.CreateModel();
                {
                    //申明exchange
                    channel.ExchangeDeclare(exchange: "fanout_exchange", type: "fanout");
                    // 创建队列
                    string queueName1 = "fanout_queue1";
                    channel.QueueDeclare(queueName1, false, false, false, null);
                    string queueName2 = "fanout_queue2";
                    channel.QueueDeclare(queueName2, false, false, false, null);
                    string queueName3 = "fanout_queue3";
                    channel.QueueDeclare(queueName3, false, false, false, null);
                    // 绑定到交互机
                    channel.QueueBind(queue: queueName1, exchange: "fanout_exchange", routingKey: "");
                    channel.QueueBind(queue: queueName2, exchange: "fanout_exchange", routingKey: "");
                    channel.QueueBind(queue: queueName3, exchange: "fanout_exchange", routingKey: "");


                    Console.WriteLine("[*] Waitting for fanout logs.");
                    //申明consumer
                    var consumer = new EventingBasicConsumer(channel);
                    //绑定消息接收后的事件委托
                    consumer.Received += (model, ea) => {
                        var body = ea.Body;
                        var message = Encoding.UTF8.GetString(body.ToArray());
                        Console.WriteLine("[x] {0}", message);

                    };

                    channel.BasicConsume(queue: queueName1, autoAck: true, consumer: consumer);
                    Console.WriteLine(" Press [enter] to exit.");
                    Console.ReadLine();
                }
            }
        }

思考：
1、publish/subscribe与work queues有什么区别。

区别：

1）work queues不用定义交换机，而publish/subscribe需要定义交换机。

2）publish/subscribe的生产方是面向交换机发送消息，work queues的生产方是面向队列发送消息(底层使用默认交换机)。

3）publish/subscribe需要设置队列和交换机的绑定，work queues不需要设置，实际上work queues会将队列绑定到默认的交换机 。

相同点：

所以两者实现的发布/订阅的效果是一样的，多个消费端监听同一个队列不会重复消费消息。

2、实际工作用 publish/subscribe还是work queues。

建议使用 publish/subscribe，发布订阅模式比工作队列模式更强大（也可以做到同一队列竞争），并且发布订阅模式可以指定自己专用的交换机。

5.4 Routing 路由模型（交换机类型：direct）

Routing模型示意图：

P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。

X：Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列

C1：消费者，其所在队列指定了需要routing key 为 error 的消息

C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

direct与fanout模式最大的不同是fanout是广播，不需要指定routingkey,而direct需要，direct可以根据routingkey把消息发送到指定的队列中

生产者：

        public static void SendMessage()
        {
            using (var connection = RabbitMQHelper.GetConnection())
            {
                using(var channel = connection.CreateModel())
                {
                    // 声明Direct交换机
                    channel.ExchangeDeclare("direct_exchange", "direct");
                    // 创建队列
                    string queueName1 = "direct_queue1";
                    channel.QueueDeclare(queueName1, false, false, false, null);
                    string queueName2 = "direct_queue2";
                    channel.QueueDeclare(queueName2, false, false, false, null);
                    string queueName3 = "direct_queue3";
                    channel.QueueDeclare(queueName3, false, false, false, null);
                    // 绑定到交互机 指定routingKey  同时一个队列也可以绑定多个routingKey ，消费的时候指定队列和routingKey就可以
                    channel.QueueBind(queue: queueName1, exchange: "direct_exchange", routingKey: "red1");
                    channel.QueueBind(queue: queueName1, exchange: "direct_exchange", routingKey: "red2");
                    channel.QueueBind(queue: queueName1, exchange: "direct_exchange", routingKey: "red3");
                    channel.QueueBind(queue: queueName3, exchange: "direct_exchange", routingKey: "green");

                    for (int i = 0; i < 10; i++)
                    {
                        string message = $"RabbitMQ Direct {i + 1} Message =>green";
                        var body = Encoding.UTF8.GetBytes(message);
                        // 发送消息的时候需要指定routingKey发送
                        channel.BasicPublish(exchange: "direct_exchange", routingKey: "green", null, body);
                        Console.WriteLine("Send Direct {0} message",i + 1);
                    }
                }
            }
            
        } 

消费者：

        public static void ConsumerMessage()
        {
            var connection = RabbitMQHelper.GetConnection();
            var channel = connection.CreateModel();
            channel.ExchangeDeclare(exchange: "direct_exchange", type: "direct");
            var queueName = "direct_queue2";
            channel.QueueDeclare(queueName, false, false, false, null);
            channel.QueueBind(queue: queueName,
                                      exchange: "direct_exchange",
                                      routingKey: "red1");
            channel.QueueBind(queue: queueName,
                                      exchange: "direct_exchange",
                                      routingKey: "red2");
            channel.QueueBind(queue: queueName,
                                      exchange: "direct_exchange",
                                      routingKey: "green");

            Console.WriteLine(" [*] Waiting for messages.");

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body;
                var message = Encoding.UTF8.GetString(body.ToArray());
                var routingKey = ea.RoutingKey;
                Console.WriteLine(" [x] Received '{0}':'{1}'", routingKey, message);

                // 消费完成后需要手动签收消息，如果不写该代码就容易导致重复消费问题
                channel.BasicAck(ea.DeliveryTag, true); // 可以降低每次签收性能损耗
            };

            // 消息签收模式
            // 手动签收 保证正确消费，不会丢消息(基于客户端而已)
            // 自动签收 容易丢消息 
            // 签收：意味着消息从队列中删除
            channel.BasicConsume(queue: queueName,
                                 autoAck: false,
                                 consumer: consumer);

            Console.WriteLine(" Press [enter] to exit.");
            Console.ReadLine();
        }

5.5 Topics 通配符模式（交换机类型：topics）

是一种模糊匹配模式
Topics模型示意图：

 每个消费者监听自己的队列，并且设置带统配符的routingkey,生产者将消息发给broker，由交换机根据routingkey来转发消息到指定的队列。

Routingkey一般都是有一个或者多个单词组成，多个单词之间以“.”分割，例如：inform.sms

通配符规则：

#：匹配一个或多个词

*：匹配不多不少恰好1个词

所谓词就是完整的不带空格没有符号的 比如irs是一个词，但i.rs就不是

举例：

audit.#：能够匹配audit.irs.corporate 或者 audit.irs

audit.*：只能匹配audit.irs

生产者：

        public static void SendMessage()
        {
            using (var connection = RabbitMQHelper.GetConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    channel.ExchangeDeclare("topic_exchange", "topic");
                    // 创建队列
                    string queueName1 = "topic_queue1";
                    channel.QueueDeclare(queueName1, false, false, false, null);
                    string queueName2 = "topic_queue2";
                    channel.QueueDeclare(queueName2, false, false, false, null);
                    string queueName3 = "topic_queue3";
                    channel.QueueDeclare(queueName3, false, false, false, null);
                    // 绑定到交互机  routingKey绑定的可以是模糊的
                    channel.QueueBind(queue: queueName1, exchange: "topic_exchange", routingKey: "user.data.*");
                    channel.QueueBind(queue: queueName2, exchange: "topic_exchange", routingKey: "user.data.delete");
                    channel.QueueBind(queue: queueName3, exchange: "topic_exchange", routingKey: "user.data.update");

                    for (int i = 0; i < 10; i++)
                    {
                        string message = $"RabbitMQ Topic {i + 1} Delete Message";
                        var body = Encoding.UTF8.GetBytes(message);
                        //发送是routingKey必须是具体的 比如BasicPublish的routingKey是user.data.delete，
                        //那么queueName1和queueName2都会收到消息
                        channel.BasicPublish("topic_exchange", "user.data.delete", null, body);
                        Console.WriteLine("Send Topic {0} message", i + 1);
                    }
                }
            }

        }

消费者：

        public static void ConsumerMessage()
        {
            var connection = RabbitMQHelper.GetConnection();
            var channel = connection.CreateModel();
            channel.ExchangeDeclare(exchange: "topic_exchange", type: "topic");
            var queueName = "topic_queue3";
            channel.QueueDeclare(queueName, false, false, false, null);
            // 有个bug 但是注意在topic模式下 不管你队列中指定的routingKey是什么，那么该队列中的消息都会被消费
            //后期不知道rabbitmq会不会修复
            channel.QueueBind(queue: queueName,
                                      exchange: "topic_exchange",
                                      routingKey: "user.data.insert");

            Console.WriteLine(" [*] Waiting for messages.");

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body;
                var message = Encoding.UTF8.GetString(body.ToArray());
                var routingKey = ea.RoutingKey;
                Console.WriteLine(" [x] Received '{0}':'{1}'", routingKey, message);
            };

            channel.BasicConsume(queue: queueName,
                                 autoAck: true,
                                 consumer: consumer);

            Console.WriteLine(" Press [enter] to exit.");
            Console.ReadLine();
        }

还有其他2中工作模式大家可以参考其他博客或官网

分享两道面试题：
面试题：

避免消息堆积？

1） 采用workqueue，多个消费者监听同一队列。

2）接收到消息以后，而是通过线程池，异步消费。

如何避免消息丢失？

1） 消费者的ACK机制。可以防止消费者丢失消息。

但是，如果在消费者消费之前，MQ就宕机了，消息就没了？

2）可以将消息进行持久化。要将消息持久化，前提是：队列、Exchange都持久化

交换机持久化

队列持久化

消息持久化

 

根据B站朝夕教育的rabbitmq的学习视频，

同时参考博文：RabbitMQ快速入门（详细）_kavito的博客-CSDN博客_rabbitmq