使用Redis实现最近N条数据的决策

前言

很多时候，我们会根据用户最近一段时间的行为，做出一些相应的策略，从而改变系统的运动轨迹。

举个简单的例子来说明一下：

假设A公司现在有两个合作伙伴(B和C)，B和C都是提供天气数据的，现在A公司做了一个聚合接口，把B和C的接口融合了，那么这个时候，要怎么去B和C公司获取数据呢？

其实这个要考虑的东西有很多很多，下面根据本文的主题，拿出其中一个点来讨论说明。

最简单的做法就是，随机调用。当然不是那么简单的随机调用。

根据调用的最近一百条数据的得到成功率，耗时等指标，再根据这些指标去判断一次查询要去那个公司获取数据。

思路已经有了，这个时候就是怎么实践的问题了。

本文介绍的做法是借助redis来完成的。

如何用redis来处理

redis的list类型可以说非常适合用来处理这个情况。

首先，可以把查询按顺序写进去，一个个的入队。

其次，写进去之后可以对它进行裁剪，保留最近的100条数据。(换句话说，我们可以保证在这个list里面，最多就是100条数据)

最后，获取这个list里面的100条数据，进行计算即可。

正常情况下，我们不会把计算放在查询的过程里面，在查询的时候，只需要一个决策的结果值就可以了，当然这个结果值也是计算后写进redis的。

所以要将这个计算的过程从查询中独立出来，定时去执行即可。

总结上面所说的，大概可以画出下面这样一样图。

其中的第三步操作，将查询记录写进list，然后进行裁剪这两个操作，可以直接操作redis，也可以考虑通过MQ去写，虽说没什么太大的必要。

简单的示例代码

查询的控制器

[Route("api/[controller]")]

[ApiController]

public class AreaController : ControllerBase

{

    private readonly ILogger _logger;

    private readonly IRedisCachingProvider _provider;

    public AreaController(ILoggerFactory loggerFactory, IRedisCachingProvider provider)

    {

        _logger = loggerFactory.CreateLogger<AreaController>();

        _provider = provider;

    }

    // GET api/area/11

    [HttpGet("provinceId")]

    public async Task<string> GetAsync(string provinceId)

    {

        // get datasource

        var datasource = await GetQueryDataSourceIdAsync(provinceId);

        if (string.IsNullOrWhiteSpace(datasource)) return "not support";

        var beginTime = DateTime.Now;

        // query

        var (val, isSucceed) = await QueryDataSourceAsync(datasource);

        var endTime = DateTime.Now;

        // datasource

        var dsInfo = new DataSourceInfo

        {

            Cost = (long)endTime.Subtract(endTime).TotalMilliseconds,

            IsSucceed = isSucceed

        };

        // record

        _ = Task.Run(async () =>

        {

            try

            {

                await _provider.LPushAsync($"info:{datasource}", new List<DataSourceInfo> { dsInfo });

                await _provider.LTrimAsync($"info:{datasource}", 0, 99);

            }

            catch (Exception ex)

            {

                _logger.LogError(ex, $"record #{datasource}# error");

            }

        });

        return val;

    }

    private async Task<string> GetQueryDataSourceIdAsync(string provinceId)

    {

        var datasourceIds = GetDataSourceIdProvinceId(provinceId);

        if (datasourceIds.Count <= 0) return string.Empty;

        var cacheKey = "dskpi";

        var kpis = await _provider.HMGetAsync(cacheKey, datasourceIds);

        var datasource = datasourceIds.First();

        if (kpis != null && kpis.Any())

        {

            // policy

            datasource = kpis.OrderByDescending(x => x.Value).First().Key;

        }

        return datasource;

    }

    private async Task<(string val, bool isSucceed)> QueryDataSourceAsync(string datasource)

    {

        await Task.Delay(100);

        var rd = new Random().NextDouble();

        return (datasource, rd > 0.5d);

    }

    private List<string> GetDataSourceIdProvinceId(string provinceId)

    {

        return new List<string> { "100", "900" };

    }

}

由调度系统触发的计算控制器

[Route("api/cal")]

[ApiController]

public class CalculatiionController : ControllerBase

{

    private readonly ILogger _logger;

    private readonly IRedisCachingProvider _provider;

    public CalculatiionController(ILoggerFactory loggerFactory, IRedisCachingProvider provider)

    {

        _logger = loggerFactory.CreateLogger<CalculatiionController>();

        _provider = provider;

    }

    // GET api/cal/

    [HttpGet]

    public string Get()

    {

        var id = Guid.NewGuid().ToString("N");

        _ = Task.Run(async () => await CalAsync(id));

        return "ok";

    }

    private async Task CalAsync(string id)

    {

        _logger.LogInformation($"{id} begin at {DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}");

        var datasourceIds = GetAllDataSourceIds();

        foreach (var item in datasourceIds)

        {

            try

            {

                var topN = await _provider.LRangeAsync<DataSourceInfo>($"info:{item}", 0, 99);

                var cost = topN.Average(x => x.Cost);

                var rate = topN.Count(x => x.IsSucceed) / 100;

                var score = GetScore(cost, rate);

                await _provider.HSetAsync($"dskpi", item, score.ToString());

            }

            catch (Exception ex)

            {

                _logger.LogError(ex, $"{id} {item} calculate fail ...");

            }

        }

        _logger.LogInformation($"{id} end at {DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}");

    }

    private int GetScore(double cost, int rate)

    {

        return new Random().Next(1, 100);

    }

    private List<string> GetAllDataSourceIds()

    {

        return new List<string> { "100", "900" };

    }

}

也可以在Github上面找到上面的示例代码 RecentRecordsDemo

巴特西

使用Redis实现最近N条数据的决策

前言

如何用redis来处理

简单的示例代码

最新文章

热门文章