PHP中98行出错代码打造.netcore生产力工具-小而美的后台异步组件

Posted 2020-12-10 tomatoc

相信绝大部分开发者都接触过用户注册的流程，通常情况下大概的流程如下所示：

1. 接收用户提交注册信息
2. 持久化注册信息（数据库+redis）
3. 发送注册成功短信（邮件）
4. 写操作日志（可选）

伪代码如下：

public async Task<IActionResult> Reg([FromBody] User user)
{
    _logger.LogInformation("持久化数据开始");
    await Task.Delay(50);
    _logger.LogInformation("持久化结束");
    _logger.LogInformation("发送短信开始");
    await Task.Delay(100);
    _logger.LogInformation("发送短信结束");
    _logger.LogInformation("操作日志开始");
    await _logRepository.Insert(new Log { Txt = "注册日志" });
    _logger.LogInformation("操作日志结束");
    return Ok("注册成功");
}

在以上的代码中，我使用Task.Delay方法阻塞主线程，用以模拟实际场景中的执行耗时。以上流程应该是包含了绝大部分注册流程所需要的操作。对于任何开发者来讲，以上业务流程没任何难度，无非是顺序的执行各个流程的代码即可。

稍微有点开发经验的应该会将以上的流程进行拆分，但有些人可能就要问了，为什么要拆分呢？拆分之后的代码应该怎么写呢？下面我们就来简单聊下如此场景的正确打开方式。

首先，注册成功的依据应该是是否成功的将用户信息持久化（至于是先持久化到数据库，异或是先写到redis不在本篇文章讨论的范畴），至于发送注册短信（邮件）以及写日志的操作应该不能成为影响注册是否成功的因素，而发送短信/邮件等相关操作通常情况下也是比较耗时的，所以在对此接口做性能优化时，可优先考虑将短信/邮件以及写日志等相关操作与主流程（持久化数据）拆分，使其不阻塞主流程的执行，从而达到提高响应速度的目的。

知道了为什么要拆，但具体如何拆分呢？怎样才能用最少的改动，达到所需的目的呢？

条条大路通罗马，所以要达成我们的目的也是有很多方案的，具体选择哪种方案需要根据具体的业务场景，业务体量等多种因素综合考虑，下面我将一一介绍分析相关方案。

在正式介绍可用方案前，笔者想先介绍一种很多新手容易错误使用的一种方案（因为笔者就曾经天真的使用过这种错误的方案）。

提到异步，绝大部分.net开发者应该第一想到的就是Task，async，await等，的确，async，await的语法糖简化了.net开发者异步编程的门槛，减少了很多代码量。通常一个返回Task类型的方法，在被调用时，会在方法的前面加上await，表示需要等待此方法的执行结果，再继续执行后面的代码。但如果不加await时，则不会等待方法的执行结果，进而也不会阻塞主线程。所以，有些人可能就会将发送短信/邮件以及写日志的操作如下方式进行改造。

public async Task<IActionResult> Reg1([FromBody] User user)
{
    _logger.LogInformation("持久化数据开始");
    await Task.Delay(50);
    _logger.LogInformation("持久化结束");
    _ = Task.Run(async () =>
     {
         _logger.LogInformation("发送短信开始");
         await Task.Delay(100);
         _logger.LogInformation("发送短信结束");
         _logger.LogInformation("操作日志开始");
         await _logRepository.Insert(new Log { Txt = "注册日志" });
         _logger.LogInformation("操作日志结束");
     });
    return Ok("注册成功");
}

然后使用jmeter分别压测改造前和改造后的接口，结果如下：

有没有被惊讶到？就这样一个简单的改造，吞吐量就提高了三四倍。既然已经提高了三四倍，那为什么说这是一种错误的改造方法吗？各位看官且往下看。

熟悉.netcore的大佬，应该都知道.netcore的依赖注入的生命周期吧。通常情况下，注入的生命周期包括：Singleton，Scope，Transient。
在以上的流程中，假如写操作日志的实例的生命周期是Scope，当在Task中调用Controller获取到的实例的方法时，因为Task.Run并没有阻塞主线程，当调用Action return后，当前请求的scope注入的对象会被回收，如果对象会回收之前，Task.Run还未执行完，则会报System.ObjectDisposedException: Cannot access a disposed object. 异常。意思是，不能访问一个已disposed的对象。正确的做法是使用IServiceScopeFactory创建一个新的作用域，在新的作用域中获取获取日志仓储服务的实例。这样就可以避免System.ObjectDisposedException异常了。
改造后的示例代码如下：

public async Task<IActionResult> Reg1([FromBody] User user)
{
    _logger.LogInformation("持久化数据开始");
    await Task.Delay(50);
    _logger.LogInformation("持久化结束");
    _ = Task.Run(async () =>
    {
        using (var scope = _scopeFactory.CreateScope())
        {
            var sp = scope.ServiceProvider;
            var logRepository = sp.GetService<ILogRepository>();
            _logger.LogInformation("发送短信开始");
            await Task.Delay(100);
            _logger.LogInformation("发送短信结束");

            _logger.LogInformation("操作日志开始");
            await logRepository.Insert(new Log { Txt = "注册日志" });
            _logger.LogInformation("操作日志结束");
        }
    });
    return Ok("注册成功");
}

虽然得到了正解，但上述的代码着实有点多，如果一个项目有多个相似的业务场景，就要考虑对CreateScope相关的操作进行封装。

下面就来一一介绍下笔者觉得实现此业务场景的几种方案。
1.消息队列
2.Quartz任务调度组件
3.Hangfire任务调度组件
4.Weshare.TransferJob(推荐)
首先说下消息队列的方式。准确的说，消息队列应该是这种场景的最优解决方案，消息队列的其中一个比较重要的特性就是解耦，从而提高吞吐量。但并不是所有的应用程序都需要上消息队列。有些业务场景使用消息队列时，往往会给人一种"杀鸡用牛刀"的感觉。

其次Quartz和Hangfire都是任务调度框架，都提供了可实现以上业务场景的逻辑，但Quartz和Hangfire都需要持久化作业数据。虽然Hangfire提供了内存版本，但经过我的测试，发现Hangfire的内存版本特别消耗内存，所以不太推荐使用任务调度框架来实现类似于这样的业务逻辑。

最后，也就是本文的重点，笔者结合了消息队列和任务调度的思想，实现了一个轻量级的转移作业到后台执行的组件。此组件完美的解决了Scope生命周期实例获取的问题，一行代码将不需要等待的操作转移到后台线程执行。
接入步骤如下：
1.使用nuget安装Weshare.TransferJob
2.在Stratup中注入服务。

services.AddTransferJob();

3.通过构造函数或其他方法获取到IBackgroundRunService的实例。
4.调用实例的Transfer方法将作业转移到后台线程。

_backgroundRunService.Transfer(log=>log.Insert(new Log(){Txt = "注册日志"}));

就是这么简单的实现了这样的业务场景，不仅简化了代码，而且大大提高了系统的吞吐量。

下面再来一起分析下Weshare.TransferJob的核心代码（毕竟文章要点题）。各位器宇不凡的看官请继续往下看。
下面的代码是AddTransferJob方法的实现：

public static IServiceCollection AddTransferJob(this IServiceCollection services)
{
    services.AddSingleton<IBackgroundRunService, BackgroundRunService>();
    services.AddHostedService<TransferJobHostedService>();
    return services;
}

聪明"绝顶"的各位看官应该已经发现上述代码的关键所在。是的， 你没有看错，此组件的就是利用.net core提供的HostedService在后台执行被转移的作业的。
我们再来一起看看TransferJobHostedService的代码：

public class TransferJobHostedService:BackgroundService
{
    private IBackgroundRunService _runService;
    public TransferJobHostedService(IBackgroundRunService runService)
    {
        _runService = runService;
    }
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            await _runService.Execute(stoppingToken);
        }
    }
}

这个类的代码也很简单，重写了BackgroundService类的ExecuteAsync，循环调用IBackgroundRunService实例的Execute方法。所以，最最关键的代码是IBackgroundRunService的实现类中。
详细代码如下：

public class BackgroundRunService : IBackgroundRunService
{
    private readonly SemaphoreSlim _slim;
    private readonly ConcurrentQueue<LambdaExpression> queue;
    private ILogger<BackgroundRunService> _logger;
    private readonly IServiceProvider _serviceProvider;
    public BackgroundRunService(ILogger<BackgroundRunService> logger, IServiceProvider serviceProvider)
    {
        _slim = new SemaphoreSlim(1);
        _logger = logger;
        _serviceProvider = serviceProvider;
        queue = new ConcurrentQueue<LambdaExpression>();
    }
    public async Task Execute(CancellationToken cancellationToken)
    {
        try
        {
            await _slim.WaitAsync(cancellationToken);
            if (queue.TryDequeue(out var job))
            {
                using (var scope = _serviceProvider.GetRequiredService<IServiceScopeFactory>().CreateScope())
                {
                    var action = job.Compile();
                    var isTask = action.Method.ReturnType == typeof(Task);
                    var parameters = job.Parameters;
                    var pars = new List<object>();
                    if (parameters.Any())
                    {
                        var type = parameters[0].Type;
                        var param = scope.ServiceProvider.GetRequiredService(type);
                        pars.Add(param);
                    }
                    if (isTask)
                    {
                        await (Task)action.DynamicInvoke(pars.ToArray());
                    }
                    else
                    {
                        action.DynamicInvoke(pars.ToArray());
                    }
                }
            }
        }
        catch (Exception e)
        {
            _logger.LogError(e.ToString());
        }
    }
    public void Transfer<T>(Expression<Func<T, Task>> expression)
    {
        queue.Enqueue(expression);
        _slim.Release();
    }
    public void Transfer(Expression<Action> expression)
    {
        queue.Enqueue(expression);
        _slim.Release();
    }
}

纳尼？嫌代码多看不懂？那咱们一起来剖析下吧。
首先，此类有三个较重要的私有变量，对应的类型分别是SemaphoreSlim， ConcurrentQueue，IServiceProvider。
其中SemaphoreSlim是为了控制后台作业执行的顺序的，在构造函数中初始化了此对象的信号量为1，表示在后台服务的ExecuteAsync方法的循环中每次只能有一个作业执行。
ConcurrentQueue的对象是用来存储被转移到后台服务执行的作业的逻辑，所以使用LambdaExpression作为队列的类型。
IServiceProvider是为了解决依赖注入的生命周期的。

然后在Execute方法中，第一行代码如下：

await _slim.WaitAsync(cancellationToken);

作用是等待一个信号量，当没有可用的信号量时，会阻塞线程的执行，这样在后台服务的ExecuteAsync方法的死循环就不会一直执行下去，只有获取到信号量才会继续执行。
当获取到信号量后，则说明有新的作业等待执行，所以此时则需要从队列中读出要执行的LambdaExpression表达式，创建一个新的Scope后，编译此表达式树，判断返回类型，获取泛型的具体类型，最后获取到泛型对应的实例，执行对应的方法。

另外，Transfer方法就是暴露给调用者的方法，用于将表达式树写到队列中，同时释放信号量。

到此为止，Weshare.TransferJob的实现原理已分析完毕，由于此组件的原理只是将任务转移到后台进行执行，所以并不是适合对事务有要求的场景。正如本文开头所假设的场景，TransferJob最适合的场景还是那些和主操作关联性较低的、失败或成功并不会影响业务的正常运行。
同时，此组件的定位就是小而美，像延迟执行、定时执行的功能在最初的规划中其实是有的，后来发现这些功能quartz已经有了，所以没必要重复造这样的轮子。
后期会根据使用场景，尝试加入异常重试机制，以及异常通知回调机制。

最后，不知道有没有较真的看官想计算下代码量是否超过120行。
为了证明我不是标题党，现将此组件进行开源，地址是：
https://github.com/fuluteam/WeShare.TransferJob

桥豆麻袋，笔者辛苦敲的代码，难道各位看官想白嫖吗? 点个赞再走呗。点完赞还有力气的话，如果git上能点个star的话，那也是最好不过的。小生这厢先行谢过。