wcf利用IDispatchMessageInspector實現接口監控日志記錄和並發限流

一般對於提供出來的接口，雖然知道在哪些業務場景下才會被調用，但是不知道什么時候被調用、調用的頻率、接口性能，當出現問題的時候也不容易重現請求；為了追蹤這些內容就需要把每次接口的調用信息給完整的記錄下來，也就是記錄日志。日志中可以把調用方ip、服務器ip、調用時間點、時長、輸入輸出都給完整的記錄下來，有了這些數據，排查問題、重現異常、性能瓶頸都能准確的找到切入點。

這種功能，當然沒人想要去在每個Operation里邊插入一段代碼，如果有類似AOP的玩意就再好不過了。

wcf中有IDispatchMessageInspector分發消息檢查器這么個玩意，

namespace System.ServiceModel.Dispatcher
{
    using System;
    using System.ServiceModel;
    using System.ServiceModel.Channels;
    
    public interface IDispatchMessageInspector
    {
　　　　　　// 接收請求后觸發， 此方法的返回值將作為BeforeSendReply的第二個參數correlcationState傳入  9         object AfterReceiveRequest(ref Message request, IClientChannel channel, InstanceContext instanceContext);
　　　　　　// 在輸出相應觸發 10         void BeforeSendReply(ref Message reply, object correlationState);
    }
}

下面是英文解釋：

　　　　　　　　　

亦即允許我們對進出服務的消息進行檢查和修改，這看起來有點像mvc的過濾器

執行過程：  AfterReceiveRequest  ->  wcf操作 ->  BeforeSendReply

 

切入點找到了，需要做的就是實現這個接口方法，記錄日志，還可以計算一段時間內訪問的次數並設置相應的閥值也就是可以實現並發限流，限流的目的主要是防止惡意調用維持己方服務器的穩定。

實現思路： 

 public class ThrottleDispatchMessageInspector : IDispatchMessageInspector
    {
       public object AfterReceiveRequest(ref Message request, IClientChannel channel, InstanceContext instanceContext)
        {
　　　　　　　// [並發限流]
　　　　　　　以ContractName+OperationName 作為MemoryCache的鍵，值為調用次數； 設置絕對過期時間
　　　　　　　if(次數 > 設定的閥值) 
　　　　　　　　　　request.Close(); 直接關閉請求 
　　　　　　　　
 　　　　　　 // [log] 記錄請求輸入消息、開始時間、服務器ip、客戶端ip、 訪問的wcf契約和方法... 
　　　　　　 LogVO log
　　　　　　　// [log] 將日志實體作為返回值
　　　　　　 return log;

         }
　　     public void BeforeSendReply(ref Message reply, object correlationState) {
　　　　　　　　//[log]補充完整log的屬性： 請求結束時間， 調用時長...  然后將log丟入隊列（不直接插數據庫，防止日志記錄影響接口性能） 慢慢落地rds中
　　　　　　　　var log =　correlationState as LogVO;
　　　　　　　　log => Queue 
　　　　　}　　　　
　　}    

完整的代碼如下：

// 自定義分發消息檢查器
    public class ThrottleDispatchMessageInspector : IDispatchMessageInspector
    {
        //TODO 這兩個參數根據系統的配置處理方式存儲，作為示例就直接寫了 
        public static int throttleNum = 10; // 限流個數
        public static int throttleUnit = 4; // s

        CacheItemPolicy policy = new CacheItemPolicy();  //！ 過期策略，保證第一個set和之后set的絕對過期時間保持一致

        #region implement IDispatchMessageInspector

        // 此方法的返回值 將作為方法BeforeSendReply的第二個參數 object correlationState傳入
        public object AfterReceiveRequest(ref Message request, IClientChannel channel, InstanceContext instanceContext)
        {


            // 獲取ContractName和OperationName 用來作為緩存鍵
            var context = OperationContext.Current;
            string contractName = context.EndpointDispatcher.ContractName;
            string operationName = string.Empty;
            if (context.IncomingMessageHeaders.Action == null)
            {
                operationName = request.Properties.Values.LastOrDefault().ToString();
            }
            else
            {
                if (context.IncomingMessageHeaders.Action.Contains("/"))
                {
                    operationName = context.IncomingMessageHeaders.Action.Split('/').LastOrDefault();
                }
            }
            string throttleCacheKey = contractName + "_" + operationName;
            // 緩存當前請求頻率， 以內存緩存System.Runtime.Caching.MemoryCache為例(.net4.0+)
            ObjectCache cache = MemoryCache.Default;
            var requestCount = cache.Get(throttleCacheKey);
            int currRequestCount = 1;
            if (requestCount != null && int.TryParse(requestCount.ToString(), out currRequestCount))
            {
                // 訪問次數+1
                currRequestCount++;
                cache.Set(throttleCacheKey, currRequestCount, policy);  //必須保證過期策略和第一次set的時候一致，不然過期時間會有問題
            }
            else
            {
                policy.AbsoluteExpiration = DateTime.Now.AddSeconds(throttleUnit);
                cache.Set(throttleCacheKey, currRequestCount, policy);
            }

            // 如果當前請求數大於閥值，直接關閉
            if (currRequestCount > throttleNum)
            {
                request.Close();
            }

            //作為返回值 傳給BeforeSendReply 
            LogVO log = new LogVO
            {
                BeginTime = DateTime.Now,
                ContractName = contractName,
                OperationName = operationName,
                Request = this.MessageToString(ref request),
                Response = string.Empty
            };
            return log;
        }

        public void BeforeSendReply(ref Message reply, object correlationState)
        {

            // 補充AfterReceiveRequest 傳遞過來的日志實體的屬性, 記錄
            LogVO log = correlationState as LogVO;
            log.EndTime = DateTime.Now;
            log.Response = this.MessageToString(ref reply);
            log.Duration = (log.EndTime - log.BeginTime).TotalMilliseconds;

            //attention 為不影響接口性能，日志實體push進隊列（redis .etc），然后慢慢落地
            //TODO 這里直接寫文本啦~
            try
            {
                string logPath = "D:\\WcfLog.txt";
                if (!File.Exists(logPath))
                {
                    File.Create(logPath);
                }
                StreamWriter writer = new StreamWriter(logPath, true);
                writer.Write(string.Format("at {0} , {1} is called , duration: {2} \r\n", log.BeginTime, log.ContractName + "." + log.OperationName, log.Duration));
                writer.Close();
            }
            catch (Exception ex) { }
        }
        #endregion
}

這邊需要注意 ： 1. 類似於Web上的HttpContext.Current.Cache 這邊使用的是相應的內存緩存 MemoryCache，在更新緩存值的時候 過期策略要保持和初次設置的時候一致，如果沒傳入則緩存將不過期；

　　　　　　　  2. 並發限制的配置根據自身系統框架來設定，不要寫死

　　　　　　　  3. 日志記錄不要直接落rds，不然並發高了rds的連接數很容易爆，而且影響api的處理速度（可以push to redis， job/service land data）

　　　　　　　  4. 讀取wcf的消息載體Message的數據后，要重新寫入（這個方法是直接用老外的）

接着，只要自定義服務行為在ApplyDispatchBehavior方法里將自定義的分發消息檢查器給注入到分發運行時就可以了，直接貼代碼：

    // 應用自定義服務行為的2中方式：
    // 1. 繼承Attribute作為特性 服務上打上標示
    // 2. 繼承BehaviorExtensionElement, 然后修改配置文件
    public class ThrottleServiceBehaviorAttribute : Attribute, IServiceBehavior
    {
        #region implement IServiceBehavior
        public void AddBindingParameters(ServiceDescription serviceDescription, System.ServiceModel.ServiceHostBase serviceHostBase, System.Collections.ObjectModel.Collection<ServiceEndpoint> endpoints, System.ServiceModel.Channels.BindingParameterCollection bindingParameters)
        {

        }

        public void ApplyDispatchBehavior(ServiceDescription serviceDescription, System.ServiceModel.ServiceHostBase serviceHostBase)
        {
            foreach (ChannelDispatcher channelDispather in serviceHostBase.ChannelDispatchers)
            {
                foreach (var endpoint in channelDispather.Endpoints)
                {
                    // holyshit DispatchRuntime 
                    endpoint.DispatchRuntime.MessageInspectors.Add(new ThrottleDispatchMessageInspector());
                }
            }
        }

        public void Validate(ServiceDescription serviceDescription, System.ServiceModel.ServiceHostBase serviceHostBase)
        {

        }
        #endregion

        #region override BehaviorExtensionElement
        //public override Type BehaviorType
        //{
        //    get { return typeof(ThrottleServiceBehavior); }
        //}

        //protected override object CreateBehavior()
        //{
        //    return new ThrottleServiceBehavior();
        //}
        #endregion
    }

這邊，由於本人比較懶，直接就繼承Attribute后 將服務行為貼在服務上；更好的做法是 繼承 BehaviorExtensionElement ， 然后在配置文件里邊注冊自定義行為讓所有的接口走自定義檢查器的邏輯。

 

試驗： 閥值 10次每4秒

隨便弄個Service

 [ThrottleServiceBehavior]
    public class Service1 : IService1
    {
        public string GetData()
        {
            object num = MemoryCache.Default.Get("IService1_GetData") ?? "0";

            return string.Format("already request {0} times, Throttle is : {1} per {2} seconds", num, WcfDispatchMessageInspector.ThrottleDispatchMessageInspector.throttleNum, WcfDispatchMessageInspector.ThrottleDispatchMessageInspector.throttleUnit);
        }
    }

 

1. 四秒內刷5次：　　  

2. 超過10次的時候： 

直接就斷開了~

 

完整的代碼