AzureCosmosDBV3Repository.cs 44 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026
  1. using Microsoft.Azure.Cosmos;
  2. using Microsoft.Azure.Cosmos.Linq;
  3. using System;
  4. using System.Collections.Concurrent;
  5. using System.Collections.Generic;
  6. using System.Diagnostics;
  7. using System.IO;
  8. using System.Linq;
  9. using System.Linq.Expressions;
  10. using System.Net;
  11. using System.Reflection;
  12. using System.Text;
  13. using System.Text.Json;
  14. using System.Threading;
  15. using System.Threading.Tasks;
  16. using TEAMModelOS.SDK.Context.Attributes.Azure;
  17. using TEAMModelOS.SDK.Context.Exception;
  18. using TEAMModelOS.SDK.Helper.Common.CollectionHelper;
  19. using TEAMModelOS.SDK.Helper.Common.ReflectorExtensions;
  20. using TEAMModelOS.SDK.Module.AzureCosmosDB.Configuration;
  21. namespace TEAMModelOS.SDK.Module.AzureCosmosDBV3
  22. {
  23. public class CosmosModelInfo
  24. {
  25. public Container container { get; set; }
  26. public bool cache { get; set; }
  27. public bool monitor { get; set; } = false;
  28. public Type type { get; set; }
  29. }
  30. public class AzureCosmosDBV3Repository : IAzureCosmosDBV3Repository
  31. {
  32. private CosmosClient CosmosClient { get; set; }
  33. /// <summary>
  34. /// 线程安全的dict类型
  35. /// </summary>
  36. private Dictionary<string, CosmosModelInfo> DocumentCollectionDict { get; set; } = new Dictionary<string, CosmosModelInfo>();
  37. private string DatabaseId { get; set; }
  38. private int CollectionThroughput { get; set; }
  39. private Database database { get; set; }
  40. int pageSize = 200;
  41. private const string CacheCosmosPrefix = "cosmos:";
  42. private string[] ScanModel { get; set; }
  43. private const int timeoutSeconds = 86400;
  44. private string leaseId = "AleaseContainer";
  45. public AzureCosmosDBV3Repository(AzureCosmosDBOptions options, CosmosSerializer cosmosSerializer)
  46. {
  47. try
  48. {
  49. if (!string.IsNullOrEmpty(options.ConnectionString))
  50. {
  51. CosmosClient = CosmosDBV3ClientSingleton.getInstance(options.ConnectionString, options.ConnectionKey, cosmosSerializer).GetCosmosDBClient();
  52. }
  53. else
  54. {
  55. throw new BizException("请设置正确的AzureCosmosDB数据库配置信息!");
  56. }
  57. DatabaseId = options.Database;
  58. CollectionThroughput = options.CollectionThroughput;
  59. ScanModel = options.ScanModel;
  60. // InitializeDatabase().GetAwaiter().GetResult();
  61. }
  62. catch (CosmosException e)
  63. {
  64. // Dispose(true);
  65. throw new BizException(e.Message, 500, e.StackTrace);
  66. }
  67. }
  68. public AzureCosmosDBV3Repository(AzureCosmosDBOptions options)
  69. {
  70. try
  71. {
  72. if (!string.IsNullOrEmpty(options.ConnectionString))
  73. {
  74. CosmosClient = CosmosDBV3ClientSingleton.getInstance(options.ConnectionString, options.ConnectionKey, null).GetCosmosDBClient();
  75. }
  76. else
  77. {
  78. throw new BizException("请设置正确的AzureCosmosDB数据库配置信息!");
  79. }
  80. DatabaseId = options.Database;
  81. CollectionThroughput = options.CollectionThroughput;
  82. ScanModel = options.ScanModel;
  83. // InitializeDatabase().GetAwaiter().GetResult();
  84. }
  85. catch (CosmosException e)
  86. {
  87. // Dispose(true);
  88. throw new BizException(e.Message, 500, e.StackTrace);
  89. }
  90. }
  91. public async Task<Dictionary<string, CosmosModelInfo>> InitializeDatabase()
  92. {
  93. try
  94. {
  95. database = await CosmosClient.CreateDatabaseIfNotExistsAsync(DatabaseId, CollectionThroughput);
  96. FeedIterator<ContainerProperties> resultSetIterator = database.GetContainerQueryIterator<ContainerProperties>();
  97. while (resultSetIterator.HasMoreResults)
  98. {
  99. foreach (ContainerProperties container in await resultSetIterator.ReadNextAsync())
  100. {
  101. DocumentCollectionDict.TryAdd(container.Id, new CosmosModelInfo { container = database.GetContainer(container.Id), cache = false, monitor = false });
  102. }
  103. }
  104. bool isMonitor = false;
  105. //获取数据库所有的表
  106. List<Type> types = ReflectorExtensions.GetAllTypeAsAttribute<CosmosDBAttribute>(ScanModel);
  107. foreach (Type type in types)
  108. {
  109. string PartitionKey = GetPartitionKey(type);
  110. string CollectionName = "";
  111. int RU = 0;
  112. bool cache = false;
  113. bool monitor = false;
  114. IEnumerable<CosmosDBAttribute> attributes = type.GetCustomAttributes<CosmosDBAttribute>(true);
  115. if (!string.IsNullOrEmpty(attributes.First<CosmosDBAttribute>().Name))
  116. {
  117. CollectionName = attributes.First<CosmosDBAttribute>().Name;
  118. }
  119. else
  120. {
  121. CollectionName = type.Name;
  122. }
  123. if (attributes.First<CosmosDBAttribute>().Cache)
  124. {
  125. cache = attributes.First<CosmosDBAttribute>().Cache;
  126. }
  127. if (attributes.First<CosmosDBAttribute>().Monitor)
  128. {
  129. monitor = attributes.First<CosmosDBAttribute>().Monitor;
  130. if (monitor)
  131. {
  132. isMonitor = true;
  133. }
  134. }
  135. //else
  136. //{
  137. // cache = false;
  138. //}
  139. if (attributes.First<CosmosDBAttribute>().RU > 400)
  140. {
  141. RU = attributes.First<CosmosDBAttribute>().RU;
  142. }
  143. else
  144. {
  145. RU = CollectionThroughput;
  146. }
  147. //如果表存在于数据则检查RU是否变动,如果不存在则执行创建DocumentCollection
  148. if (DocumentCollectionDict.TryGetValue(CollectionName, out CosmosModelInfo cosmosModelInfo))
  149. { //更新RU
  150. cosmosModelInfo.cache = cache;
  151. Container container = CosmosClient.GetDatabase(DatabaseId).GetContainer(cosmosModelInfo.container.Id);
  152. int? throughputResponse = await container.ReadThroughputAsync();
  153. if (throughputResponse < RU)
  154. {
  155. await CosmosClient.GetDatabase(DatabaseId).GetContainer(cosmosModelInfo.container.Id).ReplaceThroughputAsync(RU);
  156. }
  157. DocumentCollectionDict[CollectionName] = new CosmosModelInfo { container = container, cache = cache, monitor = monitor, type = type };
  158. }
  159. else
  160. {
  161. ContainerProperties containerProperties = new ContainerProperties { Id = CollectionName };
  162. if (!string.IsNullOrEmpty(PartitionKey))
  163. {
  164. containerProperties.PartitionKeyPath = "/" + PartitionKey;
  165. }
  166. if (RU > CollectionThroughput)
  167. {
  168. CollectionThroughput = RU;
  169. }
  170. Container containerWithConsistentIndexing = await database.CreateContainerIfNotExistsAsync(containerProperties, throughput: CollectionThroughput);
  171. DocumentCollectionDict.TryAdd(CollectionName, new CosmosModelInfo { container = containerWithConsistentIndexing, cache = cache, monitor = monitor, type = type });
  172. }
  173. }
  174. if (isMonitor)
  175. {
  176. ContainerProperties leaseProperties = new ContainerProperties { Id = leaseId, PartitionKeyPath = "/id" };
  177. Container leaseContainer = await database.CreateContainerIfNotExistsAsync(leaseProperties, throughput: CollectionThroughput);
  178. DocumentCollectionDict.TryAdd(leaseId, new CosmosModelInfo { container = leaseContainer, cache = false, monitor = false });
  179. }
  180. return DocumentCollectionDict;
  181. }
  182. catch (CosmosException e)
  183. {
  184. throw new BizException(e.Message, 500, e.StackTrace);
  185. }
  186. }
  187. private string GetPartitionKey<T>()
  188. {
  189. Type type = typeof(T);
  190. return GetPartitionKey(type);
  191. }
  192. private string GetPartitionKey(Type type)
  193. {
  194. PropertyInfo[] properties = type.GetProperties();
  195. List<PropertyInfo> attrProperties = new List<PropertyInfo>();
  196. foreach (PropertyInfo property in properties)
  197. {
  198. if (property.Name.Equals("PartitionKey"))
  199. {
  200. attrProperties.Add(property);
  201. break;
  202. }
  203. object[] attributes = property.GetCustomAttributes(true);
  204. foreach (object attribute in attributes) //2.通过映射,找到成员属性上关联的特性类实例,
  205. {
  206. if (attribute is PartitionKeyAttribute)
  207. {
  208. attrProperties.Add(property);
  209. }
  210. }
  211. }
  212. if (attrProperties.Count <= 0)
  213. {
  214. throw new BizException(type.Name + "has no PartitionKey !");
  215. }
  216. else
  217. {
  218. if (attrProperties.Count == 1)
  219. {
  220. return attrProperties[0].Name;
  221. }
  222. else { throw new BizException("PartitionKey can only be single!"); }
  223. }
  224. }
  225. private async Task<CosmosModelInfo> InitializeCollection<T>()
  226. {
  227. Type type = typeof(T);
  228. string partitionKey = GetPartitionKey<T>();
  229. string CollectionName;
  230. IEnumerable<CosmosDBAttribute> attributes = type.GetCustomAttributes<CosmosDBAttribute>(true);
  231. if (!string.IsNullOrEmpty(attributes.First<CosmosDBAttribute>().Name))
  232. {
  233. CollectionName = attributes.First<CosmosDBAttribute>().Name;
  234. }
  235. else
  236. {
  237. CollectionName = type.Name;
  238. }
  239. return await InitializeCollection(CollectionName, partitionKey);
  240. }
  241. private async Task<CosmosModelInfo> InitializeCollection(string CollectionName, string PartitionKey)
  242. {
  243. /////内存中已经存在这个表则直接返回
  244. if (DocumentCollectionDict.TryGetValue(CollectionName, out CosmosModelInfo cosmosModelInfo))
  245. {
  246. return cosmosModelInfo;
  247. }///如果没有则尝试默认创建
  248. else
  249. {
  250. ContainerProperties containerProperties = new ContainerProperties { Id = CollectionName };
  251. if (!string.IsNullOrEmpty(PartitionKey))
  252. {
  253. containerProperties.PartitionKeyPath = "/" + PartitionKey;
  254. }
  255. Container containerWithConsistentIndexing = await database.CreateContainerIfNotExistsAsync(containerProperties, throughput: CollectionThroughput);
  256. CosmosModelInfo cosmosModel = new CosmosModelInfo { container = containerWithConsistentIndexing, cache = false };
  257. DocumentCollectionDict.TryAdd(CollectionName, cosmosModel);
  258. return cosmosModel;
  259. }
  260. }
  261. public async Task<List<IdPk>> DeleteAll<T>(List<IdPk> ids) where T : ID
  262. {
  263. CosmosModelInfo container = await InitializeCollection<T>();
  264. //string partitionKey = GetPartitionKey<T>();
  265. //await Task.Run(() => Parallel.ForEach(ids, new ParallelOptions { MaxDegreeOfParallelism = 2 }, (item) =>
  266. //{
  267. // Task.WaitAll(DeleteAsync<T>(item.Value, item.Key));
  268. //}));
  269. List<IdPk> idPks = new List<IdPk>();
  270. int pages = (int)Math.Ceiling((double)ids.Count / pageSize);
  271. Stopwatch stopwatch = Stopwatch.StartNew();
  272. for (int i = 0; i < pages; i++)
  273. {
  274. List<IdPk> lists = ids.Skip((i) * pageSize).Take(pageSize).ToList();
  275. List<Task> tasks = new List<Task>(lists.Count);
  276. lists.ForEach(item =>
  277. {
  278. tasks.Add(container.container.DeleteItemStreamAsync(item.id, new PartitionKey(item.pk))
  279. .ContinueWith((Task<ResponseMessage> task) =>
  280. {
  281. using (ResponseMessage response = task.Result)
  282. {
  283. idPks.Add(new IdPk { id = item.id, pk = item.pk.ToString(), StatusCode = response.StatusCode });
  284. // if (!response.IsSuccessStatusCode)
  285. // {
  286. // }
  287. }
  288. }
  289. ));
  290. });
  291. await Task.WhenAll(tasks);
  292. if (container.cache && RedisHelper.Instance != null)
  293. {
  294. lists.ForEach(async x => {
  295. await RedisHelper.HDelAsync(CacheCosmosPrefix + container.container.Id, x.id);
  296. });
  297. }
  298. }
  299. stopwatch.Stop();
  300. return idPks;
  301. }
  302. public async Task<List<IdPk>> DeleteAll<T>(Dictionary<string, object> dict) where T : ID
  303. {
  304. List<T> list = await FindByDict<T>(dict);
  305. return await DeleteAll(list);
  306. }
  307. public async Task<List<IdPk>> DeleteAll<T>(List<T> enyites) where T : ID
  308. {
  309. List<IdPk> idPks = new List<IdPk>();
  310. CosmosModelInfo container = await InitializeCollection<T>();
  311. string pk = GetPartitionKey<T>();
  312. Type type = typeof(T);
  313. int pages = (int)Math.Ceiling((double)enyites.Count / pageSize);
  314. Stopwatch stopwatch = Stopwatch.StartNew();
  315. for (int i = 0; i < pages; i++)
  316. {
  317. List<T> lists = enyites.Skip((i) * pageSize).Take(pageSize).ToList();
  318. List<KeyValuePair<PartitionKey, string>> itemsToInsert = new List<KeyValuePair<PartitionKey, string>>();
  319. lists.ForEach(x =>
  320. {
  321. object o = type.GetProperty(pk).GetValue(x, null);
  322. KeyValuePair<PartitionKey, string> keyValue = new KeyValuePair<PartitionKey, string>(new PartitionKey(o.ToString()), x.id);
  323. itemsToInsert.Add(keyValue);
  324. });
  325. List<Task> tasks = new List<Task>(lists.Count);
  326. itemsToInsert.ForEach(item =>
  327. {
  328. tasks.Add(container.container.DeleteItemStreamAsync(item.Value, item.Key)
  329. .ContinueWith((Task<ResponseMessage> task) =>
  330. {
  331. using (ResponseMessage response = task.Result)
  332. {
  333. idPks.Add(new IdPk { id = item.Value, pk = item.Key.ToString(), StatusCode = response.StatusCode });
  334. }
  335. }
  336. ));
  337. });
  338. await Task.WhenAll(tasks); if (container.cache && RedisHelper.Instance != null)
  339. {
  340. lists.ForEach(async x => {
  341. await RedisHelper.HDelAsync(CacheCosmosPrefix + container.container.Id, x.id);
  342. });
  343. }
  344. }
  345. stopwatch.Stop();
  346. return idPks;
  347. }
  348. public async Task<IdPk> DeleteAsync<T>(IdPk idPk) where T : ID
  349. {
  350. return await DeleteAsync<T>(idPk.id, idPk.pk);
  351. }
  352. public async Task<IdPk> DeleteAsync<T>(string id, string pk) where T : ID
  353. {
  354. CosmosModelInfo container = await InitializeCollection<T>();
  355. ResponseMessage response = await container.container.DeleteItemStreamAsync(id: id, partitionKey: new PartitionKey(pk));
  356. if (container.cache && RedisHelper.Instance != null)
  357. {
  358. await RedisHelper.HDelAsync(CacheCosmosPrefix + container.container.Id, id);
  359. }
  360. return new IdPk { id = id, pk = pk, StatusCode = response.StatusCode };
  361. }
  362. public async Task<IdPk> DeleteAsync<T>(T entity) where T : ID
  363. {
  364. CosmosModelInfo container = await InitializeCollection<T>();
  365. string partitionKey = GetPartitionKey<T>();
  366. Type type = typeof(T);
  367. object o = type.GetProperty(partitionKey).GetValue(entity, null);
  368. ResponseMessage response = await container.container.DeleteItemStreamAsync(id: entity.id, partitionKey: new PartitionKey(o.ToString()));
  369. if (container.cache && RedisHelper.Instance != null)
  370. {
  371. await RedisHelper.HDelAsync(CacheCosmosPrefix + container.container.Id, entity.id);
  372. }
  373. return new IdPk { id = entity.id, pk = o.ToString(), StatusCode = response.StatusCode };
  374. }
  375. //public async Task<T> DeleteAsync<T>(string id) where T : ID
  376. //{
  377. // Container container = await InitializeCollection<T>();
  378. // ItemResponse<T> response = await container.DeleteItemAsync<T>(id: id, partitionKey: new PartitionKey(GetPartitionKey<T>()));
  379. // return response.Resource;
  380. //}
  381. public async Task<List<T>> FindAll<T>(List<string> propertys = null) where T : ID
  382. {
  383. CosmosModelInfo container = await InitializeCollection<T>();
  384. StringBuilder sql;
  385. sql = SQLHelperParametric.GetSQLSelect(propertys);
  386. CosmosDbQuery cosmosDbQuery = new CosmosDbQuery { QueryText = sql.ToString() };
  387. FeedIterator<T> query = container.container.GetItemQueryIterator<T>(queryDefinition: cosmosDbQuery.CosmosQueryDefinition);
  388. return await ResultsFromFeedIterator(query);
  389. }
  390. private async Task<List<T>> ResultsFromFeedIterator<T>(FeedIterator<T> query, int? maxItemCount = null)
  391. {
  392. List<T> results = new List<T>();
  393. while (query.HasMoreResults)
  394. {
  395. foreach (T t in await query.ReadNextAsync())
  396. {
  397. results.Add(t);
  398. if (results.Count == maxItemCount)
  399. {
  400. return results;
  401. }
  402. }
  403. }
  404. return results;
  405. }
  406. private async Task<List<T>> ResultsFromFeedIterator<T>(FeedIterator<T> query, Func<List<T>, Task> batchAction, int itemsPerPage)
  407. {
  408. List<T> results = new List<T>();
  409. while (query.HasMoreResults)
  410. {
  411. if (results.Count() >= itemsPerPage)
  412. {
  413. await batchAction(results);
  414. results.Clear();
  415. }
  416. results.AddRange(await query.ReadNextAsync());
  417. }
  418. if (results.Count() > 0)
  419. {
  420. await batchAction(results);
  421. results.Clear();
  422. }
  423. return results;
  424. }
  425. public async Task<List<dynamic>> FindByDict(string CollectionName, Dictionary<string, object> dict, string partitionKey = null, List<string> propertys = null)
  426. {
  427. if (DocumentCollectionDict.TryGetValue(CollectionName, out CosmosModelInfo container))
  428. {
  429. //StringBuilder sql = new StringBuilder("select value(c) from c");
  430. //SQLHelper.GetSQL(dict, ref sql);
  431. //CosmosDbQuery cosmosDbQuery = new CosmosDbQuery, int itemsPerPage = -1, int?
  432. //{
  433. // QueryText = sql.ToString()
  434. //};
  435. StringBuilder sql;
  436. sql = SQLHelperParametric.GetSQLSelect(propertys);
  437. CosmosDbQuery cosmosDbQuery = SQLHelperParametric.GetSQL(dict, sql);
  438. QueryRequestOptions queryRequestOptions = GetDefaultQueryRequestOptions(itemsPerPage: GetEffectivePageSize(-1, null));
  439. FeedIterator<dynamic> query = container.container.GetItemQueryIterator<dynamic>(queryDefinition: cosmosDbQuery.CosmosQueryDefinition, requestOptions: queryRequestOptions);
  440. return await ResultsFromFeedIterator(query);
  441. }
  442. else
  443. {
  444. throw new BizException("CollectionName named:" + CollectionName + " dose not exsit in Database!");
  445. }
  446. }
  447. public async Task<List<dynamic>> FindCountByDict(string CollectionName, Dictionary<string, object> dict, string partitionKey = null)
  448. {
  449. if (DocumentCollectionDict.TryGetValue(CollectionName, out CosmosModelInfo container))
  450. {
  451. dict.Remove("@CURRPAGE");
  452. dict.Remove("@PAGESIZE");
  453. dict.Remove("@ASC");
  454. dict.Remove("@DESC");
  455. StringBuilder sql = new StringBuilder("select value count(c) from c");
  456. CosmosDbQuery cosmosDbQuery = SQLHelperParametric.GetSQL(dict, sql);
  457. QueryRequestOptions queryRequestOptions = GetDefaultQueryRequestOptions(itemsPerPage: GetEffectivePageSize(-1, null));
  458. FeedIterator<dynamic> query = container.container.GetItemQueryIterator<dynamic>(queryDefinition: cosmosDbQuery.CosmosQueryDefinition, requestOptions: queryRequestOptions);
  459. return await ResultsFromFeedIterator(query);
  460. }
  461. else
  462. {
  463. throw new BizException("CollectionName named:" + CollectionName + " dose not exsit in Database!");
  464. }
  465. }
  466. public async Task<List<T>> FindByParams<T>(Dictionary<string, object> dict, string partitionKey = null, List<string> propertys = null) where T : ID
  467. {
  468. return await FindByDict<T>(dict, partitionKey, propertys);
  469. }
  470. public async Task<List<T>> FindByDict<T>(Dictionary<string, object> dict, string partitionKey = null, List<string> propertys = null) where T : ID
  471. {
  472. StringBuilder sql;
  473. sql = SQLHelperParametric.GetSQLSelect(propertys);
  474. CosmosDbQuery cosmosDbQuery = SQLHelperParametric.GetSQL(dict, sql);
  475. QueryRequestOptions queryRequestOptions = GetDefaultQueryRequestOptions(itemsPerPage: GetEffectivePageSize(-1, null));
  476. return await ResultsFromQueryAndOptions<T>(cosmosDbQuery, queryRequestOptions);
  477. }
  478. private async Task<List<T>> ResultsFromQueryAndOptions<T>(CosmosDbQuery cosmosDbQuery, QueryRequestOptions queryOptions)
  479. {
  480. CosmosModelInfo container = await InitializeCollection<T>();
  481. FeedIterator<T> query = container.container.GetItemQueryIterator<T>(
  482. queryDefinition: cosmosDbQuery.CosmosQueryDefinition,
  483. requestOptions: queryOptions);
  484. return await ResultsFromFeedIterator(query);
  485. }
  486. private int GetEffectivePageSize(int itemsPerPage, int? maxItemCount)
  487. {
  488. return itemsPerPage == -1 ? maxItemCount ?? itemsPerPage : Math.Min(maxItemCount ?? itemsPerPage, itemsPerPage);
  489. }
  490. private QueryRequestOptions GetDefaultQueryRequestOptions(int? itemsPerPage = null,
  491. int? maxBufferedItemCount = null,
  492. int? maxConcurrency = null)
  493. {
  494. QueryRequestOptions queryRequestOptions = new QueryRequestOptions
  495. {
  496. MaxItemCount = itemsPerPage == -1 ? 1000 : itemsPerPage,
  497. MaxBufferedItemCount = maxBufferedItemCount ?? 100,
  498. MaxConcurrency = maxConcurrency ?? 50
  499. };
  500. return queryRequestOptions;
  501. }
  502. private async Task<List<T>> ResultsFromQueryAndOptions<T>(CosmosDbQuery cosmosDbQuery, Func<List<T>, Task> batchAction, QueryRequestOptions queryOptions)
  503. {
  504. CosmosModelInfo container = await InitializeCollection<T>();
  505. FeedIterator<T> query = container.container.GetItemQueryIterator<T>(
  506. queryDefinition: cosmosDbQuery.CosmosQueryDefinition,
  507. requestOptions: queryOptions);
  508. return await ResultsFromFeedIterator(query, batchAction, queryOptions.MaxItemCount ?? 0);
  509. }
  510. private QueryRequestOptions GetQueryRequestOptions(int itemsPerPage)
  511. {
  512. QueryRequestOptions queryRequestOptions = new QueryRequestOptions
  513. {
  514. MaxItemCount = itemsPerPage
  515. };
  516. return queryRequestOptions;
  517. }
  518. public async Task<List<T>> FindLinq<T>(Expression<Func<T, bool>> query = null, Expression<Func<T, object>> order = null, bool isDesc = false) where T : ID
  519. {
  520. //QueryRequestOptions queryRequestOptions = GetQueryRequestOptions(itemsPerPage);
  521. QueryRequestOptions queryRequestOptions = GetDefaultQueryRequestOptions(itemsPerPage: GetEffectivePageSize(-1, null));
  522. FeedIterator<T> feedIterator;
  523. CosmosModelInfo container = await InitializeCollection<T>();
  524. if (query == null)
  525. {
  526. if (order != null)
  527. {
  528. if (isDesc)
  529. {
  530. feedIterator = container.container
  531. .GetItemLinqQueryable<T>(requestOptions: queryRequestOptions).OrderByDescending(order)
  532. .ToFeedIterator();
  533. }
  534. else
  535. {
  536. feedIterator = container.container
  537. .GetItemLinqQueryable<T>(requestOptions: queryRequestOptions).OrderBy(order)
  538. .ToFeedIterator();
  539. }
  540. }
  541. else
  542. {
  543. feedIterator = container.container
  544. .GetItemLinqQueryable<T>(requestOptions: queryRequestOptions)
  545. .ToFeedIterator();
  546. }
  547. }
  548. else
  549. {
  550. if (order != null)
  551. {
  552. if (isDesc)
  553. {
  554. feedIterator = container.container
  555. .GetItemLinqQueryable<T>(requestOptions: queryRequestOptions)
  556. .Where(query).OrderByDescending(order)
  557. .ToFeedIterator();
  558. }
  559. else
  560. {
  561. feedIterator = container.container
  562. .GetItemLinqQueryable<T>(requestOptions: queryRequestOptions)
  563. .Where(query).OrderBy(order)
  564. .ToFeedIterator();
  565. }
  566. }
  567. else
  568. {
  569. feedIterator = container.container
  570. .GetItemLinqQueryable<T>(requestOptions: queryRequestOptions)
  571. .Where(query)
  572. .ToFeedIterator();
  573. }
  574. }
  575. return await ResultsFromFeedIterator<T>(feedIterator);
  576. }
  577. public async Task<List<T>> FindSQL<T>(string sql, Dictionary<string, object> Parameters = null) where T : ID
  578. {
  579. CosmosModelInfo container = await InitializeCollection<T>();
  580. QueryRequestOptions queryOptions = GetQueryRequestOptions(GetEffectivePageSize(-1, null));
  581. if (Parameters != null)
  582. {
  583. CosmosDbQuery cosmosDbQuery = new CosmosDbQuery
  584. {
  585. QueryText = sql,
  586. Parameters = Parameters
  587. };
  588. FeedIterator<T> feedIterator = container.container
  589. .GetItemQueryIterator<T>(cosmosDbQuery.CosmosQueryDefinition, requestOptions: queryOptions);
  590. return await ResultsFromFeedIterator(feedIterator);
  591. }
  592. else
  593. {
  594. QueryDefinition queryDefinition = new QueryDefinition(sql);
  595. return await ResultsFromFeedIterator<T>(container.container.GetItemQueryIterator<T>(queryDefinition));
  596. }
  597. }
  598. public async Task<T> Save<T>(T entity) where T : ID
  599. {
  600. try
  601. {
  602. CosmosModelInfo container = await InitializeCollection<T>();
  603. ItemResponse<T> response = await container.container.CreateItemAsync<T>(entity);
  604. if (container.cache && RedisHelper.Instance != null)
  605. {
  606. if (!RedisHelper.Exists(CacheCosmosPrefix + container.container.Id))
  607. {
  608. await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, entity.id, entity);
  609. }
  610. else
  611. {
  612. await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, entity.id, entity);
  613. await RedisHelper.ExpireAsync(CacheCosmosPrefix + container.container.Id, timeoutSeconds);
  614. }
  615. }
  616. return response.Resource;
  617. }
  618. catch (Exception e)
  619. {
  620. throw new BizException(e.Message);
  621. }
  622. }
  623. public async Task<List<T>> SaveAll<T>(List<T> enyites) where T : ID
  624. {
  625. int pages = (int)Math.Ceiling((double)enyites.Count / pageSize);
  626. CosmosModelInfo container = await InitializeCollection<T>();
  627. bool flag = false;
  628. if (RedisHelper.Exists(CacheCosmosPrefix + container.container.Id))
  629. {
  630. flag = true;
  631. }
  632. string pk = GetPartitionKey<T>();
  633. Type type = typeof(T);
  634. Stopwatch stopwatch = Stopwatch.StartNew();
  635. for (int i = 0; i < pages; i++)
  636. {
  637. List<T> lists = enyites.Skip((i) * pageSize).Take(pageSize).ToList();
  638. List<KeyValuePair<PartitionKey, Stream>> itemsToInsert = new List<KeyValuePair<PartitionKey, Stream>>();
  639. lists.ForEach(async x =>
  640. {
  641. MemoryStream stream = new MemoryStream();
  642. await JsonSerializer.SerializeAsync(stream, x, new JsonSerializerOptions { IgnoreNullValues = true });
  643. object o = type.GetProperty(pk).GetValue(x, null);
  644. KeyValuePair<PartitionKey, Stream> keyValue = new KeyValuePair<PartitionKey, Stream>(new PartitionKey(o.ToString()), stream);
  645. itemsToInsert.Add(keyValue);
  646. });
  647. List<Task> tasks = new List<Task>(lists.Count);
  648. itemsToInsert.ForEach(item =>
  649. {
  650. tasks.Add(container.container.CreateItemStreamAsync(item.Value, item.Key)
  651. .ContinueWith((Task<ResponseMessage> task) =>
  652. {
  653. using (ResponseMessage response = task.Result)
  654. {
  655. if (!response.IsSuccessStatusCode)
  656. {
  657. }
  658. }
  659. }
  660. ));
  661. });
  662. await Task.WhenAll(tasks);
  663. if (container.cache && RedisHelper.Instance != null)
  664. {
  665. lists.ForEach(async x => {
  666. await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, x.id, x);
  667. });
  668. }
  669. }
  670. if (container.cache && RedisHelper.Instance != null && !flag)
  671. {
  672. await RedisHelper.ExpireAsync(CacheCosmosPrefix + container.container.Id, timeoutSeconds);
  673. }
  674. stopwatch.Stop();
  675. return enyites;
  676. }
  677. public async Task<T> SaveOrUpdate<T>(T entity) where T : ID
  678. {
  679. CosmosModelInfo container = await InitializeCollection<T>();
  680. ItemResponse<T> response = await container.container.UpsertItemAsync(item: entity);
  681. if (container.cache && RedisHelper.Instance != null)
  682. {
  683. if (!RedisHelper.Exists(CacheCosmosPrefix + container.container.Id))
  684. {
  685. await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, entity.id, entity);
  686. }
  687. else
  688. {
  689. await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, entity.id, entity);
  690. await RedisHelper.ExpireAsync(CacheCosmosPrefix + container.container.Id, timeoutSeconds);
  691. }
  692. }
  693. return response.Resource;
  694. }
  695. public async Task<List<T>> SaveOrUpdateAll<T>(List<T> enyites) where T : ID
  696. {
  697. //await Task.Run(() => Parallel.ForEach(entities, new ParallelOptions { MaxDegreeOfParallelism = 2 }, (item) =>
  698. //{
  699. // Task.WaitAll(Update(item));
  700. //}));
  701. int pages = (int)Math.Ceiling((double)enyites.Count / pageSize);
  702. CosmosModelInfo container = await InitializeCollection<T>();
  703. bool flag = false;
  704. if (RedisHelper.Exists(CacheCosmosPrefix + container.container.Id))
  705. {
  706. flag = true;
  707. }
  708. string pk = GetPartitionKey<T>();
  709. Type type = typeof(T);
  710. Stopwatch stopwatch = Stopwatch.StartNew();
  711. for (int i = 0; i < pages; i++)
  712. {
  713. List<T> lists = enyites.Skip((i) * pageSize).Take(pageSize).ToList();
  714. List<KeyValuePair<PartitionKey, Stream>> itemsToInsert = new List<KeyValuePair<PartitionKey, Stream>>();
  715. lists.ForEach(async x =>
  716. {
  717. MemoryStream stream = new MemoryStream();
  718. await JsonSerializer.SerializeAsync(stream, x, new JsonSerializerOptions { IgnoreNullValues = true });
  719. object o = type.GetProperty(pk).GetValue(x, null);
  720. KeyValuePair<PartitionKey, Stream> keyValue = new KeyValuePair<PartitionKey, Stream>(new PartitionKey(o.ToString()), stream);
  721. itemsToInsert.Add(keyValue);
  722. });
  723. List<Task> tasks = new List<Task>(lists.Count);
  724. itemsToInsert.ForEach(item =>
  725. {
  726. tasks.Add(container.container.UpsertItemStreamAsync(item.Value, item.Key)
  727. .ContinueWith((Task<ResponseMessage> task) =>
  728. {
  729. //using (ResponseMessage response = task.Result)
  730. //{
  731. // if (!response.IsSuccessStatusCode)
  732. // {
  733. // }
  734. //}
  735. }
  736. ));
  737. });
  738. await Task.WhenAll(tasks);
  739. if (container.cache && RedisHelper.Instance != null)
  740. {
  741. lists.ForEach(async x => {
  742. await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, x.id, x);
  743. });
  744. }
  745. }
  746. if (container.cache && RedisHelper.Instance != null && !flag)
  747. {
  748. await RedisHelper.ExpireAsync(CacheCosmosPrefix + container.container.Id, timeoutSeconds);
  749. }
  750. stopwatch.Stop();
  751. return enyites;
  752. }
  753. public async Task<T> Update<T>(T entity) where T : ID
  754. {
  755. CosmosModelInfo container = await InitializeCollection<T>();
  756. string pk = GetPartitionKey<T>();
  757. object o = typeof(T).GetProperty(pk).GetValue(entity, null);
  758. ItemResponse<T> response = await container.container.ReplaceItemAsync(entity, entity.id, new PartitionKey(o.ToString()));
  759. if (container.cache && RedisHelper.Instance != null)
  760. {
  761. if (!RedisHelper.Exists(CacheCosmosPrefix + container.container.Id))
  762. {
  763. await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, entity.id, entity);
  764. }
  765. else
  766. {
  767. await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, entity.id, entity);
  768. await RedisHelper.ExpireAsync(CacheCosmosPrefix + container.container.Id, timeoutSeconds);
  769. }
  770. }
  771. return response.Resource;
  772. }
  773. internal class Item
  774. {
  775. public string id { get; set; }
  776. public string pk { get; set; }
  777. public MemoryStream stream { get; set; }
  778. }
  779. public async Task<List<T>> UpdateAll<T>(List<T> enyites) where T : ID
  780. {
  781. //await Task.Run(() => Parallel.ForEach(entities, new ParallelOptions { MaxDegreeOfParallelism = 2 }, (item) =>
  782. //{
  783. // Task.WaitAll(Update(item));
  784. //}));
  785. int pages = (int)Math.Ceiling((double)enyites.Count / pageSize);
  786. CosmosModelInfo container = await InitializeCollection<T>();
  787. bool flag = false;
  788. if (RedisHelper.Exists(CacheCosmosPrefix + container.container.Id))
  789. {
  790. flag = true;
  791. }
  792. string pk = GetPartitionKey<T>();
  793. Type type = typeof(T);
  794. Stopwatch stopwatch = Stopwatch.StartNew();
  795. for (int i = 0; i < pages; i++)
  796. {
  797. List<T> lists = enyites.Skip((i) * pageSize).Take(pageSize).ToList();
  798. List<Item> itemsToInsert = new List<Item>();
  799. lists.ForEach(async x =>
  800. {
  801. MemoryStream stream = new MemoryStream();
  802. await JsonSerializer.SerializeAsync(stream, x, new JsonSerializerOptions { IgnoreNullValues = true });
  803. object o = type.GetProperty(pk).GetValue(x, null);
  804. Item keyValue = new Item { id = x.id, pk = o.ToString(), stream = stream };
  805. itemsToInsert.Add(keyValue);
  806. });
  807. List<Task> tasks = new List<Task>(lists.Count);
  808. itemsToInsert.ForEach(item =>
  809. {
  810. tasks.Add(container.container.ReplaceItemStreamAsync(item.stream, item.id, new PartitionKey(item.pk))
  811. .ContinueWith((Task<ResponseMessage> task) =>
  812. {
  813. //using (ResponseMessage response = task.Result)
  814. //{
  815. // if (!response.IsSuccessStatusCode)
  816. // {
  817. // }
  818. //}
  819. }
  820. ));
  821. });
  822. await Task.WhenAll(tasks);
  823. if (container.cache && RedisHelper.Instance != null)
  824. {
  825. lists.ForEach(async x => {
  826. await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, x.id, x);
  827. });
  828. }
  829. }
  830. if (container.cache && RedisHelper.Instance != null && !flag)
  831. {
  832. await RedisHelper.ExpireAsync(CacheCosmosPrefix + container.container.Id, timeoutSeconds);
  833. }
  834. stopwatch.Stop();
  835. return enyites;
  836. }
  837. //public void Dispose()
  838. //{
  839. // Dispose(true);
  840. //}
  841. //protected virtual void Dispose(bool disposing)
  842. //{
  843. // if (disposing)
  844. // {
  845. // CosmosClient?.Dispose();
  846. // }
  847. //}
  848. private async Task<T> FindByIdAsSql<T>(string id) where T : ID
  849. {
  850. CosmosModelInfo container = await InitializeCollection<T>();
  851. CosmosDbQuery cosmosDbQuery = new CosmosDbQuery
  852. {
  853. QueryText = @"SELECT *
  854. FROM c
  855. WHERE c.id = @id",
  856. Parameters = new Dictionary<string, object>
  857. {
  858. { "@id",id}
  859. }
  860. };
  861. FeedIterator<T> feedIterator = container.container
  862. .GetItemQueryIterator<T>(cosmosDbQuery.CosmosQueryDefinition);
  863. return (await ResultsFromFeedIterator(feedIterator)).SingleOrDefault();
  864. }
  865. public async Task<T> FindByIdPk<T>(string id, string pk) where T : ID
  866. {
  867. CosmosModelInfo container = await InitializeCollection<T>();
  868. ItemResponse<T> response = await container.container.ReadItemAsync<T>(id: id, partitionKey: new PartitionKey(pk));
  869. return response.Resource;
  870. }
  871. public async Task<T> FindById<T>(string id) where T : ID
  872. {
  873. CosmosModelInfo container = await InitializeCollection<T>();
  874. if (container.cache && RedisHelper.Instance != null)
  875. {
  876. return await RedisHelper.CacheShellAsync(CacheCosmosPrefix + container.container.Id, id, timeoutSeconds, () => { return FindByIdAsSql<T>(id); });
  877. }
  878. else
  879. {
  880. return await FindByIdAsSql<T>(id);
  881. }
  882. }
  883. public async Task<List<T>> FindByIds<T>(List<string> ids) where T : ID
  884. {
  885. CosmosModelInfo container = await InitializeCollection<T>();
  886. if (container.cache && RedisHelper.Instance != null)
  887. {
  888. List<T> list = new List<T>();
  889. List<string> NotIn = new List<string>();
  890. foreach (string id in ids)
  891. {
  892. if (!await RedisHelper.HExistsAsync(CacheCosmosPrefix + container.container.Id, id))
  893. {
  894. NotIn.Add(id);
  895. }
  896. else
  897. {
  898. list.Add(await RedisHelper.HGetAsync<T>(CacheCosmosPrefix + container.container.Id, id));
  899. }
  900. }
  901. if (NotIn.IsNotEmpty())
  902. {
  903. List<T> noInList = await FindByDict<T>(new Dictionary<string, object> { { "id", NotIn.ToArray() } });
  904. noInList.ForEach(x => { RedisHelper.HSet(CacheCosmosPrefix + container.container.Id, x.id, x); RedisHelper.Expire(CacheCosmosPrefix + container.container.Id, timeoutSeconds); });
  905. list.AddRange(noInList);
  906. }
  907. return list;
  908. }
  909. else
  910. {
  911. return await FindByDict<T>(new Dictionary<string, object> { { "id", ids.ToArray() } });
  912. }
  913. }
  914. public async Task<dynamic> FindById(string CollectionName, string id)
  915. {
  916. if (DocumentCollectionDict.TryGetValue(CollectionName, out CosmosModelInfo container))
  917. {
  918. if (container.cache && RedisHelper.Instance != null)
  919. {
  920. return await RedisHelper.CacheShellAsync(CacheCosmosPrefix + container.container.Id, id, timeoutSeconds, () => { return FindByDict(CollectionName, new Dictionary<string, object> { { "id", id } }); });
  921. }
  922. else
  923. {
  924. return await FindByDict(CollectionName, new Dictionary<string, object> { { "id", id } });
  925. }
  926. }
  927. else
  928. {
  929. throw new BizException("CollectionName named:" + CollectionName + " dose not exsit in Database!");
  930. }
  931. }
  932. public async Task<List<dynamic>> FindByIds(string CollectionName, List<string> ids)
  933. {
  934. if (DocumentCollectionDict.TryGetValue(CollectionName, out CosmosModelInfo container))
  935. {
  936. if (container.cache && RedisHelper.Instance != null)
  937. {
  938. return await RedisHelper.CacheShellAsync(CacheCosmosPrefix + container.container.Id, timeoutSeconds, () => { return FindByDict(CollectionName, new Dictionary<string, object> { { "id", ids.ToArray() } }); });
  939. }
  940. else
  941. {
  942. return await FindByDict(CollectionName, new Dictionary<string, object> { { "id", ids.ToArray() } });
  943. }
  944. }
  945. else
  946. {
  947. throw new BizException("CollectionName named:" + CollectionName + " dose not exsit in Database!");
  948. }
  949. }
  950. }
  951. }