ICache基础上,进一步封装了缓存架构ICacheProvider,用于业务应用中常用的分布式缓存、本地缓存和消息队列。强烈推荐使用ICacheProvider替代ICache

Nuget包:NewLife.Core

源码:https://github.com/NewLifeX/X/blob/master/NewLife.Core/Caching/ICacheProvider.cs

快速入门

NET项目从Nuget引入NewLife.Redis,启动代码加入以下代码,服务层直取ICacheProvider。

// 分布式服务,使用配置中心RedisCache配置
services.AddSingleton<ICacheProvider, RedisCacheProvider>();

该代码注入了ICacheProvider的Redis实现,其中Cache即是FullRedis实现,GetQueue得到的也是Redis消息队列。RedisCacheProvider将从配置中心或本地配置文件读取名为RedisCache的redis连接字符串来实例化FullRedis,如果没有配置该连接字符串,则内部自动降级为内存缓存。完美实现了代码同时兼容单机部署(无Redis)和集群部署(有Redis)。

设计思路

根据实际开发经验,即使在分布式系统中,也有大量的数据是不需要跨进程共享的,因此本接口提供了两级缓存。

进程内缓存使用InnerCache,可以规避对象序列化成本,跨进程缓存使用Cache

借助该缓存架构,可以实现各功能模块跨进程共享数据,分布式部署时可用Redis,需要考虑序列化成本。

使用队列时,可根据是否设置消费组来决定使用简单队列还是完整队列。

简单队列(如RedisQueue)可用作命令队列,Topic很多,但几乎没有消息。

完整队列(如RedisStream)可用作消息队列,Topic很少,但消息很多,并且支持多消费组。

消息队列

根据所使用ICache的不同,GetQueue返回的消息队列也不同。

IProducerConsumer<T> GetQueue<T>(String topic, String? group = null);
IProducerConsumer<T> GetInnerQueue<T>(String topic);

GetQueue有两个参数,topic是队列主题,group是消费组。对于RedisCacheProvider实现,未指定group时返回RedisQueue,指定group时返回RedisStream。

GetInnerQueue只有topic这一个参数,默认就是内存队列。

案例分析

星尘StarServer节点管理NodeController,通过引入ICacheProvider实现了内部命令所使用的消息队列。

private async Task ConsumeMessage(WebSocket socket, Node node, String ip, CancellationTokenSource source)
{
    DefaultSpan.Current = null;
    var cancellationToken = source.Token;
    var queue = _cacheProvider.GetQueue<String>($"nodecmd:{node.Code}");
    try
    {
        while (!cancellationToken.IsCancellationRequested && socket.State == WebSocketState.Open)
        {
            ISpan span = null;
            var mqMsg = await queue.TakeOneAsync(30);
            if (mqMsg != null)
            {
                // 埋点
                span = _tracer?.NewSpan($"mq:NodeCommand", mqMsg);

                // 解码
                var dic = JsonParser.Decode(mqMsg);
                var msg = JsonHelper.Convert<CommandModel>(dic);
                span.Detach(dic);

                if (msg == null || msg.Id == 0 || msg.Expire.Year > 2000 && msg.Expire < DateTime.Now)
                    WriteHistory(node, "WebSocket发送", false, "消息无效或已过期。" + mqMsg, ip);
                else
                {
                    WriteHistory(node, "WebSocket发送", true, mqMsg, ip);

                    // 向客户端传递埋点信息,构建完整调用链
                    msg.TraceId = span + "";

                    var log = NodeCommand.FindById(msg.Id);
                    if (log != null)
                    {
                        if (log.TraceId.IsNullOrEmpty()) log.TraceId = span?.TraceId;
                        log.Times++;
                        log.Status = CommandStatus.处理中;
                        log.UpdateTime = DateTime.Now;
                        log.Update();
                    }

                    await socket.SendAsync(mqMsg.GetBytes(), WebSocketMessageType.Text, true, cancellationToken);
                }

                span?.Dispose();
            }
            else
            {
                await Task.Delay(1_000, cancellationToken);
            }
        }
    }
    catch (TaskCanceledException) { }
    catch (Exception ex)
    {
        XTrace.WriteLine("WebSocket异常 node={0} ip={1}", node, ip);
        XTrace.WriteException(ex);
        WriteHistory(node, "WebSocket断开", false, ex.ToString(), ip);
    }
    finally
    {
        source.Cancel();
    }
}

这里的_cacheProvider.GetQueue只有topic没有group,而Startup里面注入了RedisCacheProvider。

在单机部署时,使用内存队列,StarWeb向StarServer发送的节点指令,进入队列后,这里能够直接消费得到。

在集群部署时,StarServer运行有多实例,都要配置相同的RedisCache连接字符串。StarWeb调用接口发送指令时,可能节点的WebSocket刚好不在这台服务器上。接收StarWeb接口指令的StarServer实例,把指令放入队列(此时是Redis),目标节点WebSocket所在的StarServer实例将能够从Redis队列中得到这个指令,并通过WebSocket下发给节点客户端。

作者:大石头 发布:2023-11-22 21:49:10 浏览:348)