kestrel网络编程--开发redis服务器

1 文章目的

本文讲解基于kestrel开发实现了部分redis命令的redis伪服务器的过程，让读者了解kestrel网络编程的完整步骤，其中redis通讯协议需要读者自行查阅，文章里不做具体解析。

2 开发顺序

创建Kestrel的Redis协议处理者
配置监听的EndPoint并使用Redis处理者
设计交互上下文RedisContext
设计Redis命令处理者
设计Redis中间件
编排Redis中间件构建应用

3. 创建Redis协议处理者

在Kestrel中，末级的中间件是一个没有next的特殊中间件，基表现出来就是一个ConnectionHandler的行为。我们开发redis应用只需要继承ConnectionHandler这个抽象类来，当kestrel接收到新的连接时将连接交给我们来处理，我们处理完成之后，不再有下一个处理者来处理这个连接了。

/// <summary>

/// 表示Redis连接处理者

/// </summary>

sealed class RedisConnectionHandler : ConnectionHandler

{

    /// <summary>

    /// 处理Redis连接

    /// </summary>

    /// <param name="context">redis连接上下文</param>

    /// <returns></returns>

    public async override Task OnConnectedAsync(ConnectionContext context)

    {

        // 开始处理这个redis连接

        ...

        // 直到redis连接断开后结束

    }

}

4. 配置监听的EndPoint

4.1 json配置文件

我们在配置文件里指定监听本机的5007端口来做服务器，当然你可以指定本机具体的某个IP或任意IP。

{

  "Kestrel": {

    "Endpoints": {

      "Redis": { // redis协议服务器，只监听loopback的IP

        "Url": "http://localhost:5007"

      }

    }

  }

}

{

  "Kestrel": {

    "Endpoints": {

      "Redis": { // redis协议服务器，监听所有IP

        "Url": "http://*:5007"

      }

    }

  }

}

4.2 在代码中配置Redis处理者

为Redis这个节点关联上RedisConnectionHandler，当redis客户端连接到5007这个端口之后，OnConnectedAsync()方法就得到触发且收到连接上下文对象。

builder.WebHost.ConfigureKestrel((context, kestrel) =>

{

    var section = context.Configuration.GetSection("Kestrel");

    kestrel.Configure(section).Endpoint("Redis", endpoint =>

    {

        endpoint.ListenOptions.UseConnectionHandler<RedisConnectionHandler>();

    });

});

5 设计RedisContext

在asp.netcore里，我们知道应用层每次http请求都创建一个HttpContext对象，里面就塞着各种与本次请求有关的对象。对于Redis的请求，我们也可以这么抄袭asp.netcore来设计Redis。

5.1 RedisContext

Redis请求上下文，包含Client、Request、Response和Features对象，我们要知道是收到了哪个Redis客户端的什么请求，从而请求命令处理者可以向它响应对应的内容。

/// <summary>

/// 表示redis上下文

/// </summary>

sealed class RedisContext : ApplicationContext

{

    /// <summary>

    /// 获取redis客户端

    /// </summary>

    public RedisClient Client { get; }

    /// <summary>

    /// 获取redis请求

    /// </summary>

    public RedisRequest Reqeust { get; }

    /// <summary>

    /// 获取redis响应

    /// </summary>

    public RedisResponse Response { get; }

    /// <summary>

    /// redis上下文

    /// </summary>

    /// <param name="client"></param>

    /// <param name="request"></param>

    /// <param name="response"></param>

    /// <param name="features"></param>

    public RedisContext(RedisClient client, RedisRequest request, RedisResponse response, IFeatureCollection features)

        : base(features)

    {

        this.Client = client;

        this.Reqeust = request;

        this.Response = response;

    }

    public override string ToString()

    {

        return $"{this.Client} {this.Reqeust}";

    }

}

5.2 ApplicationContext

这是抽象的应用层上下文，它强调Features，做为多个中间件之间的沟通渠道。

/// <summary>

/// 表示应用程序请求上下文

/// </summary>

public abstract class ApplicationContext

{

    /// <summary>

    /// 获取特征集合

    /// </summary>

    public IFeatureCollection Features { get; }

    /// <summary>

    /// 应用程序请求上下文

    /// </summary>

    /// <param name="features"></param>

    public ApplicationContext(IFeatureCollection features)

    {

        this.Features = new FeatureCollection(features);

    }

}

5.3 RedisRequest

一个redis请求包含请求的命令和0到多个参数值。

/// <summary>

/// 表示Redis请求

/// </summary>

sealed class RedisRequest

{

    private readonly List<RedisValue> values = new();

    /// <summary>

    /// 获取命令名称

    /// </summary>

    public RedisCmd Cmd { get; private set; }

    /// <summary>

    /// 获取参数数量

    /// </summary>

    public int ArgumentCount => this.values.Count - 1;

    /// <summary>

    /// 获取参数

    /// </summary>

    /// <param name="index"></param>

    /// <returns></returns>

    public RedisValue Argument(int index)

    {

        return this.values[index + 1];

    }

}

RedisRequest的解析：

/// <summary>

/// 从内存中解析

/// </summary>

/// <param name="memory"></param>

/// <param name="request"></param>

/// <exception cref="RedisProtocolException"></exception>

/// <returns></returns>

private static bool TryParse(ReadOnlyMemory<byte> memory, [MaybeNullWhen(false)] out RedisRequest request)

{

    request = default;

    if (memory.IsEmpty == true)

    {

        return false;

    }

    var span = memory.Span;

    if (span[0] != '*')

    {

        throw new RedisProtocolException();

    }

    if (span.Length < 4)

    {

        return false;

    }

    var lineLength = span.IndexOf((byte)'\n') + 1;

    if (lineLength < 4)

    {

        throw new RedisProtocolException();

    }

    var lineCountSpan = span.Slice(1, lineLength - 3);

    var lineCountString = Encoding.ASCII.GetString(lineCountSpan);

    if (int.TryParse(lineCountString, out var lineCount) == false || lineCount < 0)

    {

        throw new RedisProtocolException();

    }

    request = new RedisRequest();

    span = span.Slice(lineLength);

    for (var i = 0; i < lineCount; i++)

    {

        if (span[0] != '$')

        {

            throw new RedisProtocolException();

        }

        lineLength = span.IndexOf((byte)'\n') + 1;

        if (lineLength < 4)

        {

            throw new RedisProtocolException();

        }

        var lineContentLengthSpan = span.Slice(1, lineLength - 3);

        var lineContentLengthString = Encoding.ASCII.GetString(lineContentLengthSpan);

        if (int.TryParse(lineContentLengthString, out var lineContentLength) == false)

        {

            throw new RedisProtocolException();

        }

        span = span.Slice(lineLength);

        if (span.Length < lineContentLength + 2)

        {

            return false;

        }

        var lineContentBytes = span.Slice(0, lineContentLength).ToArray();

        var value = new RedisValue(lineContentBytes);

        request.values.Add(value);

        span = span.Slice(lineContentLength + 2);

    }

    request.Size = memory.Span.Length - span.Length;

    Enum.TryParse<RedisCmd>(request.values[0].ToString(), ignoreCase: true, out var name);

    request.Cmd = name;

    return true;

}

5.4 RedisResponse

/// <summary>

/// 表示redis回复

/// </summary>

sealed class RedisResponse

{

    private readonly PipeWriter writer;

    public RedisResponse(PipeWriter writer)

    {

        this.writer = writer;

    }

    /// <summary>

    /// 写入\r\n

    /// </summary>

    /// <returns></returns>

    public RedisResponse WriteLine()

    {

        this.writer.WriteCRLF();

        return this;

    }

    public RedisResponse Write(char value)

    {

        this.writer.Write((byte)value);

        return this;

    }

    public RedisResponse Write(ReadOnlySpan<char> value)

    {

        this.writer.Write(value, Encoding.UTF8);

        return this;

    }

    public RedisResponse Write(ReadOnlyMemory<byte> value)

    {

        this.writer.Write(value.Span);

        return this;

    }

    public ValueTask<FlushResult> FlushAsync()

    {

        return this.writer.FlushAsync();

    }

    public ValueTask<FlushResult> WriteAsync(ResponseContent content)

    {

        return this.writer.WriteAsync(content.ToMemory());

    }

}

5.5 RedisClient

Redis是有状态的长连接协议，所以在服务端，我把连接接收到的连接包装为RedisClient的概念，方便我们业务理解。对于连接级生命周期的对象属性，我们都应该放到RedisClient上，比如是否已认证授权等。

/// <summary>

/// 表示Redis客户端

/// </summary>

sealed class RedisClient

{

    private readonly ConnectionContext context;

    /// <summary>

    /// 获取或设置是否已授权

    /// </summary>

    public bool? IsAuthed { get; set; }

    /// <summary>

    /// 获取远程终结点

    /// </summary>

    public EndPoint? RemoteEndPoint => context.RemoteEndPoint;

    /// <summary>

    /// Redis客户端

    /// </summary>

    /// <param name="context"></param>

    public RedisClient(ConnectionContext context)

    {

        this.context = context;

    }

    /// <summary>

    /// 关闭连接

    /// </summary>

    public void Close()

    {

        this.context.Abort();

    }

    /// <summary>

    /// 转换为字符串

    /// </summary>

    /// <returns></returns>

    public override string? ToString()

    {

        return this.RemoteEndPoint?.ToString();

    }

}

6. 设计Redis命令处理者

redis命令非常多，我们希望有一一对应的cmdHandler来对应处理，来各尽其责。所以我们要设计cmdHandler的接口，然后每个命令增加一个实现类型，最后使用一个中间件来聚合这些cmdHandler。

6.1 IRedisCmdHanler接口

/// <summary>

/// 定义redis请求处理者

/// </summary>

interface IRedisCmdHanler

{

    /// <summary>

    /// 获取能处理的请求命令

    /// </summary>

    RedisCmd Cmd { get; }

    /// <summary>

    /// 处理请求

    /// </summary>

    /// <param name="context"></param>

    /// <returns></returns>

    ValueTask HandleAsync(RedisContext context);

}

6.2 IRedisCmdHanler实现

由于实现类型特别多，这里只举个例子

/// <summary>

/// Ping处理者

/// </summary>

sealed class PingHandler : IRedisCmdHanler

{

    public RedisCmd Cmd => RedisCmd.Ping;

    /// <summary>

    /// 处理请求

    /// </summary>

    /// <param name="context"></param>

    /// <returns></returns>

    public async ValueTask HandleAsync(RedisContext context)

    {

        await context.Response.WriteAsync(ResponseContent.Pong);

    }

}

7.设计Redis中间件

对于Redis服务器应用而言，我们处理一个请求需要经过多个大的步骤：

如果服务器要求Auth的话,验证连接是否已Auth
如果Auth验证通过之后，则查找与请求对应的IRedisCmdHanler来处理请求
如果没有IRedisCmdHanler来处理，则告诉客户端命令不支持。

7.1 中间件接口

/// <summary>

/// redis中间件

/// </summary>

interface IRedisMiddleware : IApplicationMiddleware<RedisContext>

{

}

/// <summary>

/// 应用程序中间件的接口

/// </summary>

/// <typeparam name="TContext"></typeparam>

public interface IApplicationMiddleware<TContext>

{

    /// <summary>

    /// 执行中间件

    /// </summary>

    /// <param name="next">下一个中间件</param>

    /// <param name="context">上下文</param>

    /// <returns></returns>

    Task InvokeAsync(ApplicationDelegate<TContext> next, TContext context);

}

7.2 命令处理者中间件

这里只拿重要的命令处理者中间件来做代码说明，其它中间件也是一样处理方式。

/// <summary>

/// 命令处理中间件

/// </summary>

sealed class CmdMiddleware : IRedisMiddleware

{

    private readonly Dictionary<RedisCmd, IRedisCmdHanler> cmdHandlers;

    public CmdMiddleware(IEnumerable<IRedisCmdHanler> cmdHanlers)

    {

        this.cmdHandlers = cmdHanlers.ToDictionary(item => item.Cmd, item => item);

    }

    public async Task InvokeAsync(ApplicationDelegate<RedisContext> next, RedisContext context)

    {

        if (this.cmdHandlers.TryGetValue(context.Reqeust.Cmd, out var hanler))

        {

            // 这里是本中间件要干的活

            await hanler.HandleAsync(context);

        }

        else

        {

            // 本中间件干不了，留给下一个中间件来干

            await next(context);

        }

    }

}

8 编排Redis中间件

回到RedisConnectionHandler，我们需要实现它，实现逻辑是编排Redis中间件并创建可以处理应用请求的委托application，再将收到的redis请求创建RedisContext对象的实例，最后使用application来执行RedisContext实例即可。

8.1 构建application委托

sealed class RedisConnectionHandler : ConnectionHandler

{

    private readonly ILogger<RedisConnectionHandler> logger;

    private readonly ApplicationDelegate<RedisContext> application;

    /// <summary>

    /// Redis连接处理者

    /// </summary>

    /// <param name="appServices"></param>

    /// <param name="logger"></param>

    public RedisConnectionHandler(

        IServiceProvider appServices,

        ILogger<RedisConnectionHandler> logger)

    {

        this.logger = logger;

        this.application = new ApplicationBuilder<RedisContext>(appServices)

            .Use<AuthMiddleware>()

            .Use<CmdMiddleware>()

            .Use<FallbackMiddlware>()

            .Build();

    }

}

8.2 使用application委托处理请求

sealed class RedisConnectionHandler : ConnectionHandler

{

    /// <summary>

    /// 处理Redis连接

    /// </summary>

    /// <param name="context"></param>

    /// <returns></returns>

    public async override Task OnConnectedAsync(ConnectionContext context)

    {

        try

        {

            await this.HandleRequestsAsync(context);

        }

        catch (Exception ex)

        {

            this.logger.LogDebug(ex.Message);

        }

        finally

        {

            await context.DisposeAsync();

        }

    }

    /// <summary>

    /// 处理redis请求

    /// </summary>

    /// <param name="context"></param>

    /// <returns></returns>

    private async Task HandleRequestsAsync(ConnectionContext context)

    {

        var input = context.Transport.Input;

        var client = new RedisClient(context);

        var response = new RedisResponse(context.Transport.Output);

        while (context.ConnectionClosed.IsCancellationRequested == false)

        {

            var result = await input.ReadAsync();

            if (result.IsCanceled)

            {

                break;

            }

            var requests = RedisRequest.Parse(result.Buffer, out var consumed);

            if (requests.Count > 0)

            {

                foreach (var request in requests)

                {

                    var redisContext = new RedisContext(client, request, response, context.Features);

                    await this.application.Invoke(redisContext);

                }

                input.AdvanceTo(consumed);

            }

            else

            {

                input.AdvanceTo(result.Buffer.Start, result.Buffer.End);

            }

            if (result.IsCompleted)

            {

                break;

            }

        }

    }

}

9 文章总结

在还没有进入阅读本文章之前，您可能会觉得我会大量讲解Socket知识内容，例如Socket Bind、Socket Accept、Socket Send、Socket Receive等。但实际上没完全没有任何涉及，因为终结点的监听、连接的接收、缓冲区的处理、数据接收与发送等这些基础而复杂的网络底层kestrel已经帮我处理好，我们关注是我们的应用协议层的解析、还有应用本身功能的开发两个本质问题。

您可能发也现了，本文章的RedisRequest解析，也没有多少行代码！反而文章中都是抽象的中间件、处理者、上下文等概念。实际上这不但不会带来项目复杂度，反而让项目更好的解耦，比如要增加一个新的指令的支持，只需要增加一个xxxRedisCmdHanler的文件，其它地方都不用任何修改。

本文章是KestrelApp项目里面的一个demo的讲解，希望对您有用。

巴特西