J1147
/
HealthMonitor
Suivre 1
Ajouter aux favoris 0
Bifurcation 0
Code Tickets 0 Demandes d'ajout 0 Versions 0 Wiki Activité
Parcourir la source

使用ConcurrentQueue超类作为队列

td_orm
H Vs il y a 1 an
Parent
65df0977c0
révision
f3f5ffad37
5 fichiers modifiés avec 135 ajouts et 102 suppressions
Vue séparée
Option de Diff
Voir les Statistiques Télécharger le Fichier Patch Télécharger le Fichier des Différences
  1. +92
    -92
      HealthMonitor.Service/Resolver/BloodpressResolver.cs
  2. +29
    -6
      HealthMonitor.Service/Sub/MsgManager.cs
  3. +8
    -2
      HealthMonitor.Service/Sub/MsgQueueManager.cs
  4. +5
    -2
      HealthMonitor.Service/Sub/TDengineDataSubcribe.cs
  5. +1
    -0
      HealthMonitor.WebApi/Program.cs

+ 92
- 92
HealthMonitor.Service/Resolver/BloodpressResolver.cs Voir le fichier

@@ -73,101 +73,101 @@ namespace HealthMonitor.Service.Resolver
Console.WriteLine($"ExecuteMessageAsync" + messageId);
var bp = _msgData.Value!;

//#region 获取个人信息
#region 获取个人信息

var person = await _personCacheMgr.GetDeviceGpsPersonCacheBySerialNoAsync(bp.MessageId, bp.Serialno).ConfigureAwait(false);
Console.WriteLine(person?.Person.PersonName);
// 验证这个信息是否存在
//if (person == null || person?.Person.BornDate == null)
//{
// Console.WriteLine("验证这个信息是否存在");
// return;
//}
//// 验证年龄是否在范围 （2 - 120）
//var age = SafeType.SafeInt(DateTime.Today.Year - person?.Person.BornDate!.Value.Year!);
//if (age < 1 || age > 120)
//{
// Console.WriteLine("验证年龄是否在范围 （2 - 120）");
//}
//#endregion
//var gender = person?.Person.Gender == true ? 1 : 2;
//var isHypertension = SafeType.SafeBool(person?.Person.Ishypertension!);
//var height = SafeType.SafeDouble(person?.Person.Height!);
//var weight = SafeType.SafeDouble(person?.Person.Weight!);
//#region 计算增量值
//var bpRef = await _bpRefValCacheManager.GetBloodPressReferenceValueAsync(age, gender, isHypertension);
//var systolicRefValue = bpRef?.Systolic;//?
//var diastolicRefValue = bpRef?.Diastolic;//?
//int duration = 30;
//// 获取历史数据
////DateTime now = DateTime.Now;
//DateTime now = (DateTime)bp.LastUpdate!; //测试
//DateTime startTime = now.AddDays(-duration);
//DateTime endTime = now;
////
//var systolicAggregate = await _serviceTDengine.GetAggregateValue("systolic_value", "hm_bloodpress", $"ts>='{startTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and ts <='{endTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and serialno='{bp.Serialno}'");
//var diastolicAggregate = await _serviceTDengine.GetAggregateValue("diastolic_value", "hm_bloodpress", $"ts>='{startTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and ts <='{endTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and serialno='{bp.Serialno}'");
//// 最大值
//var systolicMax = systolicAggregate.Max;
//var diastolicMax = diastolicAggregate.Max;
//// 最小值
//var systolicMin = systolicAggregate.Min;
//var diastolicMin = diastolicAggregate.Min;
//// 计算去除最大值和最小值和异常值的平均值
//var systolicAvg = await _serviceTDengine.GetAvgExceptMaxMinValue("systolic_value", "hm_bloodpress", $"ts>='{startTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and ts <='{endTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and serialno='{bp.Serialno}' and systolic_value < {systolicRefValue} ");
//var diastolicAvg = await _serviceTDengine.GetAvgExceptMaxMinValue("diastolic_value", "hm_bloodpress", $"ts>='{startTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and ts <='{endTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and serialno='{bp.Serialno}' and diastolic_value < {diastolicRefValue}");
//// 偏移参数
//var avgOffset = 0.25M;
//var systolicAvgOffset = avgOffset;
//var diastolicAvgOffset = avgOffset;
//// 增量值=（标定值-平均值）* 0.25
//var systolicInc = systolicAvg.Equals(0M) ? 0 : (int)((systolicRefValue - systolicAvg) * systolicAvgOffset)!;
//var diastolicInc = diastolicAvg.Equals(0M) ? 0 : (int)((diastolicRefValue - diastolicAvg) * diastolicAvgOffset)!;
//#endregion
//#region 插入BP增量值 hm_bloodpress_stats_inc
//var sql = $"INSERT INTO health_monitor.hm_bloodpress_stats_inc VALUES(" +
// $"'{bp.LastUpdate:yyyy-MM-ddTHH:mm:ss.fffZ}'," +
// $"'{bp.BloodPressId}'," +
// $"'{bp.MessageId}'," +
// $"'{bp.Serialno}'," +
// $"{bp.SystolicValue}," +
// $"{systolicRefValue}," +
// $"{systolicAvg}," +
// $"{systolicMax}," +
// $"{systolicMin}," +
// $"{systolicAvgOffset}," +
// $"{systolicInc}," +
// $"{bp.DiastolicValue}," +
// $"{diastolicRefValue}," +
// $"{diastolicAvg}," +
// $"{diastolicMax}," +
// $"{diastolicMin}," +
// $"{diastolicAvgOffset}," +
// $"{diastolicInc}," +
// $"{gender}," +
// $"{age}," +
// $"{height}," +
// $"{weight}," +
// $"'{bp.LastUpdate:yyyy-MM-ddTHH:mm:ss.fffZ}'," +
// $"{duration}," +
// $"'{startTime:yyyy-MM-ddTHH:mm:ss.fffZ}'," +
// $"'{endTime:yyyy-MM-ddTHH:mm:ss.fffZ}'," +
// $"'{string.Empty}')";
//var res = await _serviceTDengine.GernalRestSql(sql);
//#endregion
//验证这个信息是否存在
if (person == null || person?.Person.BornDate == null)
{
Console.WriteLine("验证这个信息是否存在");
return;
}
// 验证年龄是否在范围 （2 - 120）
var age = SafeType.SafeInt(DateTime.Today.Year - person?.Person.BornDate!.Value.Year!);
if (age < 1 || age > 120)
{
Console.WriteLine("验证年龄是否在范围 （2 - 120）");
}
#endregion
var gender = person?.Person.Gender == true ? 1 : 2;
var isHypertension = SafeType.SafeBool(person?.Person.Ishypertension!);
var height = SafeType.SafeDouble(person?.Person.Height!);
var weight = SafeType.SafeDouble(person?.Person.Weight!);
#region 计算增量值
var bpRef = await _bpRefValCacheManager.GetBloodPressReferenceValueAsync(age, gender, isHypertension);
var systolicRefValue = bpRef?.Systolic;//?
var diastolicRefValue = bpRef?.Diastolic;//?
int duration = 30;
// 获取历史数据
//DateTime now = DateTime.Now;
DateTime now = (DateTime)bp.LastUpdate!; //测试
DateTime startTime = now.AddDays(-duration);
DateTime endTime = now;
//
var systolicAggregate = await _serviceTDengine.GetAggregateValue("systolic_value", "hm_bloodpress", $"ts>='{startTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and ts <='{endTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and serialno='{bp.Serialno}'");
var diastolicAggregate = await _serviceTDengine.GetAggregateValue("diastolic_value", "hm_bloodpress", $"ts>='{startTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and ts <='{endTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and serialno='{bp.Serialno}'");
// 最大值
var systolicMax = systolicAggregate.Max;
var diastolicMax = diastolicAggregate.Max;
// 最小值
var systolicMin = systolicAggregate.Min;
var diastolicMin = diastolicAggregate.Min;
// 计算去除最大值和最小值和异常值的平均值
var systolicAvg = await _serviceTDengine.GetAvgExceptMaxMinValue("systolic_value", "hm_bloodpress", $"ts>='{startTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and ts <='{endTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and serialno='{bp.Serialno}' and systolic_value < {systolicRefValue} ");
var diastolicAvg = await _serviceTDengine.GetAvgExceptMaxMinValue("diastolic_value", "hm_bloodpress", $"ts>='{startTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and ts <='{endTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and serialno='{bp.Serialno}' and diastolic_value < {diastolicRefValue}");
// 偏移参数
var avgOffset = 0.25M;
var systolicAvgOffset = avgOffset;
var diastolicAvgOffset = avgOffset;
// 增量值=（标定值-平均值）* 0.25
var systolicInc = systolicAvg.Equals(0M) ? 0 : (int)((systolicRefValue - systolicAvg) * systolicAvgOffset)!;
var diastolicInc = diastolicAvg.Equals(0M) ? 0 : (int)((diastolicRefValue - diastolicAvg) * diastolicAvgOffset)!;
#endregion
#region 插入BP增量值 hm_bloodpress_stats_inc
var sql = $"INSERT INTO health_monitor.hm_bloodpress_stats_inc VALUES(" +
$"'{bp.LastUpdate:yyyy-MM-ddTHH:mm:ss.fffZ}'," +
$"'{bp.BloodPressId}'," +
$"'{bp.MessageId}'," +
$"'{bp.Serialno}'," +
$"{bp.SystolicValue}," +
$"{systolicRefValue}," +
$"{systolicAvg}," +
$"{systolicMax}," +
$"{systolicMin}," +
$"{systolicAvgOffset}," +
$"{systolicInc}," +
$"{bp.DiastolicValue}," +
$"{diastolicRefValue}," +
$"{diastolicAvg}," +
$"{diastolicMax}," +
$"{diastolicMin}," +
$"{diastolicAvgOffset}," +
$"{diastolicInc}," +
$"{gender}," +
$"{age}," +
$"{height}," +
$"{weight}," +
$"'{bp.LastUpdate:yyyy-MM-ddTHH:mm:ss.fffZ}'," +
$"{duration}," +
$"'{startTime:yyyy-MM-ddTHH:mm:ss.fffZ}'," +
$"'{endTime:yyyy-MM-ddTHH:mm:ss.fffZ}'," +
$"'{string.Empty}')";
var res = await _serviceTDengine.GernalRestSql(sql);
#endregion
}




+ 29
- 6
HealthMonitor.Service/Sub/MsgManager.cs Voir le fichier

@@ -2,6 +2,7 @@
using HealthMonitor.Service.Resolver.Interface;
using Microsoft.Extensions.Logging;
using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
@@ -15,13 +16,15 @@ namespace HealthMonitor.Service.Sub
public class MsgManager
{
private readonly MsgQueueManager _msgQueueManager;
private readonly BloodpressResolver _resolver;
private readonly ILogger<MsgManager> _logger;
private PackageMsgModel _msg = default!;
public MsgManager(ILogger<MsgManager> logger, BloodpressResolver resolver)
public MsgManager(ILogger<MsgManager> logger, BloodpressResolver resolver, MsgQueueManager msgQueueManager)
{
_logger = logger;
_resolver = resolver;
_msgQueueManager = msgQueueManager;
}

public void EnqueueMsg(PackageMsgModel msg)
@@ -30,18 +33,38 @@ namespace HealthMonitor.Service.Sub
Console.WriteLine($"GetMsgResolver-{_msg.MessageId}");
}


public IResolver? GetMsgResolver()
{
if (_msg == null)

//if (_msg == null)
//{
// return null;
//}
//Console.WriteLine($"GetMsgResolver-{_msg.MessageId}");
//_msgQueueManager.TryDequeue(out object obj);
_msgQueueManager.TryDequeue(out var msg);
if (msg == null)
{
return null;
}
Console.WriteLine($"GetMsgResolver-{_msg.MessageId}");
_resolver.SetResolveInfo(_msg);
_msg = default!;
Console.WriteLine($"GetMsgResolver-{msg.MessageId}");
_resolver.SetResolveInfo(msg);
return _resolver;
}
//public IReso

//public IResolver? GetMsgResolver()
//{
// if (_msg == null)
// {
// return null;
// }
// Console.WriteLine($"GetMsgResolver-{_msg.MessageId}");
// _resolver.SetResolveInfo(_msg);
// _msg = default!;
// return _resolver;
//}
//public IResolver? GetMsgResolver(PackageMsgModel msg)
//{
// if (_msg == null)


+ 8
- 2
HealthMonitor.Service/Sub/MsgQueueManager.cs Voir le fichier

@@ -1,4 +1,6 @@
using System;
using HealthMonitor.Service.Resolver;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
@@ -9,6 +11,10 @@ namespace HealthMonitor.Service.Sub
{
public class MsgQueueManager: ConcurrentQueue<PackageMsgModel>
{

private readonly ILogger<MsgQueueManager> _logger;
public MsgQueueManager(ILogger<MsgQueueManager> logger)
{
_logger = logger;
}
}
}

+ 5
- 2
HealthMonitor.Service/Sub/TDengineDataSubcribe.cs Voir le fichier

@@ -25,6 +25,7 @@ namespace HealthMonitor.Service.Sub
private IConsumer _consumer = default!;
private IntPtr _conn = default!;
private readonly MsgManager _msgManager;
private readonly MsgQueueManager _msgQueueManager;
private readonly TDengineService _serviceTDengine;
private readonly PersonCacheManager _personCacheMgr;
private readonly BloodPressReferenceValueCacheManager _bpRefValCacheManager;
@@ -39,6 +40,7 @@ namespace HealthMonitor.Service.Sub
BloodPressReferenceValueCacheManager bpRefValCacheManager,
IResolverFactory resolverFactory,
MsgManager msgManager,
MsgQueueManager msgQueueManager,
ILogger<TDengineDataSubcribe> logger
)
{
@@ -48,6 +50,7 @@ namespace HealthMonitor.Service.Sub
_logger = logger;
_resolverFactory = resolverFactory;
_msgManager = msgManager;
_msgQueueManager = msgQueueManager;
_conn = GetConnection();
}
public void BeginListen(CancellationToken stoppingToken)
@@ -206,8 +209,8 @@ namespace HealthMonitor.Service.Sub
Console.WriteLine(msg!.MessageId);
if (msg == null) return;
// ConcurrentQueue<ReceiveMessageModel> messageQueue = new ConcurrentQueue<ReceiveMessageModel>();
_msgManager.EnqueueMsg(msg!);
//_msgManager.EnqueueMsg(msg!);
_msgQueueManager.Enqueue(msg);

}



+ 1
- 0
HealthMonitor.WebApi/Program.cs Voir le fichier

@@ -166,6 +166,7 @@ namespace HealthMonitor.WebApi
#endregion

#region Worker
builder.Services.AddSingleton<MsgQueueManager>();
builder.Services.AddSingleton<IResolverFactory, ResolverFactory>();
builder.Services.AddSingleton<BloodpressResolver>();
builder.Services.AddSingleton<PackageProcess>();


Écrire Aperçu
Chargement…
Annuler
Enregistrer