记一个并发规则验证实现

Posted 2022-07-12 dotNET跨平台

最近在做一个简单的风控，其中有一块需求是这样的，当主请求参数到达后，会根据这些参数，看调起几个并发规则，这些规则各自有自己的验证逻辑，每个规则执行时间长短都不确定，当规则 执行完后，返回主请求，主请求根据规则验证返回结果，从而决定是否立即response请求，但其他后到的规则 ，要继续完成后面验证，以得到验证结果，以备后用。

初步的.net代码是这样实现的：

using System.Threading.Channels;


var builder = WebApplication.CreateBuilder(args);


var app = builder.Build();


app.MapGet("/test", async () =>

    Console.WriteLine($"开始时间 DateTime.Now.ToString("HH:mm:ss")");
    var channels = new List<Channel<Parameter>>();
    var len = 5;
    //创建Channel,并关联ReadAsync方法
    for (int i = 0; i < len; i++)
    
        var channel = Channel.CreateUnbounded<Parameter>(new UnboundedChannelOptions()  AllowSynchronousContinuations = true );
        channels.Add(channel);
        await Task.Factory.StartNew(async () =>
        
            Console.WriteLine($"Task i");
            await ReadAsync(channel);
        );
    
    //向Channel发送信息
    for (int i = 0; i < channels.Count; i++)
    
        var channel = channels[i];
        await channel.Writer.WriteAsync(new Parameter  I = i );
        Console.WriteLine($"Write i");
    


    //读取返回Channel，如果有返回True，API提前返回，留下的Channel给RemainingReadAsync执行
    for (int i = 0; i < channels.Count; i++)
    
        var channel = channels[i];
        if (channel != null && await channel.Reader.WaitToReadAsync())
        
            if (channel.Reader.TryRead(out var par))
            
                if(par.Result)
                
                    Console.ForegroundColor = ConsoleColor.Green;
                
                Console.WriteLine($"I:par.I,Result:par.Result,DateTime.Now.ToString("HH:mm:ss")");
                Console.ResetColor();
                if (par.Result)
                
                    //把剩会的推到一个线程中执行
                    for (var r = i + 1; r < channels.Count; r++)
                    
                        await Task.Factory.StartNew(async () =>
                        
                            await RemainingReadAsync(channels[r]);
                        );
                    
                    return TypedResults.Ok($"完成，有,DateTime.Now.ToString("HH:mm:ss")");


                
            
        
    
    return TypedResults.Ok($"完成，没有,DateTime.Now.ToString("HH:mm:ss")");
);


app.Run();
//处理剩余Channelr方法
async Task RemainingReadAsync(Channel<Parameter> channel)

    if (channel != null && await channel.Reader.WaitToReadAsync())
    
        if (channel.Reader.TryRead(out var par))
        
            Console.WriteLine($"I:par.I,Result:par.Result,DateTime.Now.ToString("HH:mm:ss")");
        
    

//读取Channel的方法
async Task ReadAsync(Channel<Parameter> channel)

    if (channel != null && await channel.Reader.WaitToReadAsync())
    
        if (channel.Reader.TryRead(out var par))
        
            await Task.Delay((par.I + 1) * 1000);
            if (par.I == 2)
            
                par.Result = true;
            
            else
            
                par.Result = false;
            
            await channel.Writer.WriteAsync(par);
        
    



//参数
class Parameter

    public int I  get; set; 
    public bool Result  get; set; 

下图是执行结果，当五个任务并发调起后，并发写入数据，执行中，第三个满足条件，于21:59:13返回请求，但后续的两个，依然保证完成。

这个Demo只是简单的完成了逻辑，关于性能还有待进一步测试验证改进。