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

增加几个功能

Browse Source
This commit is contained in:
TianMaiCheng
2026-03-13 15:41:13 +08:00
parent 02d6778e35
commit f4d67cb673
15 changed files with 250 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
BLWLogProduce/Models/MyPublishRedis.cs
View File

@@ -33,6 +33,8 @@ namespace BLWLogProduce.Models
{
var o = JsonConvert.DeserializeObject<OnOffLineData>(data_take);
o.CurrentStatus = "off";
o.CurrentTime = DateTime.Now;
o.UnixTime = DateTimeOffset.Now.ToUnixTimeSeconds();
string str = Newtonsoft.Json.JsonConvert.SerializeObject(o);
redis6.Publish("redis-on_off_line", str);

89
BLWLogProduce/Services/KafkaProduce.cs
View File

@@ -371,11 +371,15 @@ namespace BLWLogProduce.Services
{
return;
}
if (poo.NengHaoList == null || poo.NengHaoList.Count == 0)
{
return;
}
//logger.Error("能耗：" + body);
//2026-03-04 麦工说，要重新启用能耗数据
//if (poo.NengHaoList == null || poo.NengHaoList.Count == 0)
//{
// return;
//}
//if (poo.HotelCode == 2302)
//{
// logger.Error("能耗：" + body);
//}
//string str= Newtonsoft.Json.JsonConvert.SerializeObject(poo);
//Console.WriteLine("收到了"+str);
//byte[] qf = MyMessagePacker.FastSerialize(poo);
@@ -442,12 +446,15 @@ namespace BLWLogProduce.Services
ese.CardEvent = poo.CardEvent;
ese.PMSStatus = poo.PMS_Status;
ese.BrightG = poo.Bright_G;
ese.Version = poo.Version;
byte[] data = ese.ToByteArray();
string TopicKey1 = KafkaKey.BLWLog4BaoJing_RCU_Topic;
string DetailKey1 = KafkaKey.UDPPackagePowerMonitor;
if (poo.HotelCode == 1085)
if (poo.HotelCode == 1003)
{
logger.Error("能耗数据:" + body);
}
@@ -515,6 +522,14 @@ namespace BLWLogProduce.Services
{
NewVersionLog? poo = System.Text.Json.JsonSerializer.Deserialize<NewVersionLog>(body);
poo.ts_ms = Tools.GetUnixTime_MS();
if (poo.hotel_id.Equals("2144")&&poo.room_id.Equals("1006")||poo.room_id.Equals("1007"))
{
logger.Error("漏数据监控:"+poo.room_id+"##"+poo.comm_seq);
}
if (poo.hotel_id.Equals("1085")&&poo.room_id.Equals("8099"))
{
logger.Error("漏数据监控:"+poo.room_id+"##"+poo.comm_seq);
}
if (string.IsNullOrEmpty(poo.hotel_id) || string.IsNullOrEmpty(poo.device_id))
{
@@ -535,6 +550,51 @@ namespace BLWLogProduce.Services
}));
var TSLog_DingYue_XinTiao = ("redis-heartbeat", new Action<SubscribeMessageEventArgs>(async (args) =>
{
string body = args.Body;
try
{
Kafka_XinTiao? poo = System.Text.Json.JsonSerializer.Deserialize<Kafka_XinTiao>(body);
string TopicKey1 = KafkaKey.Blwlog4Nodejs_oldrcu_heartbeat_topic;
string DetailKey1 = poo.hotel_id;
var jsonstr = JsonConvert.SerializeObject(poo);
await p.ProduceAsync(TopicKey1, new Message<string, byte[]> { Key = DetailKey1, Value = Encoding.UTF8.GetBytes(jsonstr) });
}
catch (Exception ex)
{
Console.WriteLine(ex.Message);
Console.WriteLine(ex.StackTrace);
}
}));
//注册0xB1
var TSLog_DingYue_ZhuCe = ("redis-0XB1", new Action<SubscribeMessageEventArgs>(async (args) =>
{
string body = args.Body;
try
{
RegisterInfo? poo = System.Text.Json.JsonSerializer.Deserialize<RegisterInfo>(body);
string TopicKey1 = KafkaKey.blwlog4Nodejs_rcu_register_topic;
string DetailKey1 =poo.device_id;
var jsonstr = JsonConvert.SerializeObject(poo);
await p.ProduceAsync(TopicKey1, new Message<string, byte[]> { Key = DetailKey1, Value = Encoding.UTF8.GetBytes(jsonstr) });
}
catch (Exception ex)
{
Console.WriteLine(ex.Message);
Console.WriteLine(ex.StackTrace);
}
}));
var TSLog_DingYue_0X36 = ("redis-0X36-0X0F", new Action<SubscribeMessageEventArgs>(async (args) =>
{
string body = args.Body;
@@ -548,7 +608,7 @@ namespace BLWLogProduce.Services
{
return;
}
if (poo.cmd_word.Equals("0E")&&poo.fault_list.Count==0&&poo.device_list.Count==0)
if (poo.cmd_word.Equals("0E") && poo.fault_list.Count == 0 && poo.device_list.Count == 0)
{
return;
}
@@ -674,15 +734,15 @@ namespace BLWLogProduce.Services
usa.UnixTime = Tools.GetUnixTime_MS();
if (string.IsNullOrEmpty(usa.EndPoint))
{
logger.Error("RCUOnOffLine:" + body);
//logger.Error("RCUOnOffLine:" + body);
return;
}
if (string.IsNullOrEmpty(usa.HostNumber))
{
logger.Error("RCUOnOffLine:" + body);
//logger.Error("RCUOnOffLine:" + body);
return;
}
string Key1 = KafkaKey.BLWLog4NodeJs_RCU_OnOffLine_Topic;
string Key2 = usa.HostNumber;
@@ -770,6 +830,7 @@ namespace BLWLogProduce.Services
{
string body = args.Body;
logger.Error("TCL：" + body);
Interface3Log? usa = System.Text.Json.JsonSerializer.Deserialize<Interface3Log>(body);
byte[] bytes = MyMessagePacker.FastSerialize(usa);
@@ -779,7 +840,7 @@ namespace BLWLogProduce.Services
//var partition = new Partition(1); // 指定分区号
//var topicPartition = new TopicPartition(TopicKey, partition);
var dr = await p.ProduceAsync(TopicKey, new Message<string, byte[]> { Key = DetailKey, Value = bytes });
//var dr = await p.ProduceAsync(TopicKey, new Message<string, byte[]> { Key = DetailKey, Value = bytes });
}
catch (Exception ex)
{
@@ -807,8 +868,8 @@ namespace BLWLogProduce.Services
CurrentStatus = usa.CurrentStatus,
CurrentTime = usa.CurrentTime,
UnixTime = Tools.GetUnixTime_MS(),
LauncherVersion= usa.LauncherVersion,
RebootReason= usa.RebootReason
LauncherVersion = usa.LauncherVersion,
RebootReason = usa.RebootReason
};
byte[] bytes = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(rrr));
@@ -980,6 +1041,8 @@ namespace BLWLogProduce.Services
CSRedisCacheHelper.redis3.Subscribe(TSLog_DingYue);
CSRedisCacheHelper.redis3.Subscribe(TSLog_DingYue_0X36);
CSRedisCacheHelper.redis3.Subscribe(DingYue11);
CSRedisCacheHelper.redis3.Subscribe(TSLog_DingYue_XinTiao);
CSRedisCacheHelper.redis3.Subscribe(TSLog_DingYue_ZhuCe);
}
catch (Exception ex)
{