热门标签
当前位置: article > 正文
关于.net高并发商品秒杀方案（Redis+队列）_.net 消息队列+redis实现库存高并发设计
作者：羊村懒王 | 2024-02-15 11:39:36
赞
踩
.net 消息队列+redis实现库存高并发设计
四、Redis+队列：思路和服务器缓存+队列基本相同，使用Redis做缓存替代服务器缓存，Redis可以直接使用.net code进行管理也可安装RedisDesktopManager。
业务流程：接口触发-> 通过Redis获取库存和新增随机订单号->利用后台任务进行查询数据->新增订单\扣除库存
功能分析：
- Redis是单线程的,利用api自身的原子性，去除lock锁。Redis6.0支持多线程不在此次的讨论范围
- 数据可以持久化解决服务器异常容易导致缓存里的信息丢失
压力测试：
线程数为10，100，1000三种情况进行测试，Ramp-Up时间空，循环次数为1
搬砖开始：
首先到NuGet包管理器中安装StackExchange.Redis
appsettings.json中添加Redis连接信息
"RedisConnectionStrings": { //redis连接
"Connection": "127.0.0.1:6379,abortConnect=false",
"InstanceName": "RextecSOARedis"
}
- 1
- 2
- 3
- 4
到DBHelper新建RedisHelper.cs
public class RedisHelper { //连接字符串 private string _connectionString; //实例名称 private string _instanceName; //默认数据库 private int _defaultDB; private ConcurrentDictionary<string, ConnectionMultiplexer> _connections; public RedisHelper(string connectionString, string instanceName, int defaultDB = 0) { _connectionString = connectionString; _instanceName = instanceName; _defaultDB = defaultDB; _connections = new ConcurrentDictionary<string, ConnectionMultiplexer>(); } /// <summary> /// 获取ConnectionMultiplexer（表示可由多个线程并发访问的键/值对的线程安全集合。） /// </summary> /// <returns></returns> public ConnectionMultiplexer GetConnect() { return _connections.GetOrAdd(_instanceName, p => ConnectionMultiplexer.Connect(_connectionString)); } /// <summary> /// 获取数据库 /// </summary> /// <returns></returns> public IDatabase GetDatabase() { return GetConnect().GetDatabase(_defaultDB); } public IServer GetServer(string configName = null, int endPointsIndex = 0) { var confOption = ConfigurationOptions.Parse(_connectionString); return GetConnect().GetServer(confOption.EndPoints[endPointsIndex]); } public ISubscriber GetSubscriber(string configName = null) { return GetConnect().GetSubscriber(); } public void Dispose() { if (_connections != null && _connections.Count > 0) { foreach (var item in _connections.Values) { item.Close(); } } } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
Startup.cs中注册RedisHelper
services.AddDbContext<DbHelperContext>(option =>
{
option.UseSqlServer(Configuration[string.Join(":", new string[] { "DBConnection", "ConnectionStrings", "Dbconn" })]);
});
services.AddSingleton(new DBHelper.RedisHelper(Configuration[string.Join(":", new string[] { "RedisConnectionStrings", "Connection" })], ""));
- 1
- 2
- 3
- 4
- 5
在项目的Common文件夹（不是Common类库）新建RCacheBackService.cs、RCustomerService.cs
RCacheBackService.cs：
public class RCacheBackService: BackgroundService { private readonly IDatabase _redis; private readonly IConfiguration _Configuration; public RCacheBackService(DBHelper.RedisHelper redisHelper, IConfiguration Configuration) { _redis = redisHelper.GetDatabase(); _Configuration = Configuration; } protected async override Task ExecuteAsync(CancellationToken stoppingToken) { // EFCore的上下文默认注入的请求内单例的,而CacheBackService要注册成全局单例的 // 由于二者的生命周期不同，所以不能相互注入调用,这里手动new一个EF上下文 var optionsBuilder = new DbContextOptionsBuilder<DbHelperContext>(); optionsBuilder.UseSqlServer(_Configuration[string.Join(":", new string[] { "DBConnection", "ConnectionStrings", "Dbconn" })]); DbHelperContext _dbHelper = new DbHelperContext(optionsBuilder.Options); var data= await _dbHelper.SeckillProduct.Where(u => u.id == "21e86c6cc32b4e7bb80f96c98e4e7996").FirstOrDefaultAsync(); _redis.StringSet($"{data.productId}-sCount", data.productStockNum); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
RCustomerService.cs：
public class RCustomerService : BackgroundService { private readonly IDatabase _redis; private readonly IConfiguration _Configuration; public RCustomerService(DBHelper.RedisHelper redisHelper, IConfiguration Configuration) { _redis = redisHelper.GetDatabase(); _Configuration = Configuration; } protected override Task ExecuteAsync(CancellationToken stoppingToken) { // EFCore的上下文默认注入的请求内单例的,而CacheBackService要注册成全局单例的 // 由于二者的生命周期不同，所以不能相互注入调用,这里手动new一个EF上下文 var optionsBuilder = new DbContextOptionsBuilder<DbHelperContext>(); optionsBuilder.UseSqlServer(_Configuration[string.Join(":", new string[] { "DBConnection", "ConnectionStrings", "Dbconn" })]); return Task.Run(() => { Console.WriteLine("下面开始执行消费业务"); while (true) { try { var datalist = _redis.ListRightPop("21e86c6cc32b4e7bb80f96c98e4e7988"); var data = datalist.ToString(); if (!string.IsNullOrEmpty(data)) { using (DbHelperContext _dbHelper = new DbHelperContext(optionsBuilder.Options)) { List<string> tempData = data.Split('-').ToList(); var sArctile = _dbHelper.SeckillProduct.Where(u => u.id == "21e86c6cc32b4e7bb80f96c98e4e7996").FirstOrDefault(); sArctile.productStockNum = sArctile.productStockNum - 1; _dbHelper.Update(sArctile); //2. 插入订单信息 Models.Order tOrder = new Models.Order(); tOrder.id = Guid.NewGuid().ToString("N"); tOrder.userId = tempData[0]; tOrder.orderNum = Guid.NewGuid().ToString("N"); tOrder.productId = tempData[1]; tOrder.orderTotal = sArctile.productPrice; tOrder.addTime = DateTime.Now; tOrder.orderStatus = 0; tOrder.orderPhone = "1565555555"; tOrder.orderAddress = "test"; tOrder.delFlag = 0; _dbHelper.Add<Models.Order>(tOrder); int count = _dbHelper.SaveChanges(); _dbHelper.Entry<SeckillProduct>(sArctile).State = EntityState.Detached; Console.WriteLine($"执行成功，条数为:{count},当前库存为：{ sArctile.productStockNum}"); } } else { Console.WriteLine("暂时没有订单信息，休息一下2"); Thread.Sleep(1000); } } catch (Exception ex) { throw new Exception(ex.Message); } } }, stoppingToken); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
控制器引用RedisHelper帮助类
private static readonly object objlock = new object();
private readonly ILogger<GrabbingOrdersController> logger;
private readonly DbHelperContext dbHelper;
private readonly IDatabase _redis;
public GrabbingOrdersController(ILogger<GrabbingOrdersController> logger, DbHelperContext dbHelper, DBHelper.RedisHelper redisHelper)
{
this.logger = logger;
this.dbHelper = dbHelper;
_redis = redisHelper.GetDatabase();
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
控制器新建SetOrderRedis
[HttpGet] [Route("[action]")] public string SetOrderRedis(string userId, string proId, string requestId = "125643") { try { var count = (int)_redis.StringDecrement($"{proId}-sCount", 1); if (count >= 0) { var orderNum = Guid.NewGuid().ToString("N"); _redis.ListLeftPush(userId, $"{userId}-{proId}-{orderNum}"); return "下单成功"; } else { return "卖完了"; } } catch (Exception ex) { throw new Exception(ex.Message); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
最后到Startup.cs中注册后台任务
//注册后台任务
services.AddHostedService<RCacheBackService>();
services.AddHostedService<RCustomerService>();
services.AddHostedService<CacheBackService>();
services.AddHostedService<CustomerService>();
- 1
- 2
- 3
- 4
- 5
测试结果：
扩展：对产品进行单品限流、购买次数的限制和方法幂等进行限制（以下方法就不进行测试了）
代码加在SetOrderRedis里
单品限流：指定时间同一个商品限制请求次数
int tLimits = 100; //限制请求数量
int tSeconds = 1; //限制秒数
string prokey = $"protLimits{proId}";
long procount = (long)_redis.StringIncrement(prokey, 1);
if (procount > tLimits)
{
throw new Exception($"超出够买限制,{tSeconds}秒内只能请求{tLimits}次");
//return ($"超出够买限制,{tSeconds}秒内只能请求{tLimits}次");
}
else if (procount == 1)
{
_redis.KeyExpire(prokey, TimeSpan.FromSeconds(tSeconds));
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
购买次数的限制：指定时间每个用户对同一个商品购买限制
int tLimits2 = 3; //限制请求数量
int tSeconds2 = 10; //抢购活动持续时间
string useprokey = $"protLimits_{userId}_{proId}";
long useprocount = (long)_redis.StringIncrement(useprokey, 1);
if (useprocount > tLimits2)
{
throw new Exception($"每个用户只能请求{tLimits2}次");
}
else if (useprocount == 1)
{
_redis.KeyExpire(useprokey, TimeSpan.FromMinutes(tSeconds2));
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
方法幂等： 一个固定id指定时间内只能生成1条订单（防止防止网络延迟多次提交问题及快速点击提交按钮导致同时间多次提交请求，id由前端生成，实时更新）
int tLimits3 = 1;
int tSeconds3 = 1; //限制秒数，限制时间不可过大，防止用户通过返回导致id相同
string useprokey = $"protLimits_{requestId}_{userId}_{proId}";
long useprocount = (long)_redis.StringIncrement(useprokey, 1);
if (useprocount > tLimits3)
{
throw new Exception($"提交订单过快，请重新下单");
}
else if (useprocount == 1)
{
_redis.KeyExpire(useprokey, TimeSpan.FromSeconds(tSeconds3));
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
RCustomerService.cs实时刷新只对一条数据进行处理，这样在执行添加订单和扣除库存时是很慢的。
如何优化：
1.现在我们把Redis的数据进行统一处理，但Redis订单数据达到一定长度或者一个固定时间后去处理这些数据，EFCore.BulkExtensions组件批量处理
需要到添加EFCore.BulkExtensions包
2.把EF上下文放到内部,每次执行业务的时候都using一下,解决上下文不释放的问题,因为每次上下文都是一个新的。
RCustomerService.cs
protected override Task ExecuteAsync(CancellationToken stoppingToken) { // EFCore的上下文默认注入的请求内单例的,而CacheBackService要注册成全局单例的 // 由于二者的生命周期不同，所以不能相互注入调用,这里手动new一个EF上下文 var optionsBuilder = new DbContextOptionsBuilder<DbHelperContext>(); optionsBuilder.UseSqlServer(_Configuration[string.Join(":", new string[] { "DBConnection", "ConnectionStrings", "Dbconn" })]); #region 原代码，一次处理一条数据，效率低 //return Task.Run(() => //{ // Console.WriteLine("下面开始执行消费业务"); // while (true) // { // try // { // var datalist = _redis.ListRightPop("21e86c6cc32b4e7bb80f96c98e4e7988"); // var data = datalist.ToString(); // if (!string.IsNullOrEmpty(data)) // { // using (DbHelperContext _dbHelper = new DbHelperContext(optionsBuilder.Options)) // { // List<string> tempData = data.Split('-').ToList(); // var sArctile = _dbHelper.SeckillProduct.Where(u => u.id == "21e86c6cc32b4e7bb80f96c98e4e7996").FirstOrDefault(); // sArctile.productStockNum = sArctile.productStockNum - 1; // _dbHelper.Update(sArctile); // //2. 插入订单信息 // Models.Order tOrder = new Models.Order(); // tOrder.id = Guid.NewGuid().ToString("N"); // tOrder.userId = tempData[0]; // tOrder.orderNum = Guid.NewGuid().ToString("N"); // tOrder.productId = tempData[1]; // tOrder.orderTotal = sArctile.productPrice; // tOrder.addTime = DateTime.Now; // tOrder.orderStatus = 0; // tOrder.orderPhone = "1565555555"; // tOrder.orderAddress = "test"; // tOrder.delFlag = 0; // _dbHelper.Add<Models.Order>(tOrder); // int count = _dbHelper.SaveChanges(); // _dbHelper.Entry<SeckillProduct>(sArctile).State = EntityState.Detached; // Console.WriteLine($"执行成功，条数为:{count},当前库存为：{ sArctile.productStockNum}"); // } // } // else // { // Console.WriteLine("暂时没有订单信息，休息一下2"); // Thread.Sleep(1000); // } // } // catch (Exception ex) // { // throw new Exception(ex.Message); // } // } //}, stoppingToken); #endregion #region 当从队列中获取的数量达到200条的时候提交 或者 2s提交一次(但必须有数据) 使用EFCore.BulkExtensions 批量处理数据 优化更新和新增 Console.WriteLine("新的进程，休息一下"); return Task.Run(() => { Console.WriteLine("新的进程2"); //临时存储从队列中取出来的信息 List<string> orderlist = new List<string>(); Stopwatch stopwatch = new Stopwatch(); stopwatch.Start(); while (true) { try { var data = _redis.ListRightPop("21e86c6cc32b4e7bb80f96c98e4e0004").ToString(); if (!string.IsNullOrEmpty(data)) { orderlist.Add(data); } else { Console.WriteLine("暂时没有订单信息，休息一下"); Thread.Sleep(1000); } if (orderlist.Count >= 10 || stopwatch.ElapsedMilliseconds > 2000) { if (orderlist.Count > 0) { using (DbHelperContext _dbHelper = new DbHelperContext(optionsBuilder.Options)) { using (var transaction = _dbHelper.Database.BeginTransaction()) { try { var product = _dbHelper.SeckillProduct.Where(x => x.id == "21e86c6cc32b4e7bb80f96c98e4e7996").FirstOrDefault(); int count = product.productStockNum - orderlist.Count; int count2 = _dbHelper.SeckillProduct.Where(u => u.id.StartsWith("21e86c6cc32b4e7bb80f96c98e4e7996")) .BatchUpdate(new SeckillProduct() { productStockNum = count }); //product.productStockNum = product.productStockNum - orderlist.Count; //_dbHelper.Update(product); List<Models.Order> orders = new List<Models.Order>(); foreach (var item in orderlist) { List<string> tempData = item.Split('-').ToList(); //2. 插入订单信息 Models.Order tOrder = new Models.Order(); tOrder.id = Guid.NewGuid().ToString("N"); tOrder.userId = tempData[0]; tOrder.orderNum = Guid.NewGuid().ToString("N"); tOrder.productId = tempData[1]; tOrder.orderTotal = product.productPrice; tOrder.addTime = DateTime.Now; tOrder.orderStatus = 0; tOrder.orderPhone = "1565555555"; tOrder.orderAddress = "test"; tOrder.delFlag = 0; orders.Add(tOrder); } _dbHelper.BulkInsert(orders); transaction.Commit(); Console.WriteLine($"消费成功"); orderlist.Clear(); } catch (Exception ex) { Console.WriteLine($"消费失败:{ex.Message}"); } } } } stopwatch.Restart(); //stopwatch.Stop(); } } catch (Exception ex) { Console.WriteLine($"执行失败：{ex.Message}"); } } }, stoppingToken); #endregion }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
- 115
- 116
- 117
- 118
- 119
- 120
- 121
- 122
- 123
- 124
- 125
- 126
- 127
- 128
- 129
- 130
- 131
- 132
- 133
- 134
- 135
- 136
- 137
- 138
- 139
- 140
- 141
- 142
- 143
- 144
- 145
- 146
- 147
- 148
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/羊村懒王/article/detail/84244
推荐阅读
相关标签
