诺禾-数据库操作优化

反响式编程在客户端编程当中的应用相当普遍，而当前在效劳端中的应用相对被提及较少。本篇将引见如何在效劳端编程中应用响应时编程来改良数据库操作的性能。

开篇就是结论
应用 System.Reactive 配合 TaskCompelteSource ，能够将分散的单次数据库插入恳求兼并会一个批量插入的恳求。在确保正确性的前提下，完成数据库插入性能的优化。

假如读者曾经理解了如何操作，那么剩下的内容就不需求再看了。

预设条件
如今，我们假定存在这样一个 Repository 接口来表示一次数据库的插入操作。

csharp
namespace Newbe.RxWorld.DatabaseRepository
{
public interface IDatabaseRepository
{
///

    /// Insert one item and return total count of data in database/// /// /// Task<int> InsertData(int item);
}

}
接下来，我们在不改动该接口签名的前提下，体验一下不同的完成带来的性能区别。

根底版本
首先是根底版本，采用的是最为常规的单次数据库INSERT操作来完成数据的插入。本示例采用的是SQLite作为演示数据库，便当读者自行实验。

csharp
namespace Newbe.RxWorld.DatabaseRepository.Impl
{
public class NormalDatabaseRepository : IDatabaseRepository
{
private readonly IDatabase _database;
public NormalDatabaseRepository(
IDatabase database)
{
_database = database;
}
public Task InsertData(int item)
{
return _database.InsertOne(item);
}
}
}
常规操作。其中_database.InsertOne(item)的详细完成就是调用了一次INSERT。

根底版本在同时插入小于20次时根本上能够较快的完成。但是假如数量级增加，例如需求同时插入一万条数据库，将会破费约20秒钟，存在很大的优化空间。

TaskCompelteSource
TaskCompelteSource 是 TPL 库中一个能够生成一个可操作 Task 的类型。关于 TaskCompelteSource 不太熟习的读者能够经过该实例代码理解。

此处也简单解释一下该对象的作用，以便读者能够继续阅读。

关于熟习 javascript 的朋友，能够以为 TaskCompelteSource 相当于 Promise 对象。也能够相当于 jQuery 当中的 $.Deferred 。

假如都不理解的朋友，能够听一下笔者吃麻辣烫时想到的生活化例子。

吃麻辣烫技术解释
吃麻辣烫之前，需求先用盘子夹菜。结构参数
夹好菜之后，拿到结账处去结账调用办法
收银员结账终了之后，会得到一个叫餐牌，会响铃的那种得到一个 Task 返回值
拿着菜牌找了一个位子坐下，玩手机等餐正在 await 这个 Task ，CPU转而处置其他事情
餐牌响了，去取餐，吃起来 Task 完成，await 节数，继续执行下一行代码
那么 TaskCompelteSource 在哪儿呢？

首先，依据上面的例子，在餐牌响的时分，我们才会去取餐。那么餐牌什么时分才会响呢？当然是效劳员手动按了一个在柜台的手动开关才触发了这个响铃。

那么，柜台的这个开关，能够被技术解释为 TaskCompelteSource 。

餐台开关能够控制餐牌的响铃。同样， TaskCompelteSource 就是一种能够控制 Task 的状态的对象。

处理思绪
有了前面对 TaskCompelteSource 的理解，那么接下来就能够处理文章开头的问题了。思绪如下：

当调用 InsertData 时，能够创立一个 TaskCompelteSource 以及 item 的元组。为了便当阐明，我们将这个元组命名为BatchItem。

将 BatchItem 的 TaskCompelteSource 对应的 Task 返回进来。

调用 InsertData 的代码会 await 返回的 Task，因而只需不操作 TaskCompelteSource ，调用者就一会不断等候。

然后，另外启动一个线程，定时将 BatchItem 队列消费掉。

这样就完成了单次插入变为批量插入的操作。

笔者可能解释的不太分明，不过以下一切版本的代码均基于以上思绪。读者能够分离文字和代码停止了解。

ConcurrentQueue 版本
基于以上的思绪，我们采用 ConcurrentQueue 作为 BatchItem 队列停止完成，代码如下（代码很多，不用纠结，由于下面还有更简单的）：

csharp
namespace Newbe.RxWorld.DatabaseRepository.Impl
{
public class ConcurrentQueueDatabaseRepository : IDatabaseRepository
{
private readonly ITestOutputHelper _testOutputHelper;
private readonly IDatabase _database;
private readonly ConcurrentQueue _queue;
// ReSharper disable once PrivateFieldCanBeConvertedToLocalVariable
private readonly Task _batchInsertDataTask;
public ConcurrentQueueDatabaseRepository(
ITestOutputHelper testOutputHelper,
IDatabase database)
{
_testOutputHelper = testOutputHelper;
_database = database;
_queue = new ConcurrentQueue();
// 启动一个 Task 消费队列中的 BatchItem
_batchInsertDataTask = Task.Factory.StartNew(RunBatchInsert, TaskCreationOptions.LongRunning);
_batchInsertDataTask.ConfigureAwait(false);
}
public Task InsertData(int item)
{
// 生成 BatchItem ，将对象放入队列。返回 Task 进来
var taskCompletionSource = new TaskCompletionSource();
_queue.Enqueue(new BatchItem
{
Item = item,
TaskCompletionSource = taskCompletionSource
});
return taskCompletionSource.Task;
}
// 从队列中不时获取 BatchItem ，并且一批一批插入数据库，更新 TaskCompletionSource 的状态
private void RunBatchInsert()
{
foreach (var batchItems in GetBatches())
{
try
{
BatchInsertData(batchItems).Wait();
}
catch (Exception e)
{
_testOutputHelper.WriteLine($“there is an error : {e}”);
}
}
IEnumerable<IList> GetBatches()
{
var sleepTime = TimeSpan.FromMilliseconds(50);
while (true)
{
const int maxCount = 100;
var oneBatchItems = GetWaitingItems()
.Take(maxCount)
.ToList();
if (oneBatchItems.Any())
{
yield return oneBatchItems;
}
else
{
Thread.Sleep(sleepTime);
}
}
IEnumerable GetWaitingItems()
{
while (_queue.TryDequeue(out var item))
{
yield return item;
}
}
}
}
private async Task BatchInsertData(IEnumerable items)
{
var batchItems = items as BatchItem[] ?? items.ToArray();
try
{
// 调用数据库的批量插入操作
var totalCount = await _database.InsertMany(batchItems.Select(x => x.Item));
foreach (var batchItem in batchItems)
{
batchItem.TaskCompletionSource.SetResult(totalCount);
}
}
catch (Exception e)
{
foreach (var batchItem in batchItems)
{
batchItem.TaskCompletionSource.SetException(e);
}
throw;
}
}
private struct BatchItem
{
public TaskCompletionSource TaskCompletionSource { get; set; }
public int Item { get; set; }
}
}
}
以上代码中运用了较多的 Local Function 和 IEnumerable 的特性，不理解的读者能够点击此处停止理解。

正片开端！
接下来我们运用 System.Reactive 来改造上面较为复杂的 ConcurrentQueue 版本。如下：

csharp
namespace Newbe.RxWorld.DatabaseRepository.Impl
{
public class AutoBatchDatabaseRepository : IDatabaseRepository
{
private readonly ITestOutputHelper _testOutputHelper;
private readonly IDatabase _database;
private readonly Subject _subject;
public AutoBatchDatabaseRepository(
ITestOutputHelper testOutputHelper,
IDatabase database)
{
_testOutputHelper = testOutputHelper;
_database = database;
_subject = new Subject();
// 将恳求停止分组，每50毫秒一组或者每100个一组
_subject.Buffer(TimeSpan.FromMilliseconds(50), 100)
.Where(x => x.Count > 0)
// 将每组数据调用批量插入，写入数据库
.Select(list => Observable.FromAsync(() => BatchInsertData(list)))
.Concat()
.Subscribe();
}
// 这里和前面比照没有变化
public Task InsertData(int item)
{
var taskCompletionSource = new TaskCompletionSource();
_subject.OnNext(new BatchItem
{
Item = item,
TaskCompletionSource = taskCompletionSource
});
return taskCompletionSource.Task;
}
// 这段和前面也完整一样，没有变化
private async Task BatchInsertData(IEnumerable items)
{
var batchItems = items as BatchItem[] ?? items.ToArray();
try
{
var totalCount = await _database.InsertMany(batchItems.Select(x => x.Item));
foreach (var batchItem in batchItems)
{
batchItem.TaskCompletionSource.SetResult(totalCount);
}
}
catch (Exception e)
{
foreach (var batchItem in batchItems)
{
batchItem.TaskCompletionSource.SetException(e);
}
throw;
}
}
private struct BatchItem
{
public TaskCompletionSource TaskCompletionSource { get; set; }
public int Item { get; set; }
}
}
}
代码减少了 50 行，主要缘由就是运用 System.Reactive 中提供的很强力的 Buffer 办法完成了 ConcurrentQueue 版本中的复杂的逻辑完成。

教师，能够更给力一点吗？
我们，能够“略微”优化一下代码，将 Buffer 以及相关的逻辑独立于“数据库插入”这个业务逻辑。那么我们就会得到一个愈加简单的版本：

csharp
namespace Newbe.RxWorld.DatabaseRepository.Impl
{
public class FinalDatabaseRepository : IDatabaseRepository
{
private readonly IBatchOperator<int, int> _batchOperator;
public FinalDatabaseRepository(
IDatabase database)
{
var options = new BatchOperatorOptions<int, int>
{
BufferTime = TimeSpan.FromMilliseconds(50),
BufferCount = 100,
DoManyFunc = database.InsertMany,
};
_batchOperator = new BatchOperator<int, int>(options);
}
public Task InsertData(int item)
{
return _batchOperator.CreateTask(item);
}
}
}
其中 IBatchOperator 等代码，读者能够到代码库中停止查看，此处就不在陈列了。

诺禾-数据库操作优化相关推荐

数据库建表原则，SQL数据库建表前期优化，SQL数据库操作优化，数据库命名规范...
2019独角兽企业重金招聘Python工程师标准>>> 关键字: 数据库建表原则 ·1. 原始单据与实体之间的关系可以是一对一.一对多.多对多的关系.在一般情况下,它们是一对一的关 ...
SQL数据库操作优化
1.应尽量避免在 where 子句中对字段进行 null 值判断,否则将导致引擎放弃使用索引而进行全表扫描,如: select id from t where num is null 可以在num上设 ...
红橙Darren视频笔记数据库操作优化数据查询（数据库操作）中
上一节仅仅是做到了有这个功能,这次我们对上一次的代码进行优化主要有两个方面可以优化 1.利用数据库事务进行优化我们作如下修改: IDaoSupport增加批量插入接口 // 批量插入数据publi ...
beego mysql按时间排序_beego的数据库操作优化
1.背景描述用beego的ORM框架对sqllite进行库操作.项目中配置表存放在sqlite数据中,这样就存对每个表需要进行增删改查的操作.若按照beego提供方式对每一张表去实现增删改查的操作, ...
艾伟：基于.NET平台的Windows编程实战（四）—— 数据库操作类的编写
本系列文章导航基于.NET平台的Windows编程实战(一)--前言基于.NET平台的Windows编程实战(二)-- 需求分析与数据库设计基于.NET平台的Windows编程实战(四)-- 数 ...
利用对象池优化数据库操作
简介:这是利用对象池优化数据库操作的详细页面,介绍了和asp.net,.Net,创建,对象池,示例有关的知识,要查看更多相关信息,请点击此处说到对象池,大家都不陌生.很多人都实现过,网上的代码也满天 ...
数据库-优化-数据库系统配置优化-操作系统优化
数据库系统配置优化定义数据库是基于操作系统的,目前大多数MySQL都是安装在linux系统之上,所以对于操作系统的一些参数配置也会影响到MySQL的性能,下面就列出一些常用的系统配置. 优化配置参 ...
诺禾-实验技巧之WB篇
关于蛋白质免疫印迹(Western Blot,即 WB),想必做蛋白实验的小同伴们都不会生疏,作为免疫学中最常用的一种实验办法,其根本原理是经过特异性抗体对凝胶电泳处置过的细胞或生物组织样品停止着色. ...
利用Java存储过程简化数据库操作
利用Java存储过程沟通SQL.XML.Java.J2EE和Web服务. 存储过程(stored procedure)允许将运行于数据库层中的持久性逻辑与运行于中间层中的商务逻辑有效地分离开来. ...

诺禾-数据库操作优化

诺禾-数据库操作优化相关推荐

最新文章

热门文章