tk/FreeSql
1
0
Fork 0
mirror of https://github.com/nsnail/FreeSql.git synced 2025-04-22 02:32:50 +08:00
Code Issues Packages Projects Releases Wiki Activity

- 修复 Clickhouse 连接池使用问题；#646 #968 #969 #943

Browse Source
This commit is contained in:
2881099 2022-02-23 11:55:39 +08:00
parent 9bde42a566
commit 1e55cf4e25
2 changed files with 68 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
FreeSql.DbContext/FreeSql.DbContext.xml
View File

@ -538,14 +538,5 @@
<param name="that"></param> <param name="that"></param>
<returns></returns> <returns></returns>
</member> </member>
<member name="M:Microsoft.Extensions.DependencyInjection.FreeSqlRepositoryDependencyInjection.AddFreeRepository(Microsoft.Extensions.DependencyInjection.IServiceCollection,System.Action{FreeSql.FluentDataFilter},System.Reflection.Assembly[])">
<summary>
批量注入 Repository，可以参考代码自行调整
</summary>
<param name="services"></param>
<param name="globalDataFilter"></param>
<param name="assemblies"></param>
<returns></returns>
</member>
</members> </members>
</doc> </doc>

76
Providers/FreeSql.Provider.ClickHouse/Curd/ClickHouseInsert.cs
View File

@ -50,20 +50,23 @@ namespace FreeSql.ClickHouse.Curd
{ {
var affrows = 0; var affrows = 0;
Exception exception = null; Exception exception = null;
Aop.CurdBeforeEventArgs before=null; Aop.CurdBeforeEventArgs before = null;
if (_source.Count>1) if (_source.Count > 1)
{ {
try try
{ {
before = new Aop.CurdBeforeEventArgs(_table.Type, _table, Aop.CurdType.Insert, null, _params); before = new Aop.CurdBeforeEventArgs(_table.Type, _table, Aop.CurdType.Insert, null, _params);
_orm.Aop.CurdBeforeHandler?.Invoke(this, before); _orm.Aop.CurdBeforeHandler?.Invoke(this, before);
using var bulkCopyInterface = new ClickHouseBulkCopy(_orm.Ado.MasterPool.Get().Value as ClickHouseConnection) using (var conn = _orm.Ado.MasterPool.Get())
{ {
DestinationTableName = _table.DbName, using var bulkCopyInterface = new ClickHouseBulkCopy(conn.Value as ClickHouseConnection)
BatchSize = _source.Count {
}; DestinationTableName = _table.DbName,
var data=ToDataTable(); BatchSize = _source.Count
bulkCopyInterface.WriteToServerAsync(data, default).Wait(); };
var data = ToDataTable();
bulkCopyInterface.WriteToServerAsync(data, default).Wait();
}
return affrows; return affrows;
} }
catch (Exception ex) catch (Exception ex)
@ -173,6 +176,63 @@ namespace FreeSql.ClickHouse.Curd
public override Task<long> ExecuteIdentityAsync(CancellationToken cancellationToken = default) => SplitExecuteIdentityAsync(_batchValuesLimit > 0 ? _batchValuesLimit : int.MaxValue, _batchValuesLimit > 0 ? _batchValuesLimit : int.MaxValue, cancellationToken); public override Task<long> ExecuteIdentityAsync(CancellationToken cancellationToken = default) => SplitExecuteIdentityAsync(_batchValuesLimit > 0 ? _batchValuesLimit : int.MaxValue, _batchValuesLimit > 0 ? _batchValuesLimit : int.MaxValue, cancellationToken);
public override Task<List<T1>> ExecuteInsertedAsync(CancellationToken cancellationToken = default) => SplitExecuteInsertedAsync(_batchValuesLimit > 0 ? _batchValuesLimit : int.MaxValue, _batchValuesLimit > 0 ? _batchValuesLimit : int.MaxValue, cancellationToken); public override Task<List<T1>> ExecuteInsertedAsync(CancellationToken cancellationToken = default) => SplitExecuteInsertedAsync(_batchValuesLimit > 0 ? _batchValuesLimit : int.MaxValue, _batchValuesLimit > 0 ? _batchValuesLimit : int.MaxValue, cancellationToken);
async protected override Task<int> RawExecuteAffrowsAsync(CancellationToken cancellationToken = default)
{
var affrows = 0;
Exception exception = null;
Aop.CurdBeforeEventArgs before = null;
if (_source.Count > 1)
{
try
{
before = new Aop.CurdBeforeEventArgs(_table.Type, _table, Aop.CurdType.Insert, null, _params);
_orm.Aop.CurdBeforeHandler?.Invoke(this, before);
using (var conn = await _orm.Ado.MasterPool.GetAsync())
{
using var bulkCopyInterface = new ClickHouseBulkCopy(conn.Value as ClickHouseConnection)
{
DestinationTableName = _table.DbName,
BatchSize = _source.Count
};
var data = ToDataTable();
await bulkCopyInterface.WriteToServerAsync(data, default);
}
return affrows;
}
catch (Exception ex)
{
exception = ex;
throw ex;
}
finally
{
var after = new Aop.CurdAfterEventArgs(before, exception, affrows);
_orm.Aop.CurdAfterHandler?.Invoke(this, after);
}
}
else
{
var sql = this.ToSql();
before = new Aop.CurdBeforeEventArgs(_table.Type, _table, Aop.CurdType.Insert, sql, _params);
_orm.Aop.CurdBeforeHandler?.Invoke(this, before);
try
{
affrows = await _orm.Ado.ExecuteNonQueryAsync(_connection, null, CommandType.Text, sql, _commandTimeout, _params);
}
catch (Exception ex)
{
exception = ex;
throw ex;
}
finally
{
var after = new Aop.CurdAfterEventArgs(before, exception, affrows);
_orm.Aop.CurdAfterHandler?.Invoke(this, after);
}
return affrows;
}
}
async protected override Task<long> RawExecuteIdentityAsync(CancellationToken cancellationToken = default) async protected override Task<long> RawExecuteIdentityAsync(CancellationToken cancellationToken = default)
{ {
//var sql = this.ToSql(); //var sql = this.ToSql();