高并发解决方案orleans实践

2023-01-10 14:52:23

开具一张图，展开来聊天。有从单个服务、consul集群和orleans来展开高并发测试一个小小数据库并发实例。

首先介绍下场景，创建一个order，同时去product表里面减掉一个库存。很简单的业务但是遇到并发问题在项目中就很头痛。

由于内容比较多，简单介绍了。

对外的接口很简单，客户端代码如下，通过不同的方法去控制并发问题，当然者都是在单个服务跑起来的时候。下面介绍下怎么去测试。

 [Route("api/[controller]/[action]")]
    [ApiController]
    public class OrderController : ControllerBase
    {
        private readonly IClusterClient orderService;
        public OrderController(IClusterClient orderService)
        {
            this.orderService = orderService;//500请求 并发50 . 100库存
        }

        [HttpPost]
        public async Task Create([FromServices] Channel<CreateOrderDto> channel, string sku, int count)
        {
            await channel.Writer.WriteAsync(new CreateOrderDto(sku, count));   //高并发高效解决方案  并发测试工具postjson_windows 10s
        }

        [HttpPost]
        public async Task CreateTestLock(string sku, int count)//非阻塞锁
        {
            await orderService.GetGrain<IOrderGrains>(Random.Shared.Next()).CreateTestLock(sku, count); //执行时间快,库存少量扣减 10s
        }
        [HttpPost]
        public async Task CreateBlockingLock(string sku, int count)//阻塞锁
        {
            await orderService.GetGrain<IOrderGrains>(Random.Shared.Next()).CreateBlockingLock(sku, count); //卖不完,时间长 50s
        }
        [HttpPost]
        public async Task CreateDistLock(string sku, int count) //colder组件 分布式锁
        {
            await orderService.GetGrain<IOrderGrains>(Random.Shared.Next()).CreateDistLock(sku, count); //库存扣完，时间长 50s
        }

        [HttpPost]
        public async Task CreateNetLock(string sku, int count)   //netlock.net锁 
        {
            await orderService.GetGrain<IOrderGrains>(Random.Shared.Next()).CreateNetLock(sku, count); //库存扣完，时间长 50s
        }

        static System.Threading.SpinLock semaphore = new SpinLock(false);
        [HttpPost]
        public async Task CreateLock(string sku, int count)   //卖不完
        {
            bool lockTaken = false;
            try
            {
                semaphore.Enter(ref lockTaken);
                await orderService.GetGrain<IOrderGrains>().CreateLock(sku, count);
            }
            finally
            {
                if (lockTaken)
                    semaphore.Exit();
            }
        }

        [HttpPost]
        public  void CreateLocalLock(string sku, int count)  //能卖完
        {
             orderService.GetGrain<IOrderGrains>(Random.Shared.Next()).CreateLocalLock(sku, count); //
        }

        [HttpPost]
        public async Task CreateNoLock(string sku, int count)
        {
           await orderService.GetGrain<IOrderGrains>(Random.Shared.Next()).CreateNoLock(sku, count); //乱的
        }
        [HttpGet]
        public async Task ChangeOrderStatus(int orderId, OrderStatus status)
        {
            switch (status)
            {
                case OrderStatus.Shipment:
                    await orderService.GetGrain<IOrderGrains>().Shipment(orderId);
                    break;
                case OrderStatus.Completed:
                    await orderService.GetGrain<IOrderGrains>().Completed(orderId);
                    break;
                case OrderStatus.Rejected:
                    await orderService.GetGrain<IOrderGrains>().Rejected(orderId);
                    break;
                default:
                    break;
            }
        }
    }

redis必不可少，有用到分布式锁。

高并发测试工具我用的是postjson_windows，自行百度。很方便。

下面简单说一下单机测试吧，上面的代码注释不多但是简单总结了结果。通过测试我个人认为channel是理想的方式，速度快，而且结果很完美，lock锁其次。其他结果或多或少有瑕疵。

但是如果部署成集群会怎样呢，因为lock锁和channel是进程内安全的，多开几个就以为这多开了几个进程，这样的话分布式锁是大家先想到的，但是结果呢

下面简单介绍下consul集群，以前有写过例子，这次拿来真的是很快就跑起来了。个就封装的consul注册中心和心跳检查的类库，第二个是我们的服务，第三个是我们的网关。具体代码后面有链接。consul自行安装，配置什么的默认。

这里部署了三个服务，把生成的代码拷贝了三份，分别执行。再启动网关。然后按照对应的方法名调用并发测试。（这里拷贝三份有点麻烦，后面orleans再看新的执行方法）

多开服务

dotnet eapi.dll --urls="http://*:5007" --ip="127.0.0.1" --port="5007" --weight=1
dotnet eapi.dll --urls="http://*:5008" --ip="127.0.0.1" --port="5008" --weight=2
dotnet eapi.dll --urls="http://*:5009" --ip="127.0.0.1" --port="5009" --weight=5

consul启动走命令，应对闪退
Consul.exe agent -dev

dotnet eapi.gateway.dll --urls="https://*:5000" --ip="127.0.0.1" --port="5000"

简单说下结果吧，集群下只有分布式锁能解决问题，但是只生成了50条数据，扣减库存也是50。channel和本地锁结果都跟想的差不多，是不对的。

下面说写orleans下的表现，首先eapi是服务，gateway是网关，interfaces就是网关和服务侨链的接口。这里代码跟上面的有些差别，就是事务实现方式不一样。因为运行起来的时候发现每次只能成功次，后面一直报错数据库链接被占用，所以我改掉了事务实现方式。因为前面的没这问题所以很纳闷，实际项目中也是感觉到orleans对数据库的链接管理是有点问题的，所以不去深究。

下面跑起来看看orleans怎么样，dotnet命令我们是可以指定appsettings.json的配置的，所以配置文件指定好OrleansOptions,只需要在生产代码目录orleans多实例服务\eapi\bin\Debug\net7.0下面执行者三个服务命令就开了三服务，下面执行下网关。

{
  "urls": "https://*:5005",
  "Logging": {
    "LogLevel": {
      "Default": "Information",
      "Microsoft.AspNetCore": "Warning"
    }
  },
  "AllowedHosts": "*",
  "distributedLock": {
    "LockType": "Redis",
    "RedisEndPoints": [ "127.0.0.1:6379" ]
  },
  "OrleansOptions": {
    "GatewayPort": 30001,
    "SiloPort": 1112
  }


}

  builder.Host.ConfigureDistributedLockDefaults();
            var gport = int.Parse(builder.Configuration["OrleansOptions:GatewayPort"]);
            var sport = int.Parse(builder.Configuration["OrleansOptions:SiloPort"]);
            builder.Host.UseOrleans(b => b.UseLocalhostClustering(sport, gport));

先说下遇到的问题， ReposioryBase<T>下面的代码原来返回的是iquerable<T> 这是大佬推荐的做法返回iquerable，结果到多服务就出现占用，我把事务换成现在的写法发现还是徒劳，直接查询出来ToListAsync就可以了。

public async Task<IReadOnlyList<T>> FindByCondition(Expression<Func<T, bool>> expression)
{
return await DbSet.Where(expression).ToListAsync();
}

orderService下面的代码

var product = (await _productRepository.FindByCondition(x => x.Sku.Equals(sku))).SingleOrDefault(); //执行一尺order创建后此处链接就不释放了。

if (product == null || product.Count < count)
{
_logger.LogInformation("库存不足,稍后重试");
return;
}
else
{
product.Count -= count;
}

下面的问题就是负载的问题，单个服务的话GetGrain<IOrderGrains>(0)没问题，多个服务这样的话所有请求都只会落到一个服务上，改成随机的三个服务就都可以了。

 internal class NotificationDispatcher : BackgroundService
    {
        private readonly ILogger<NotificationDispatcher> logger;
        private readonly Channel<CreateOrderDto> channel;
        private readonly IServiceProvider serviceProvider;
        public NotificationDispatcher( ILogger<NotificationDispatcher> logger, Channel<CreateOrderDto> channel, IServiceProvider serviceProvider)
        {
            this.logger = logger;
            this.channel = channel;
            this.serviceProvider = serviceProvider;
        }

        protected async override Task ExecuteAsync(CancellationToken stoppingToken)
        {
            while (!channel.Reader.Completion.IsCompleted)
            {
                var createOrderDto = await channel.Reader.ReadAsync();
                try
                {
                    using (var scope = serviceProvider.CreateScope())
                    {
                        var client = scope.ServiceProvider.GetRequiredService<IClusterClient>();
                        var orderService = client.GetGrain<IOrderGrains>(Random.Shared.Next()); //设置为0导致指挥获取一个服务,随机的话就是多服务负载
                        //var orderService = client.GetGrain<IOrderGrains>(0);
                        await orderService.Create(createOrderDto.sku, createOrderDto.count);
                    }
                }
                catch (Exception ex)
                {
                    logger.LogError(ex, "notification failed");
                }

            }
        }
    }