高并发解决方案orleans实践 - 星仔007
source link: https://www.cnblogs.com/morec/p/17038959.html
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.
高并发解决方案orleans实践
开具一张图，展开来聊天。有从单个服务、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>(0).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>(0).Shipment(orderId);
break;
case OrderStatus.Completed:
await orderService.GetGrain<IOrderGrains>(0).Completed(orderId);
break;
case OrderStatus.Rejected:
await orderService.GetGrain<IOrderGrains>(0).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");
}
}
}
}
结果发现channel竟然很迅速的完成500次请求，50并发数，只用了8秒钟。数据库库存扣减完毕，order生成100条。有去了解过orleans，理解它的设计思路。但是说不上来具体的。
这是启动的三兄弟：
下面只要启动了网关就会链接上服务，这里需要启动先后顺序，先服务后网关要不然网关会起不来。启动后服务掉线或者关掉都不会再影响网关，自动注册和发现。
通过swagger试了一条数据，服务正常，服务被服务端口5008,网关端口30002的执行了。
并发测试结果，看sql注释有提前把库存还原，order清空：
看看是哪些服务拿到了请求，因为500次请求只有100个库存所以后面就是一直库存不足，而且每个服务都参与了处理。
个人推荐用orleans做微服务集群，channel处理防止并发。其次是分布式锁redlock.net。
下面redlock.net执行结果,orleans集群上b表现还不错。在单机和consul只能完成50条。
consul不知道为什么特别慢，而且和单机一样遇到有些锁完全跑不完。对于channel和本地锁跑到consul上结果是乱的。
只有orlans表现相当突出。
结论，单机用本地锁或者channel。 consul和orleans我选orleans。 consul下只能分布式锁。orleans用channel就够了。
源代码：liuzhixin405/orleans-consul-cluster: orleans和consul并发测试 (github.com)
Recommend
About Joyk
Aggregate valuable and interesting links.
Joyk means Joy of geeK