Boonlive_RD_Web/Web_BLSKafka_Server_Prod
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

初始化

Browse Source
This commit is contained in:
徐
2025-11-21 08:48:01 +08:00
commit b4d684a84c
202 changed files with 28585 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

469
BLWLogServer/Services/KafkaConsume.cs Normal file
View File

@@ -0,0 +1,469 @@
using System.Text;
using Common;
using CommonEntity;
using CommonTools;
using Confluent.Kafka;
using MongoDB.Driver;
using NLog;
using System.Collections.Generic;
namespace BLWLogServer.Services
{
public class KafkaConsume : BackgroundService
{
public IConfiguration Configuration { get; set; }
public KafkaConsume(IConfiguration configuration)
{
Configuration = configuration;
}
public static Logger logger = NLog.LogManager.GetCurrentClassLogger();
// 添加两个队列用于存储最近10条TotalCount
private static Queue<long> _cpuMaxQueue = new(10);
private static Queue<long> _cpuAvgQueue = new(10);
private static Queue<long> _cpuMinQueue = new(10);
private static Queue<long> _TotalSendPackage = new(10);
private static Queue<long> _TotalRecvPackage = new(10);
private static Queue<long> _RCUOnLine = new(10);
protected async override Task ExecuteAsync(CancellationToken stoppingToken)
{
await Task.Factory.StartNew(async () =>
{
var consumers = new List<Task>();
for (int i = 0; i < 3; i++)
{
consumers.Add(Task.Run(() => StartConsumer(stoppingToken)));
}
await Task.WhenAll(consumers);
//await StartConsumer(stoppingToken);
}, TaskCreationOptions.LongRunning);
}
private async Task StartConsumer(CancellationToken stoppingToken)
{
string? ipport = Configuration["Kafka:EndPoint"];
string? user = Configuration["Kafka:UserName"];
string? pwd = Configuration["Kafka:PassWord"];
string? mongodbconnectstr = Configuration["Mongodb:Connectstr"];
while (!stoppingToken.IsCancellationRequested)
{
var config = new ConsumerConfig
{
GroupId = "blwlogserver-consumer-group",
AutoOffsetReset = AutoOffsetReset.Earliest,
BootstrapServers = ipport,
SecurityProtocol = SecurityProtocol.SaslPlaintext,
SaslMechanism = SaslMechanism.Plain,
SaslUsername = user,
SaslPassword = pwd
};
var c = new ConsumerBuilder<string, byte[]>(config).Build();
c.Subscribe(KafkaKey.BLWLog_RCU_Topic);
var connectionString = mongodbconnectstr;
var client = new MongoClient(connectionString);
try
{
while (true)
{
var cr = c.Consume(stoppingToken);
try
{
var k = cr.Message.Key;
var v = cr.Message.Value;
if (k.Equals(KafkaKey.UDPPackageKey))
{
//var UUU = Encoding.UTF8.GetString(v);
//Console.WriteLine($"Consumed message '{k} {v}' at: '{cr.TopicPartitionOffset}'.");
var collection = client.GetDatabase("udppackage").GetCollection<UDPPackage_db>("totalcount");
UDPPackage UDPPPP = MyMessagePacker.FastDeserialize<UDPPackage>(v);
UDPPackage_db us = new UDPPackage_db();
us.TotalCount = UDPPPP.TotalCount;
us.CommandType = UDPPPP.CommandType;
us.FenLei = UDPPPP.FenLei;
us.ExtraData = UDPPPP.ExtraData;
us.RemoveTimeString = UDPPPP.RemoveTime;
us.RemoveTime = DateTime.Parse(UDPPPP.RemoveTime);
await collection.InsertOneAsync(us);
// xu修改20250908
// 修改CPUMax处理逻辑
if (UDPPPP.CommandType == "UDPPackage_CPUMax")
{
// 维护队列长度不超过10
if (_cpuMaxQueue.Count >= 10)
{
_cpuMaxQueue.Dequeue();
}
_cpuMaxQueue.Enqueue(UDPPPP.TotalCount);
// 将队列转为逗号分隔字符串
string arrayString = string.Join(",", _cpuMaxQueue);
CSRedisCacheHelper.Set("UDPPackage_CPUMax", arrayString, 10);
}
if (UDPPPP.CommandType == "UDPPackage_CPUMin")
{
// 维护队列长度不超过10
if (_cpuMinQueue.Count >= 10)
{
_cpuMinQueue.Dequeue();
}
_cpuMinQueue.Enqueue(UDPPPP.TotalCount);
// 将队列转为逗号分隔字符串
string arrayString = string.Join(",", _cpuMinQueue);
CSRedisCacheHelper.Set("UDPPackage_CPUMin", arrayString, 10);
}
// 修改CPUAvg处理逻辑
if (UDPPPP.CommandType == "UDPPackage_CPUAvg")
{
// 维护队列长度不超过10
if (_cpuAvgQueue.Count >= 10)
{
_cpuAvgQueue.Dequeue();
}
_cpuAvgQueue.Enqueue(UDPPPP.TotalCount);
// 将队列转为逗号分隔字符串
string arrayString = string.Join(",", _cpuAvgQueue);
CSRedisCacheHelper.Set("UDPPackage_CPUAvg", arrayString, 10);
CSRedisCacheHelper.Set("UDPPackage_DetectTime", DateTime.UtcNow.ToString("o"), 10);
}
if (UDPPPP.CommandType == "UDPPackage_TotalSendPackage")
{
// 维护队列长度不超过10
if (_TotalSendPackage.Count >= 10)
{
_TotalSendPackage.Dequeue();
}
_TotalSendPackage.Enqueue(UDPPPP.TotalCount);
// 将队列转为逗号分隔字符串
string arrayString = string.Join(",", _TotalSendPackage);
CSRedisCacheHelper.Set("UDPPackage_TotalSendPackage", arrayString, 10);
}
if (UDPPPP.CommandType == "UDPPackage_TotalRecvPackage")
{
// 维护队列长度不超过10
if (_TotalRecvPackage.Count >= 10)
{
_TotalRecvPackage.Dequeue();
}
_TotalRecvPackage.Enqueue(UDPPPP.TotalCount);
// 将队列转为逗号分隔字符串
string arrayString = string.Join(",", _TotalRecvPackage);
CSRedisCacheHelper.Set("UDPPackage_TotalRecvPackage", arrayString, 10);
}
if (UDPPPP.CommandType == "RCUOnLine")
{
// 维护队列长度不超过10
if (_RCUOnLine.Count >= 10)
{
_RCUOnLine.Dequeue();
}
_RCUOnLine.Enqueue(UDPPPP.TotalCount);
// 将队列转为逗号分隔字符串
string arrayString = string.Join(",", _RCUOnLine);
CSRedisCacheHelper.Set("RCUOnLine", arrayString, 10);
}
}
else if (k.Equals(KafkaKey.UDPPackageStepMonitor))
{
var collection = client.GetDatabase("udppackage").GetCollection<StepInfo_db>("zeroe_stepmonitor");
StepInfo UDPPPP = MyMessagePacker.FastDeserialize<StepInfo>(v);
StepInfo_db us = new StepInfo_db();
us.MessageId = UDPPPP.MessageId;
us.Step = UDPPPP.Step;
us.StepDescription = UDPPPP.StepDescription;
us.Content = UDPPPP.Content;
us.TriggerTime = DateTime.Parse(UDPPPP.TriggerTime);
us.TriggerTimeString = UDPPPP.TriggerTime;
us.EveryMessageId = UDPPPP.EveryMessageId;
us.Monitor_0E_01 = UDPPPP.Monitor_0E_01;
us.HostNumber = UDPPPP.HostNumber;
us.HotelCode = UDPPPP.HotelCode;
await collection.InsertOneAsync(us);
}
else if (k.Equals(KafkaKey.UDPPackagePowerMonitor))
{
var collection = client.GetDatabase("udppackage").GetCollection<NengHao_db>("powermonitor");
NengHao UDPPPP = MyMessagePacker.FastDeserialize<NengHao>(v);
NengHao_db us = new NengHao_db();
us.Version = UDPPPP.Version;
us.HotelCode = UDPPPP.HotelCode;
us.HostNumber = UDPPPP.HostNumber;
us.Mac = UDPPPP.Mac;
us.EndPoint = UDPPPP.EndPoint;
us.V = UDPPPP.V;
us.A = UDPPPP.A;
us.P = UDPPPP.P;
us.Energy_Consumption = UDPPPP.Energy_Consumption;
us.Sum_Energy_Consumption = UDPPPP.Sum_Energy_Consumption;
us.IsTakeCard = UDPPPP.IsTakeCard;
us.CreateTime = UDPPPP.CreateTime;
us.ReportTime = DateTime.Parse(UDPPPP.ReportTime);
us.RoomNumber = UDPPPP.RoomNumber;
us.CarbonVIP = UDPPPP.CarbonVIP;
us.AllDeviceData = UDPPPP.AllDeviceData;
us.IdentityInfo = UDPPPP.IdentityInfo;
await collection.InsertOneAsync(us);
}
else if (k.Equals(KafkaKey.IotMonitor))
{
var collection = client.GetDatabase("udppackage").GetCollection<IOTMonitorData_db>("voiceiotlog");
IOTMonitorData UDPPPP = MyMessagePacker.FastDeserialize<IOTMonitorData>(v);
IOTMonitorData_db us = new IOTMonitorData_db();
us.Step = UDPPPP.Step;
us.TriggerTime = UDPPPP.TriggerTime;
us.HotelCode = UDPPPP.HotelCode;
us.HotelId = UDPPPP.HotelId;
us.HotelName = UDPPPP.HotelName;
us.RoomNumber = UDPPPP.RoomNumber;
us.RequestId = UDPPPP.RequestId;
us.CommandDescription = UDPPPP.CommandDescription;
us.Platform = UDPPPP.Platform;
us.CreateTime = UDPPPP.CreateTime;
us.RemoteIP = UDPPPP.RemoteIP;
us.ControlClass = UDPPPP.ControlClass;
us.SceneName = UDPPPP.SceneName;
us.WhichOneDevice = UDPPPP.WhichOneDevice;
await collection.InsertOneAsync(us);
}
//
else if (k.Equals(KafkaKey.RCUOnLineStatus))
{
var collection = client.GetDatabase("udppackage").GetCollection<OnOffLineData_db>("rcustatuslog");
OnOffLineData UDPPPP = MyMessagePacker.FastDeserialize<OnOffLineData>(v);
OnOffLineData_db us = new OnOffLineData_db();
us.MAC = UDPPPP.MAC;
us.HostNumber = UDPPPP.HostNumber;
us.CurrentStatus = UDPPPP.CurrentStatus;
us.HotelCode = UDPPPP.HotelCode;
us.CurrentTime = UDPPPP.CurrentTime;
us.EndPoint = UDPPPP.EndPoint;
await collection.InsertOneAsync(us);
}
else if (k.Equals(KafkaKey.InvokceThirdHttpInterface))
{
var collection = client.GetDatabase("udppackage").GetCollection<Interface3Log_db>("invokehttpinterfacelog");
Interface3Log UDPPPP = MyMessagePacker.FastDeserialize<Interface3Log>(v);
Interface3Log_db us = new Interface3Log_db();
us.HotelCode = UDPPPP.HotelCode;
us.HostNumber = UDPPPP.HostNumber;
us.RoomNumber = UDPPPP.RoomNumber;
us.TriggerTime = UDPPPP.TriggerTime;
us.CommandType = UDPPPP.CommandType;
us.ActionData = UDPPPP.ActionData;
await collection.InsertOneAsync(us);
}
else if (k.Equals(KafkaKey.TakeCardStatus))
{
var collection = client.GetDatabase("udppackage").GetCollection<MTakeCardData_db>("takecardlog");
MTakeCardData UDPPPP = MyMessagePacker.FastDeserialize<MTakeCardData>(v);
MTakeCardData_db us = new MTakeCardData_db();
us.Status = UDPPPP.Status;
us.HostNUMBER = UDPPPP.HostNUMBER;
us.LastUpdateTime = UDPPPP.LastUpdateTime;
us.HotelCode = UDPPPP.HotelCode;
us.Status = UDPPPP.Status;
await collection.InsertOneAsync(us);
}
else if (k.Equals(KafkaKey.ServiceInfoStatus))
{
var collection = client.GetDatabase("udppackage").GetCollection<OtherServiceInfo_db>("serviceinfolog");
OtherServiceInfo UDPPPP = MyMessagePacker.FastDeserialize<OtherServiceInfo>(v);
OtherServiceInfo_db us = new OtherServiceInfo_db();
us.HotelCode = UDPPPP.HotelCode;
us.HostNumber = UDPPPP.HostNumber;
us.Address = UDPPPP.Address;
us.StatusReceiver = UDPPPP.StatusReceiver;
us.LastUpdateTime = UDPPPP.LastUpdateTime;
await collection.InsertOneAsync(us);
}
else if (k.Equals(KafkaKey.PMSLogMonitor))
{
//Console.WriteLine("收到了 PMSData");
var collection = client.GetDatabase("udppackage").GetCollection<CheckInYuanShidata_db>("pmslog");
CheckInYuanShidata UDPPPP = MyMessagePacker.FastDeserialize<CheckInYuanShidata>(v);
CheckInYuanShidata_db us = new CheckInYuanShidata_db();
us.Step = UDPPPP.Step;
us.IP = UDPPPP.IP;
us.RequestId = UDPPPP.RequestId;
us.CurrentTime = UDPPPP.CurrentTime;
us.CommandType = UDPPPP.CommandType;
us.JianJieData = UDPPPP.JianJieData;
us.ZhiJieData = UDPPPP.ZhiJieData;
await collection.InsertOneAsync(us);
}
#region 新版协议记录
else if (k.Equals(KafkaKey.RCUNewVersion_Register))
{
//P0~P3：子网掩码（4Byte）
//P4~P7：网关（4Byte）
//P8~P9：RCU端口（2Byte）
//P10~P15：Mac地址（6Byte）
//P16~P35：软件版本号（20Byte）
//P36~P38：配置版本号（3Byte）
//P39~P42：DNS服务器1 IP（4Byte）
//P43~P58：房型备注（16Byte）
//P59~P74：房号备注（16Byte）
//P75~P78：房型（4Byte）
//P79~P82：房号（4Byte）
var collection = client.GetDatabase("udppackage").GetCollection<NewVersionHexData_db>("rcu_hexdata");
NewVersionHexData UDPPPP = MyMessagePacker.FastDeserialize<NewVersionHexData>(v);
//byte[] hexdata = Tools.GetBytesFromString(UDPPPP.HexData);
//var 子网掩码 = hexdata[0..4];
//string subnet_mask = String.Join(".", 子网掩码);//子网掩码
//var 网关 = hexdata[4..8];
//string gateway = string.Join(".", 网关);
//var RCU端口 = hexdata[8..10];
//int lan_port = BitConverter.ToInt32(RCU端口);//局域网端口
//var MAC = hexdata[10..16];
//var 软件版本号 = hexdata[16..36];
//var 配置版本 = hexdata[36..39];
//var DNS服务器 = hexdata[39..43];
//var 房型备注 = hexdata[43..59];
//var 房号备注 = hexdata[59..75];
//var 房型 = hexdata[75..79];
//var 房号 = hexdata[79..83];
//int ipType = reader.ReadByte();//IP类型
//string type_number = Encoding.GetEncoding("GBK").GetString(reader.ReadBytes(16)).Replace(@"\u0000", "").Replace("", "").Trim();//机型编码
//string lan_ip = String.Join(".", reader.ReadBytes(4));//局域网IP
//string server_ip = String.Join(".", reader.ReadBytes(4));//服务器IP
//string subnet_mask = String.Join(".", reader.ReadBytes(4));//子网掩码
//string gateway = String.Join(".", reader.ReadBytes(4));//网关
//int lan_port = BitConverter.ToInt32(reader.ReadBytes(4), 0);//局域网端口
//string dns = String.Join(".", reader.ReadBytes(4));//DNS
//string software_version = Encoding.GetEncoding("GBK").GetString(reader.ReadBytes(20)).Replace(@"\u0000", "").Replace("", "").Trim();//软件版本号
NewVersionHexData_db cv = new NewVersionHexData_db();
cv.HotelCode = UDPPPP.HotelCode;
cv.HostNumber = UDPPPP.HostNumber;
cv.RoomNumber = UDPPPP.RoomNumber;
cv.CurrentTime = UDPPPP.CurrentTime;
cv.RemoteEndPoint = UDPPPP.RemoteEndPoint;
cv.CmdType = UDPPPP.CmdType;
cv.HexData = UDPPPP.HexData;
cv.CurrentTime = UDPPPP.CurrentTime;
await collection.InsertOneAsync(cv);
}
else if (k.Equals(KafkaKey.RCUNewVersion_RestartReason))
{
var collection = client.GetDatabase("udppackage").GetCollection<NewVersionHexData_db>("rcu_hexdata");
NewVersionHexData UDPPPP = MyMessagePacker.FastDeserialize<NewVersionHexData>(v);
NewVersionHexData_db cv = new NewVersionHexData_db();
cv.HotelCode = UDPPPP.HotelCode;
cv.HostNumber = UDPPPP.HostNumber;
cv.RoomNumber = UDPPPP.RoomNumber;
cv.CurrentTime = UDPPPP.CurrentTime;
cv.RemoteEndPoint = UDPPPP.RemoteEndPoint;
cv.CmdType = UDPPPP.CmdType;
cv.HexData = UDPPPP.HexData;
cv.CurrentTime = UDPPPP.CurrentTime;
await collection.InsertOneAsync(cv);
}
else if (k.Equals(KafkaKey.RCUNewVersion_0E))
{
var collection = client.GetDatabase("udppackage").GetCollection<NewVersionHexData_db>("rcu_hexdata");
NewVersionHexData UDPPPP = MyMessagePacker.FastDeserialize<NewVersionHexData>(v);
int length = 30 - 15 - 2;
using MemoryStream stream = new MemoryStream(null, 15, length);
NewVersionHexData_db cv = new NewVersionHexData_db();
cv.HotelCode = UDPPPP.HotelCode;
cv.HostNumber = UDPPPP.HostNumber;
cv.RoomNumber = UDPPPP.RoomNumber;
cv.CurrentTime = UDPPPP.CurrentTime;
cv.RemoteEndPoint = UDPPPP.RemoteEndPoint;
cv.CmdType = UDPPPP.CmdType;
cv.HexData = UDPPPP.HexData;
cv.CurrentTime = UDPPPP.CurrentTime;
await collection.InsertOneAsync(cv);
}
else if (k.Equals(KafkaKey.RCUNewVersion_TakeCard))
{
var collection = client.GetDatabase("udppackage").GetCollection<NewVersionHexData_db>("rcu_hexdata");
NewVersionHexData UDPPPP = MyMessagePacker.FastDeserialize<NewVersionHexData>(v);
NewVersionHexData_db cv = new NewVersionHexData_db();
cv.HotelCode = UDPPPP.HotelCode;
cv.HostNumber = UDPPPP.HostNumber;
cv.RoomNumber = UDPPPP.RoomNumber;
cv.CurrentTime = UDPPPP.CurrentTime;
cv.RemoteEndPoint = UDPPPP.RemoteEndPoint;
cv.CmdType = UDPPPP.CmdType;
cv.HexData = UDPPPP.HexData;
cv.CurrentTime = UDPPPP.CurrentTime;
await collection.InsertOneAsync(cv);
}
else if (k.Equals(KafkaKey.RCUNewTimerData))
{
}
else
{
}
#endregion
//Console.WriteLine($"消费者1收到消息: [分区{cr.Partition}] " +
// $"偏移量{cr.Offset}");
//默认是开着的
//c.Commit(cr);
}
catch (ConsumeException e)
{
logger.Error("出错：" + e.Message);
Console.WriteLine(111111111111111);
}
catch (Exception ex)
{
logger.Error("Ex出错：" + ex.Message);
Console.WriteLine(22222222222222);
}
}
}
catch (OperationCanceledException)
{
logger.Error("操作出错");
Console.WriteLine("操作错误");
}
catch (Exception ex)
{
Console.WriteLine("未知错误" + ex.Message);
logger.Error("未知错误" + ex.Message);
}
}
}
}
}

154
BLWLogServer/Services/KafkaConsume1.cs Normal file
View File

@@ -0,0 +1,154 @@
using System.Text;
using Common;
using CommonEntity;
using CommonTools;
using Confluent.Kafka;
using MongoDB.Driver;
using NLog;
using System.Collections.Generic;
using Confluent.Kafka.Admin;
using static Confluent.Kafka.ConfigPropertyNames;
using System.Configuration;
namespace BLWLogServer.Services
{
public class KafkaConsume1 : BackgroundService
{
public IConfiguration Configuration { get; set; }
public KafkaConsume1(IConfiguration configuration)
{
Configuration = configuration;
}
public static Logger logger = NLog.LogManager.GetCurrentClassLogger();
private async Task StartConsumer(CancellationToken stoppingToken)
{
string? ipport = Configuration["Kafka:EndPoint"];
string? user = Configuration["Kafka:UserName"];
string? pwd = Configuration["Kafka:PassWord"];
string? mongodbconnectstr = Configuration["Mongodb:Connectstr"];
while (!stoppingToken.IsCancellationRequested)
{
var conf = new ConsumerConfig
{
GroupId = "blwlogserver-consumer-group",
AutoOffsetReset = AutoOffsetReset.Earliest,
BootstrapServers = ipport,
SecurityProtocol = SecurityProtocol.SaslPlaintext,
SaslMechanism = SaslMechanism.Plain,
EnableAutoCommit = true,
SaslUsername = user,
SaslPassword = pwd
};
var c = new ConsumerBuilder<string, byte[]>(conf)
.SetErrorHandler((_, e) =>
{
logger.Error($"消费者错误: {e.Reason}");
}).SetPartitionsAssignedHandler((c, partitions) =>
{
logger.Error($"消费者分配到分区: {string.Join(", ", partitions)}");
}).Build();
//c.Subscribe(KafkaKey.BLWLog_RCU_Topic_Partition);
c.Subscribe(KafkaKey.BLWLog_RCU_Topic);
var connectionString = mongodbconnectstr;
var client = new MongoClient(connectionString);
try
{
while (true)
{
var consumeResult = c.Consume(stoppingToken);
try
{
var k = consumeResult.Message.Key;
var v = consumeResult.Message.Value;
if (k.Equals("testtest"))
{
Console.WriteLine("11111111111111111111");
}
if (k.Equals(KafkaKey.InvokceThirdHttpInterface))
{
var collection = client.GetDatabase("udppackage").GetCollection<Interface3Log_db>("invokehttpinterfacelog");
Interface3Log UDPPPP = MyMessagePacker.FastDeserialize<Interface3Log>(v);
Interface3Log_db us = new Interface3Log_db();
us.HotelCode = UDPPPP.HotelCode;
us.HostNumber = UDPPPP.HostNumber;
us.RoomNumber = UDPPPP.RoomNumber;
us.TriggerTime = UDPPPP.TriggerTime;
us.ActionData = UDPPPP.ActionData;
await collection.InsertOneAsync(us);
}
else if (k.Equals(KafkaKey.TakeCardStatus))
{
var collection = client.GetDatabase("udppackage").GetCollection<MTakeCardData_db>("takecardlog");
MTakeCardData UDPPPP = MyMessagePacker.FastDeserialize<MTakeCardData>(v);
MTakeCardData_db us = new MTakeCardData_db();
us.Status = UDPPPP.Status;
us.HostNUMBER = UDPPPP.HostNUMBER;
us.LastUpdateTime = UDPPPP.LastUpdateTime;
us.HotelCode = UDPPPP.HotelCode;
us.Status = UDPPPP.Status;
await collection.InsertOneAsync(us);
}
else if (k.Equals(KafkaKey.ServiceInfoStatus))
{
var collection = client.GetDatabase("udppackage").GetCollection<OtherServiceInfo_db>("serviceinfolog");
OtherServiceInfo UDPPPP = MyMessagePacker.FastDeserialize<OtherServiceInfo>(v);
OtherServiceInfo_db us = new OtherServiceInfo_db();
us.HotelCode = UDPPPP.HotelCode;
us.HostNumber = UDPPPP.HostNumber;
us.Address = UDPPPP.Address;
us.StatusReceiver = UDPPPP.StatusReceiver;
us.LastUpdateTime = UDPPPP.LastUpdateTime;
await collection.InsertOneAsync(us);
}
else
{
Console.WriteLine( "11111111111111111");
}
if (consumeResult.IsPartitionEOF)
{
Console.WriteLine($"消费者1到达分区末尾: {consumeResult.TopicPartitionOffset}");
}
//Console.WriteLine($"消费者1收到消息: [分区{consumeResult.Partition}] " +
// $"偏移量{consumeResult.Offset}: {consumeResult.Message.Value}");
// 手动提交偏移量
//c.Commit(consumeResult);
}
catch (ConsumeException e)
{
logger.Error(e.Message);
}
}
}
catch (OperationCanceledException)
{
c.Close();
}
}
}
protected async override Task ExecuteAsync(CancellationToken stoppingToken)
{
await Task.Factory.StartNew(async () =>
{
var consumers = new List<Task>();
for (int i = 0; i < 2; i++)
{
consumers.Add(Task.Run(() => StartConsumer(stoppingToken)));
}
await Task.WhenAll(consumers);
}, TaskCreationOptions.LongRunning);
await Task.CompletedTask;
}
}
}

246
BLWLogServer/Services/S.cs Normal file
View File

@@ -0,0 +1,246 @@
using System.Collections.Concurrent;
namespace BLWLogServer.Services
{
public class Device
{
/// <summary>
/// 设备地址
/// </summary>
public string Address { get; set; }
/// <summary>
/// 地址类型
/// </summary>
//public AddressType AddressType { get; set; }
/// <summary>
/// 设备开关状态：1开，2关，6停（窗帘），下发1个字节
/// </summary>
public byte Status { get; set; }
/// <summary>
/// 设备开关状态：1开，2关，6停（窗帘）,接收2个字节
/// </summary>
public ushort StatusReceiver { get; set; }
/// <summary>
/// 亮度值
/// </summary>
public byte Brightness { get; set; }
/// <summary>
/// 温度
/// </summary>
public byte Temperature { get; set; }
/// <summary>
/// 风速
/// </summary>
public byte FanSpeed { get; set; }
/// <summary>
/// 模式
/// </summary>
public byte Mode { get; set; }
/// <summary>
/// 阀门开关
/// </summary>
public byte Valve { get; set; }
/// <summary>
/// 空调执行方式和内容
/// </summary>
public Int32 AirExecMode { get; set; }
/// <summary>
/// 地暖执行方式和内容
/// </summary>
public Int32 FloorHotExecMode { get; set; }
/// <summary>
/// 背景音乐执行方式和内容
/// </summary>
public Int32 MusicExecMode { get; set; }
/// <summary>
/// 色温执行方式和内容
/// </summary>
//public Int32 ColorTempExecMode { get; set; }
}
public class Status
{
/// <summary>
/// RCU是否锁定
/// </summary>
public bool SysLock { get; set; }
/// <summary>
/// 房卡类型
/// </summary>
public byte CardType { get; set; }
/// <summary>
/// 房门开关
/// </summary>
public bool Door { get; set; }
/// <summary>
/// 电量
/// </summary>
public ushort ElecQty { get; set; }
/// <summary>
/// 主机温度
/// </summary>
public byte HostTemp { get; set; }
/// <summary>
/// 空调状态
/// </summary>
public ConcurrentDictionary<byte, AirConditionStatus> AirConditions = new ConcurrentDictionary<byte, AirConditionStatus>();
/// <summary>
/// 设备状态
/// </summary>
public ConcurrentDictionary<string, Device> Devices = new ConcurrentDictionary<string, Device>();
/// <summary>
/// 故障列表
/// </summary>
public ConcurrentDictionary<string, FaultStaus> Faults = new ConcurrentDictionary<string, FaultStaus>();
}
/// <summary>
/// 空调状态
/// </summary>
public class AirConditionStatus
{
/// <summary>
/// 空调号
/// </summary>
public byte AirNo { get; set; }
/// <summary>
/// 当前温度
/// </summary>
public int CurrentTemp { get; set; }
/// <summary>
/// 设定温度
/// </summary>
public int SettingTemp { get; set; }
/// <summary>
/// 风机开关
/// </summary>
public bool OnOff { get; set; }
/// <summary>
/// 风速：0/停止, 1/低速, 2/中速, 3/高速, 4/自动
/// </summary>
public int Speed { get; set; }
/// <summary>
/// 模式: 0/制冷，1/制热，2/送风，3/除湿
/// </summary>
public int Mode { get; set; }
/// <summary>
/// 阀状态： 1/阀开，2/阀关
/// </summary>
public int Valve { get; set; }
/// <summary>
/// 补偿温度，范围-6.0~6.0
/// </summary>
public float CompensatoryTemp { get; set; }
/// <summary>
/// 保温温度
/// </summary>
public int KeepTemp { get; set; }
/// <summary>
/// 冷热模式：0/手动，1/自动
/// </summary>
public int ColdHotMode { get; set; }
/// <summary>
/// 死区温度
/// </summary>
public int DeadTemp { get; set; }
/// <summary>
/// 热偏差
/// </summary>
public int HotDevition { get; set; }
/// <summary>
/// 冷偏差
/// </summary>
public int ColdDevition { get; set; }
}
public class FaultStaus
{
/// <summary>
/// 故障号
/// </summary>
public string FaultNo { get; set; }
/// <summary>
/// 类型：1在线状态（0在线，1离线），2电量（0~100%），3电流，4 1901故障检测次数，5设备单个回路状态（0正常，1损坏）
/// </summary>
public byte Type { get; set; }
/// <summary>
/// 内容
/// </summary>
public byte Data { get; set; }
}
public class S
{
public static Status NewStatusParse(Stream stream)
{
Status roomStatus = new Status();
try
{
using (BinaryReader reader = new BinaryReader(stream))
{
roomStatus.SysLock = reader.ReadBoolean(); //RCU是否锁定：0/否，1/是
roomStatus.CardType = reader.ReadByte(); //房卡类型：0/无人，1/有人，2/客人，3/经理，4/服务员
roomStatus.Door = reader.ReadBoolean(); //门磁开关：1/开，2/关
roomStatus.ElecQty = reader.ReadUInt16(); //门锁电量，单位：MV
roomStatus.HostTemp = reader.ReadByte(); //主机温度
//空调数量，默认0，新的回路方式，此处无用，兼容老版本
int airConditionNumber = reader.ReadByte();
for (int i = 0; i < airConditionNumber; i++)
{
byte airNo = reader.ReadByte();
if (!roomStatus.AirConditions.ContainsKey(airNo))
{
roomStatus.AirConditions[airNo] = new AirConditionStatus();
}
roomStatus.AirConditions[airNo].AirNo = airNo;
roomStatus.AirConditions[airNo].CurrentTemp = reader.ReadSByte();
roomStatus.AirConditions[airNo].SettingTemp = reader.ReadSByte();
roomStatus.AirConditions[airNo].OnOff = reader.ReadBoolean();
roomStatus.AirConditions[airNo].Speed = reader.ReadByte();
roomStatus.AirConditions[airNo].Mode = reader.ReadByte();
roomStatus.AirConditions[airNo].Valve = reader.ReadByte();
roomStatus.AirConditions[airNo].CompensatoryTemp = reader.ReadByte();
roomStatus.AirConditions[airNo].KeepTemp = reader.ReadByte();
roomStatus.AirConditions[airNo].ColdHotMode = reader.ReadByte();
roomStatus.AirConditions[airNo].DeadTemp = reader.ReadByte();
roomStatus.AirConditions[airNo].HotDevition = reader.ReadByte();
roomStatus.AirConditions[airNo].ColdDevition = reader.ReadByte();
}
//设备状态数量，包含所有回路状态
int deviceNumber = reader.ReadByte();
long originalPosition = reader.BaseStream.Position;
// 读取前先记录当前位置
try
{
//0x19数据缓冲区异常问题处理
//0x19 取电数据不处理
//非0x19 取电数据处理，但是不触发 欢迎词
for (int i = 0; i < deviceNumber; i++)
{
var QA = reader.ReadBytes(4);
var Status = reader.ReadUInt16();
string address = String.Format("{0:000}{1:000}{2:000}", QA[0], QA[1], QA[0]);
if (!roomStatus.Devices.ContainsKey(address))
{
roomStatus.Devices[address] = new Device();
}
roomStatus.Devices[address].Address = address;//设备地址
roomStatus.Devices[address].StatusReceiver = Status;//设备状态改为2个字节,Modified By 20181213
originalPosition = reader.BaseStream.Position;
}
}
catch (Exception)
{
// 发生异常时回退到原始位置
reader.BaseStream.Position = originalPosition;
}
return roomStatus;
}
}
catch (Exception)
{
return null;
}
}
}
}