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传入
        object AfterReceiveRequest(ref Message request, IClientChannel channel, InstanceContext instanceContext);
　　　　　　// 在输出相应触发
        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} 
", 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次的时候： 

直接就断开了~

完整的代码