RabbitMQ 发布订阅版本

public class R_MQHelper
    {
        /// <summary>
        /// 获取Rabbitmq连接对象
        /// </summary>
        /// <returns></returns>
        public static IConnection GetConnection() 
        {
            var connFactory = new ConnectionFactory()
            {
                HostName = "127.0.0.1",
                Port=5672,
                UserName="guest",
                Password="guest"
            };
            return connFactory.CreateConnection();// 返回 创建连接对象
        }
    }

首先  ，加一个上边的类。

====生产者代码：

static void Main(string[] args)
        {
            using var connection = R_MQHelper.GetConnection();//创建连接对象
            //创建信道
            using var channel = connection.CreateModel();
            //创建交换机
            channel.ExchangeDeclare("chn_Exchange","fanout",false,false,null);
            //创建队列
            channel.QueueDeclare("chn_queue001",false,false,false,null);
            channel.QueueDeclare("chn_queue002", false, false, false, null);
            channel.QueueDeclare("chn_queue003", false, false, false, null);
            //把队列绑定到交换机（就是队列订阅交换机）
            channel.QueueBind("chn_queue001", "chn_Exchange","",null);
            channel.QueueBind("chn_queue002", "chn_Exchange", "", null);
            channel.QueueBind("chn_queue003", "chn_Exchange", "", null);

            for (int i = 0; i < 10; i++)
            {
                string msg = $"chn is changzhi small pirlo{i+1}";//陈浩南是长治小皮尔洛
                var body = Encoding.UTF8.GetBytes(msg);
                //发送消息
                //创建了交换机则不需要routingKey了，因为只需要队列订阅交换机即可
                //消息发送到交换机，创建了的队列  订阅了 交换机 ，就会自动去交换机拿值
                channel.BasicPublish("chn_Exchange","",false,null,body);
                Console.WriteLine("已经生产消息："+msg);
                Thread.Sleep(500);
            }

            Console.ReadKey();
        }

=======消费者代码：

static void Main(string[] args)
        {
            //创建连接对象
            var connection = R_MQHelper.GetConnection();
            //创建 通信管道（即信道）
            var channel = connection.CreateModel();

            /*
                说明：
                        在消费者创建队列、交换机是以防范 先启动消费者  而引发报错
                        如：队列已存在的情况下可不用再次创建   此创建为：确保先开启消费者，生产者未创建队列而引发的报错
             */
            //创建交换机
            channel.ExchangeDeclare("chn_Exchange", "fanout", false, false, null);
            //创建队列
            channel.QueueDeclare("chn_queue001", false, false, false, null);
            channel.QueueDeclare("chn_queue002", false, false, false, null);
            channel.QueueDeclare("chn_queue003", false, false, false, null);
            //把队列绑定到交换机（就是队列订阅交换机）
            channel.QueueBind("chn_queue001", "chn_Exchange", "", null);
            channel.QueueBind("chn_queue002", "chn_Exchange", "", null);
            channel.QueueBind("chn_queue003", "chn_Exchange", "", null);

            //事件对象
            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) => {
                //获取消息
                var body = ea.Body;//获取队列中消息主体
                var msg = Encoding.UTF8.GetString(body.ToArray());
                var routingKey = ea.RoutingKey;//队列名
                Console.WriteLine($"recive message is :{msg}  ,routingKey is: {routingKey}");
            };

            //消费消息
            //可以一次消费多个队列消息（也可以消费单个队列）
            channel.BasicConsume("chn_queue001", true,consumer);
            channel.BasicConsume("chn_queue002", true, consumer);
            channel.BasicConsume("chn_queue003", true, consumer);

            Console.ReadKey();
        }

这时候，可以打开rabbitmq页面，观察相关数据