120行代码打造.netcore生产力工具-小而美的后台异步组件

2020/06/11 12:10
阅读数 63


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

  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的话,那也是最好不过的。小生这厢先行谢过。

展开阅读全文
打赏
0
0 收藏
分享
加载中
更多评论
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部