RabbitMQ從零到集群高可用.NetCore(.NET5) - RabbitMQ簡(jiǎn)介和六種工作模式詳解
2021-10-22 09:57:55

#15672端口為web管理端的端口，5672為RabbitMQ服務(wù)的端口
docker run -d  --name rabbitmq -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=123456 -p 15672:15672 -p 5672:5672 rabbitmq:3-management

/// <summary>
    /// RabbitMQ幫助類
    /// </summary>
    public class RabbitMQHelper
    {
        private static ConnectionFactory factory;
        private static object lockObj = new object();
        /// <summary>
        /// 獲取單個(gè)RabbitMQ連接
        /// </summary>
        /// <returns></returns>
        public static IConnection GetConnection()
        {
            if (factory == null)
            {
                lock (lockObj)
                {
                    if (factory == null)
                    {
                         factory = new ConnectionFactory
                        {
                            HostName = "172.16.2.84",//ip
                            Port = 5672,//端口
                            UserName = "admin",//賬號(hào)
                            Password = "123456",//密碼
                            VirtualHost = "develop" //虛擬主機(jī)
                        };
                    }
                }
            }
            return factory.CreateConnection();
        }
    }

public static void SimpleSendMsg()
        {
            string queueName = "simple_order";//隊(duì)列名
            //創(chuàng)建連接
            using (var connection = RabbitMQHelper.GetConnection())
            {
                //創(chuàng)建信道
                using (var channel = connection.CreateModel())
                {//創(chuàng)建隊(duì)列
                    channel.QueueDeclare(queueName, durable: false, exclusive: false, autoDelete: false, arguments: null);
                    for (var i = 0; i < 10; i++)
                    {
                        string message = $"Hello RabbitMQ MessageHello,{i + 1}";
                        var body = Encoding.UTF8.GetBytes(message);//發(fā)送消息
                        channel.BasicPublish(exchange: "", routingKey: queueName, mandatory: false, basicProperties: null, body);
                        Console.WriteLine($"發(fā)送消息到隊(duì)列:{queueName},內(nèi)容:{message}");
                    }
                }
            }
        }

public static void SimpleConsumer()
        {
            string queueName = "simple_order";
            var connection = RabbitMQHelper.GetConnection();
            {
                //創(chuàng)建信道
                var channel = connection.CreateModel();
                {
                    //創(chuàng)建隊(duì)列
                     channel.QueueDeclare(queueName, durable: false, exclusive: false, autoDelete: false, arguments: null);
                    var consumer = new EventingBasicConsumer(channel);
                    int i = 0;
                    consumer.Received += (model, ea) =>
                    {
                        //消費(fèi)者業(yè)務(wù)處理
                        var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                        Console.WriteLine($"{i},隊(duì)列{queueName}消費(fèi)消息長(zhǎng)度:{message.Length}");
                        i++;
                    };
                    channel.BasicConsume(queueName, true, consumer);
                }
            }
        }

/// <summary>
        /// 工作隊(duì)列模式
        /// </summary>
        public static void WorkerSendMsg()
        {
            string queueName = "worker_order";//隊(duì)列名
            //創(chuàng)建連接
            using (var connection = RabbitMQHelper.GetConnection())
            {
                //創(chuàng)建信道
                using (var channel = connection.CreateModel())
                {
                    //創(chuàng)建隊(duì)列
                    channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, arguments: null);
　　　　　　　　　　　　 var properties = channel.CreateBasicProperties();
　　　　　　　　　　　　　properties.Persistent = true; //消息持久化
                    for ( var i=0;i<10;i++)
                    {
                        string message = $"Hello RabbitMQ MessageHello,{i+1}";
                        var body = Encoding.UTF8.GetBytes(message);
                        //發(fā)送消息到rabbitmq
                        channel.BasicPublish(exchange: "", routingKey: queueName, mandatory: false, basicProperties: properties, body);
                        Console.WriteLine($"發(fā)送消息到隊(duì)列:{queueName},內(nèi)容:{message}");
                    }
                }
            }
        }

public static void WorkerConsumer()
        {
            string queueName = "worker_order";
            var connection = RabbitMQHelper.GetConnection();
            {
                //創(chuàng)建信道
                var channel = connection.CreateModel();
                {
                    //創(chuàng)建隊(duì)列
                    channel.QueueDeclare(queueName, durable: false, exclusive: false, autoDelete: false, arguments: null);
                    var consumer = new EventingBasicConsumer(channel);
                    //prefetchCount:1來(lái)告知RabbitMQ,不要同時(shí)給一個(gè)消費(fèi)者推送多于 N 個(gè)消息，也確保了消費(fèi)速度和性能
                    channel.BasicQos(prefetchSize: 0, prefetchCount:1, global: false);
                    int i = 1;
                    int index = new Random().Next(10);
                    consumer.Received += (model, ea) =>
                    {
                        //處理業(yè)務(wù)
                        var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                        Console.WriteLine($"{i},消費(fèi)者:{index},隊(duì)列{queueName}消費(fèi)消息長(zhǎng)度:{message.Length}");
                         channel.BasicAck(ea.DeliveryTag, false); //消息ack確認(rèn)，告訴mq這條隊(duì)列處理完，可以從mq刪除了

Thread.Sleep(1000);

i++;
                    };
                    channel.BasicConsume(queueName,autoAck:false, consumer);
                }
            }
        }

/// <summary>
        /// 發(fā)布訂閱， 扇形隊(duì)列
        /// </summary>
        public static void SendMessageFanout()
        {
            //創(chuàng)建連接
            using (var connection = RabbitMQHelper.GetConnection())
            {
                //創(chuàng)建信道
                using (var channel = connection.CreateModel())
                {
                    string exchangeName = "fanout_exchange";
                    //創(chuàng)建交換機(jī),fanout類型
                    channel.ExchangeDeclare(exchangeName, ExchangeType.Fanout);
                    string queueName1 = "fanout_queue1";
                    string queueName2 = "fanout_queue2";
                    string queueName3 = "fanout_queue3";
                    //創(chuàng)建隊(duì)列
                    channel.QueueDeclare(queueName1, false, false, false);
                    channel.QueueDeclare(queueName2, false, false, false);
                    channel.QueueDeclare(queueName3, false, false, false);

                    //把創(chuàng)建的隊(duì)列綁定交換機(jī),routingKey不用給值，給了也沒(méi)意義的
                    channel.QueueBind(queue: queueName1, exchange: exchangeName, routingKey: "");
                    channel.QueueBind(queue: queueName2, exchange: exchangeName, routingKey: "");
                    channel.QueueBind(queue: queueName3, exchange: exchangeName, routingKey: "");
　　　　　　　　　　　　　var properties = channel.CreateBasicProperties();
　　　　　　　　　　　　　properties.Persistent = true; //消息持久化
                    //向交換機(jī)寫10條消息
                    for (int i = 0; i < 10; i++)
                    {
                        string message = $"RabbitMQ Fanout {i + 1} Message";
                        var body = Encoding.UTF8.GetBytes(message);
                        channel.BasicPublish(exchangeName, routingKey: "", null, body);
                        Console.WriteLine($"發(fā)送Fanout消息:{message}");
                    }
                }
            }
        }

/// <summary>
        /// 路由模式，點(diǎn)到點(diǎn)直連隊(duì)列
        /// </summary>
        public static void SendMessageDirect()
        {
            //創(chuàng)建連接
            using (var connection = RabbitMQHelper.GetConnection())
            {
                //創(chuàng)建信道
                using (var channel = connection.CreateModel())
                {
                    //聲明交換機(jī)對(duì)象,fanout類型
                    string exchangeName = "direct_exchange";
                    channel.ExchangeDeclare(exchangeName, ExchangeType.Direct);
                    //創(chuàng)建隊(duì)列
                    string queueName1 = "direct_errorlog";
                    string queueName2 = "direct_alllog";
                    channel.QueueDeclare(queueName1, true, false, false);
                    channel.QueueDeclare(queueName2, true, false, false);

                    //把創(chuàng)建的隊(duì)列綁定交換機(jī),direct_errorlog隊(duì)列只綁定routingKey:error
                    channel.QueueBind(queue: queueName1, exchange: exchangeName, routingKey: "error");
                    //direct_alllog隊(duì)列綁定routingKey:error,info
                    channel.QueueBind(queue: queueName2, exchange: exchangeName, routingKey: "info");
                    channel.QueueBind(queue: queueName2, exchange: exchangeName, routingKey: "error");
　　　　　　　　　　　　　var properties = channel.CreateBasicProperties();
　　　　　　　　　　　　　properties.Persistent = true; //消息持久化
                    //向交換機(jī)寫10條錯(cuò)誤日志和10條Info日志
                    for (int i = 0; i < 10; i++)
                    {
                        string message = $"RabbitMQ Direct {i + 1} error Message";
                        var body = Encoding.UTF8.GetBytes(message);
                        channel.BasicPublish(exchangeName, routingKey: "error", properties, body);
                        Console.WriteLine($"發(fā)送Direct消息error:{message}");

                        string message2 = $"RabbitMQ Direct {i + 1} info Message";
                        var body2 = Encoding.UTF8.GetBytes(message);
                        channel.BasicPublish(exchangeName, routingKey: "info", properties, body2);
                        Console.WriteLine($"info:{message2}");

                    }
                }
            }
        }

public static void DirectConsumer()
        {
            string queueName = "direct_errorlog";
            var connection = RabbitMQHelper.GetConnection();
            {
                //創(chuàng)建信道
                var channel = connection.CreateModel();
                {
                    //創(chuàng)建隊(duì)列
                    channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, arguments: null);
                    var consumer = new EventingBasicConsumer(channel);
                    ///prefetchCount:1來(lái)告知RabbitMQ,不要同時(shí)給一個(gè)消費(fèi)者推送多于 N 個(gè)消息，也確保了消費(fèi)速度和性能
                    ///global：是否設(shè)為全局的
                    ///prefetchSize:單條消息大小，通常設(shè)0，表示不做限制
                    //是autoAck=false才會(huì)有效
                    channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: true);
                    int i = 1;
                    consumer.Received += (model, ea) =>
                    {
                        //處理業(yè)務(wù)
                        var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                        Console.WriteLine($"{i},隊(duì)列{queueName}消費(fèi)消息長(zhǎng)度:{message.Length}");
                        channel.BasicAck(ea.DeliveryTag, false); //消息ack確認(rèn)，告訴mq這條隊(duì)列處理完，可以從mq刪除了
                        i++;
                    };
                    channel.BasicConsume(queueName, autoAck: false, consumer);
                }
            }
        }

public static void SendMessageTopic()
        {
            //創(chuàng)建連接
            using (var connection = RabbitMQHelper.GetConnection())
            {
                //創(chuàng)建信道
                using (var channel = connection.CreateModel())
                {
                    //聲明交換機(jī)對(duì)象,fanout類型
                    string exchangeName = "topic_exchange";
                    channel.ExchangeDeclare(exchangeName, ExchangeType.Topic);
                    //隊(duì)列名
                    string queueName1 = "topic_queue1";
                    string queueName2 = "topic_queue2";
                    //路由名
                    string routingKey1 = "*.orange.*";
                    string routingKey2 = "*.*.rabbit";
                    string routingKey3 = "lazy.#";
                    channel.QueueDeclare(queueName1, true, false, false);
                    channel.QueueDeclare(queueName2, true, false, false);

                    //把創(chuàng)建的隊(duì)列綁定交換機(jī),routingKey指定routingKey
                    channel.QueueBind(queue: queueName1, exchange: exchangeName, routingKey: routingKey1);
                    channel.QueueBind(queue: queueName2, exchange: exchangeName, routingKey: routingKey2);
                    channel.QueueBind(queue: queueName2, exchange: exchangeName, routingKey: routingKey3);
                    //向交換機(jī)寫10條消息
                    for (int i = 0; i < 10; i++)
                    {
                        string message = $"RabbitMQ Direct {i + 1} Message";
                        var body = Encoding.UTF8.GetBytes(message);
                        channel.BasicPublish(exchangeName, routingKey: "aaa.orange.rabbit", null, body);
                        channel.BasicPublish(exchangeName, routingKey: "lazy.aa.rabbit", null, body);
                        Console.WriteLine($"發(fā)送Topic消息:{message}");
                    }
                }
            }
        }

public class RPCServer
    {
        public static void RpcHandle()
        {

            var connection = RabbitMQHelper.GetConnection();
            {
                var channel = connection.CreateModel();
                {
                    string queueName = "rpc_queue";
                    channel.QueueDeclare(queue: queueName, durable: false,
                      exclusive: false, autoDelete: false, arguments: null);
                    channel.BasicQos(0, 1, false);
                    var consumer = new EventingBasicConsumer(channel);
                    channel.BasicConsume(queue: queueName,
                      autoAck: false, consumer: consumer);
                    Console.WriteLine("【服務(wù)端】等待RPC請(qǐng)求...");

                    consumer.Received += (model, ea) =>
                    {
                        string response = null;

                        var body = ea.Body.ToArray();
                        var props = ea.BasicProperties;
                        var replyProps = channel.CreateBasicProperties();
                        replyProps.CorrelationId = props.CorrelationId;

                        try
                        {
                            var message = Encoding.UTF8.GetString(body);
                            Console.WriteLine($"【服務(wù)端】接收到數(shù)據(jù):{ message},開始處理");
                            response = $"消息:{message},處理完成";
                        }
                        catch (Exception e)
                        {
                            Console.WriteLine("錯(cuò)誤:" + e.Message);
                            response = "";
                        }
                        finally
                        {
                            var responseBytes = Encoding.UTF8.GetBytes(response);
                            channel.BasicPublish(exchange: "", routingKey: props.ReplyTo,
                              basicProperties: replyProps, body: responseBytes);
                            channel.BasicAck(deliveryTag: ea.DeliveryTag,
                              multiple: false);
                        }
                    };
                }
            }
        }

    }

public class RPCClient
    {
        private readonly IConnection connection;
        private readonly IModel channel;
        private readonly string replyQueueName;
        private readonly EventingBasicConsumer consumer;
        private readonly BlockingCollection<string> respQueue = new BlockingCollection<string>();
        private readonly IBasicProperties props;

        public RPCClient()
        {
            connection = RabbitMQHelper.GetConnection();

            channel = connection.CreateModel();
            replyQueueName = channel.QueueDeclare().QueueName;
            consumer = new EventingBasicConsumer(channel);

            props = channel.CreateBasicProperties();
            var correlationId = Guid.NewGuid().ToString();
            props.CorrelationId = correlationId; //給消息id
            props.ReplyTo = replyQueueName;//回調(diào)的隊(duì)列名，Client關(guān)閉后會(huì)自動(dòng)刪除

            consumer.Received += (model, ea) =>
            {
                var body = ea.Body.ToArray();
                var response = Encoding.UTF8.GetString(body);
                //監(jiān)聽的消息Id和定義的消息Id相同代表這條消息服務(wù)端處理完成
                if (ea.BasicProperties.CorrelationId == correlationId)
                {
                    respQueue.Add(response);
                }
            };

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

        public string Call(string message)
        {
            var messageBytes = Encoding.UTF8.GetBytes(message);
            //發(fā)送消息
            channel.BasicPublish(
                exchange: "",
                routingKey: "rpc_queue",
                basicProperties: props,
                body: messageBytes);
            //等待回復(fù)
            return respQueue.Take();
        }

        public void Close()
        {
            connection.Close();
        }
    }

static void Main(string[] args)
        {
            Console.WriteLine("Hello World!");
            //啟動(dòng)服務(wù)端，正常邏輯是在另一個(gè)程序
            RPCServer.RpcHandle();
            //實(shí)例化客戶端
            var rpcClient = new RPCClient();
            string message = $"消息id:{new Random().Next(1, 1000)}";
            Console.WriteLine($"【客服端】RPC請(qǐng)求中，{message}");
            //向服務(wù)端發(fā)送消息，等待回復(fù)
            var response = rpcClient.Call(message);
            Console.WriteLine("【客服端】收到回復(fù)響應(yīng):{0}", response);
            rpcClient.Close();
            Console.ReadKey();
        }