RabbitMQ実戦経験共有

38066 ワード
前言
最近1つの大学入試の項目に忙しくて、システムが順調に今回の大学入試を完成したのを見て、やっとほっとすることができます.大学入試を受ける学生を見て、その年の高校3年生の自分を思い出した.
RabbitMQの実戦経験を共有し、皆さんの役に立つことを望んでいます.
 
 
一、生産情報
RabbitMQの基本使用についてはここでは紹介しませんが、プロジェクトではExchangeでのtopicモードを使用しています.
メッセージを先に送信するコード
private bool MarkErrorSend(string[] lstMsg)
        {
            try
            {
                var factory = new ConnectionFactory()
                {
                    UserName = "guest",//   
                    Password = "guest",//  
                    HostName = "localhost",//ConfigurationManager.AppSettings["sHostName"],
                };
                //    
                var connection = factory.CreateConnection();
                //    
                var channel = connection.CreateModel();
                try
                {
                    //    Direct     
                    channel.ExchangeDeclare(
                        exchange: "TestTopicChange", //exchange  
                        type: ExchangeType.Topic, //Topic  ，      
                        durable: true,//exchange   
                        autoDelete: false,//      ，    false
                        arguments: null//       ，  :alternate-exchange
                        );

                    //      
                    channel.QueueDeclare(
                        queue: "Test_Queue", //    
                        durable: true, //       (              )
                        exclusive: false,//     ，false。             ，             ，           
                        autoDelete: false,//      ，    false
                        arguments: null
                        );

                    //         
                    string routeKey = "TestRouteKey.*";//*      
                    channel.QueueBind(
                        queue: "Test_Queue",
                        exchange: "TestTopicChange",
                        routingKey: routeKey,
                        arguments: null
                        );

                    //       ， DeliveryMode  2（              ）
                    IBasicProperties properties = channel.CreateBasicProperties();
                    properties.DeliveryMode = 2;

                    channel.ConfirmSelect();//      
                    foreach (var itemMsg in lstMsg)
                    {
                        byte[] sendBytes = Encoding.UTF8.GetBytes(itemMsg);
                        //    
                        channel.BasicPublish(
                            exchange: "TestTopicChange",
                            routingKey: "TestRouteKey.one",
                            basicProperties: properties,
                            body: sendBytes
                            );
                    }
                    bool isAllPublished = channel.WaitForConfirms();//  （channel）           true
                    return isAllPublished;
                }
                catch (Exception ex)
                {
                    //     
                    return false;
                }
                finally
                {
                    channel.Close();
                    connection.Close();
                }
            }
            catch
            {
                //RabbitMQ.Client.Exceptions.BrokerUnreachableException:
                //When the configured hostname was not reachable.
                return false;
            }
        }

メッセージを送るのは特にありません.ニュースの持続化についての紹介はここでも紹介しません.分からないのは前の文章を読むことができます.メッセージを送信する上で注意しなければならないのは、複数のメッセージを一緒に送信することを選択して、最後にメッセージの送信に成功したことを確定することができて、このように効率が高いことです.また、各メッセージの長さをできるだけ短くする必要があります(スレ主はここで損をしたことがあります).そうしないと、メッセージが長すぎるため、送信時間が長くなります.実際のプロジェクトでは一度に4万件以上のデータを送ったが、問題はなかった.
 
 
二、受信メッセージ
次に、メッセージを消費するプロセスについてお話しします.私は単一の複数のchannelを接続し、各channelは毎回1つのメッセージメソッドしか取りません.単一のTCP接続で、複数のchannelが通信効率に影響するかどうかを聞く人がいます.この理論にはきっと影響があるに違いない.影響は大きくないと思う.私が運転しているchannelの数は普通30ぐらいまで行って、効率に影響を与えていないのは、私がchannelごとにメッセージを持っているからかもしれません.複数のchannelを単一に接続する方法で、多くの接続を少なくすることができます.私がなぜチャンネルごとに1つのメッセージしか取らないのかについては、外部の要因が制限されており、具体的には自分のニーズを見ています.
次にメッセージを受信するプロセスでは、まずRabbitMQHelperクラスを定義します.グローバルなconn接続変数が含まれています.また、接続の作成、接続のクローズ、接続が開いているかどうかを確認する方法もあります.プログラムはタイマーを実行し、
接続が開いていないことが検出された場合、接続処理メッセージをアクティブに作成します.
 public class RabbitMQHelper
    {
        public IConnection conn = null;

        /// 
        ///   RabbitMQ       
        /// 
        ///       
        public IConnection RabbitConnection(string sHostName, ushort nChannelMax)
        {
            try
            {
                if (conn == null)
                {
                    var factory = new ConnectionFactory()
                    {
                        UserName = "guest",//   
                        Password = "guest",//  
                        HostName = sHostName,//ConfigurationManager.AppSettings["MQIP"],
                        AutomaticRecoveryEnabled = false,//      ，               
                        RequestedConnectionTimeout = 10000,//        10 ，    30 
                        RequestedChannelMax = nChannelMax//          
                    };
                    //    
                    conn = factory.CreateConnection();
                    Console.WriteLine("RabbitMQ     ！");
                }

                return conn;
            }
            catch
            {
                Console.WriteLine("      ，   RabbitMQ      ！");
                return null;
            }
        }

        /// 
        ///   RabbitMQ  
        /// 
        public void Close()
        {
            try
            {
                if (conn != null)
                {
                    if (conn.IsOpen)
                        conn.Close();
                    conn = null;
                    Console.WriteLine("RabbitMQ     ！");
                }
            }
            catch { }
        }

        /// 
        ///   RabbitMQ      
        /// 
        /// 
        public bool IsOpen()
        {
            try
            {
                if (conn != null)
                {
                    if (conn.IsOpen)
                        return true;
                }
                return false;
            }
            catch
            {
                return false;
            }
        }
    }

 

       接下来我们看具体如何接收消息。

private static AutoResetEvent myEvent = new AutoResetEvent(false);
private RabbitMQHelper rabbit = new RabbitMQHelper();
private ushort nChannel = 10;//

まずrabbitインスタンスを初期化し、RabbitConnectionメソッドでRabbitMQ接続を作成します.
接続が開くと、スレッドプールでメッセージを受信する方法が実行されます.ここで開いているスレッドは、開いているchannelの数と一致しなければならないことに注意してください.そうしないと問題が発生します(具体的な問題は、RabbitMQ接続タイムアウト時間を10秒に設定し、使用しない場合があります.原因は不明です.RabbitMQ接続のデフォルトタイムアウト時間は30秒で、この時間内に作成を呼び出すと、2倍のchannelが得られる可能性があります.)/// /// RabbitMQ ， channel /// private void CreateConnecttion() { try { rabbit.RabbitConnection("localhost", nChannel); if (rabbit.conn != null) { ThreadPool.SetMinThreads(1, 1); ThreadPool.SetMaxThreads(100, 100); for (int i = 1; i <= nChannel; i++) { ThreadPool.QueueUserWorkItem(new WaitCallback(ReceiveMsg), ""); } myEvent.WaitOne();// ， rabbit.Close(); } } catch (Exception ex) { rabbit.Close(); Console.WriteLine(ex.Message); } }
 
次にメッセージを受信する方法であり,メッセージを処理する過程は省略される. /// /// ， （channel）， /// /// RabbitMQ private void ReceiveMsg(object obj) { IModel channel = null; try { #region ， channel = rabbit.conn.CreateModel(); channel.ExchangeDeclare( exchange: "TestTopicChange", //exchange type: ExchangeType.Topic, //Topic ， durable: true,//exchange autoDelete: false,// ， false arguments: null// ， :alternate-exchange ); // channel.QueueDeclare( queue: "Test_Queue", // durable: true, // ( ) exclusive: false,// ，false。 ， ， autoDelete: false, arguments: null ); #endregion channel.BasicQos(0, 1, false);// channel.QueueBind(queue: "Test_Queue", exchange: "TestTopicChange", routingKey: "TestRouteKey.*"); var consumer = new EventingBasicConsumer(channel); consumer.Received += (model, ea) => { var body = ea.Body; var message = Encoding.UTF8.GetString(body); var routingKey = ea.RoutingKey; // try { bool isMark = AutoMark(message); if (isMark) { //Function.writeMarkLog(message); // , RabbitMQ channel.BasicAck(ea.DeliveryTag, false); } else { if (MarkErrorSend(message))// channel.BasicReject(ea.DeliveryTag, false); else // ， ， ， channel.BasicReject(ea.DeliveryTag, true); } } catch (Exception ex) { try { Console.WriteLine(ex.Message); if (channel != null && channel.IsOpen)// RabbitMQ { // ， ， ， channel.BasicReject(ea.DeliveryTag, true); } } catch { } } }; // channel.BasicConsume(queue: "Test_Queue", autoAck: false, consumer: consumer); } catch (Exception ex) { try { Console.WriteLine(" ：" + ex.Message); if (channel != null && channel.IsOpen)// channel.Close(); if (rabbit.conn != null)// RabbitMQ rabbit.Close(); } catch { } } }
 
 
三、エラーメッセージの処理
処理に失敗したメッセージを「エラーキュー」に配置し、その後、元のキューのメッセージを削除します(ここで主な問題は、処理に失敗したり処理できないメッセージが複数ある場合、これらのメッセージを元のキューに戻すと、他のスレッドのchannelに配布され続けますが、結果的に処理できず、デッドサイクルをもたらし、後のメッセージが処理できなくなります).最初に処理できなかったメッセージを「エラーキュー」に配置した後、新しい接続を再開して「エラーキュー」のメッセージを処理します./// /// “ ” /// /// /// private bool MarkErrorSend(string msg) { RabbitMQHelper MQ = new RabbitMQHelper(); MQ.RabbitConnection("localhost",1); // var channel = MQ.conn.CreateModel(); try { // Direct channel.ExchangeDeclare( exchange: "ErrorTopicChange", //exchange type: ExchangeType.Topic, //Topic ， durable: true,//exchange autoDelete: false,// ， false arguments: null// ， :alternate-exchange ); // channel.QueueDeclare( queue: "Error_Queue", // durable: true, // ( ) exclusive: false,// ，false。 ， ， autoDelete: false,// ， false arguments: null ); // string routeKey = "ErrorRouteKey.*";//* channel.QueueBind( queue: "Error_Queue", exchange: "ErrorTopicChange", routingKey: routeKey, arguments: null ); // ， DeliveryMode 2（ ） IBasicProperties properties = channel.CreateBasicProperties(); properties.DeliveryMode = 2; channel.ConfirmSelect();// byte[] sendBytes = Encoding.UTF8.GetBytes(msg); // channel.BasicPublish( exchange: "ErrorTopicChange", routingKey: "ErrorRouteKey.one", basicProperties: properties, body: sendBytes ); bool isAllPublished = channel.WaitForConfirms();// （channel） true return isAllPublished; } catch (Exception ex) { // return false; } finally { channel.Close(); MQ.conn.Close(); } }
 
まとめ:RabbitMQ自体は安定しており、性能もよく、すべての不安定な要素がメッセージを処理する過程にあるので、安心して使用できます.
Demoソースアドレス:https://github.com/Bingjian-Zhu/RabbitMQHelper
python gb 2312回転utf-8
Androidはどのように最も優れてActivityの中で資源を釈放することを開発します