J1147
/
GpsCardGatewayPosition
Слідкувати 1
В обрані 0
Форк 0
Код Проблеми 0 Запити на злиття 0 Релізи 0 Вікі Активність
Ви не можете вибрати більше 25 тем Теми мають розпочинатися з літери або цифри, можуть містити дефіси (-) і не повинні перевищувати 35 символів.
13 Коміти
2 Гілки
227KB
Гілка: develop
Гілки Теги
${ item.name }
Створити гілку ${ searchTerm }
з 'develop'
${ noResults }
GpsCardGatewayPosition/GpsCardGatewayPosition.Postion/Worker.cs

1034 lines
44KB
Неформатований Постійне посилання Blame Історія 

  1. using Confluent.Kafka;

  2. using GpsCardGatewayPosition.Common;

  3. using GpsCardGatewayPosition.Model.Config;

  4. using GpsCardGatewayPosition.Model.Enum;

  5. using GpsCardGatewayPosition.Model.IoT;

  6. using GpsCardGatewayPosition.Service.Cache;

  7. using GpsCardGatewayPosition.Service.MqProducer;

  8. using GpsCardGatewayPosition.Service.MqProducer.Model;

  9. using GpsCardGatewayPosition.Service.Resolver.Factory;

  10. using GpsCardGatewayPosition.Service.Resolver.Property;

  11. using GpsCardGatewayPosition.Service.Resolver.Property.Dto;

  12. using Microsoft.Extensions.Hosting;

  13. using Microsoft.Extensions.Logging;

  14. using Microsoft.Extensions.Options;

  15. using Newtonsoft.Json;

  16. using Newtonsoft.Json.Linq;

  17. using System;

  18. using System.Collections.Generic;

  19. using System.Linq;

  20. using System.Net;

  21. using System.Reflection.Metadata;

  22. using System.Text;

  23. using System.Threading.Tasks;

  24. using System.Web;

  25. using static Confluent.Kafka.ConfigPropertyNames;

  26. 

  27. namespace GpsCardGatewayPosition.Postion

  28. {

  29.     public class Worker : BackgroundService

  30.     {

  31.         private readonly ILogger<Worker> _logger;

  32.         private CancellationTokenSource _tokenSource = default!;

  33.         private readonly MqProcessLogic _serviceMqProcess;

  34.         private readonly ServiceConfig _configService;

  35.         private readonly AppsettingsConfig _configAppsettings;

  36.         private readonly ResolverFactory _factoryResolver;

  37. 

  38. 

  39. 

  40.         private readonly WifiPositionResolver _resolverWifiPosition;

  41.         private readonly WifiPlus2PositionResolver _resolverWifiPlus2Position;

  42.         private readonly GpsPositionResolver _resolverGpsPosition;  

  43.         private readonly LbsGsmPositionResolver _resolverLbsGsmPosition;

  44. 

  45.         public Worker(ILogger<Worker> logger,IOptions<ServiceConfig> _optConfigService, IOptions<RedisConfig> optConfigRedis, IOptions<AppsettingsConfig> optConfigAppsettings, ResolverFactory resolverFactory, MqProcessLogic serviceMqProcess, 

  46.             WifiPositionResolver wifiPositionResolver, WifiPlus2PositionResolver wifiPlus2PositionResolver, GpsPositionResolver gpsPositionResolver,LbsGsmPositionResolver lbsGsmPositionResolver)

  47.         {

  48.             _configService = _optConfigService.Value;

  49.             _configAppsettings = optConfigAppsettings.Value;

  50.             _serviceMqProcess = serviceMqProcess;   

  51.             _logger = logger;

  52.             _factoryResolver = resolverFactory;

  53. 

  54. 

  55.             _resolverWifiPosition = wifiPositionResolver;

  56.             _resolverWifiPlus2Position = wifiPlus2PositionResolver;

  57.             _resolverGpsPosition = gpsPositionResolver;

  58.             _resolverLbsGsmPosition= lbsGsmPositionResolver;

  59. 

  60.             //配置全局服务

  61.             #region  其他业务redis服务器

  62.             var csredis = new CSRedis.CSRedisClient(optConfigRedis.Value.ToString());

  63.             RedisHelper.Initialization(csredis);

  64. 

  65.             RedisConfig csredisDb7Con = optConfigRedis.Value;

  66.             csredisDb7Con.DefaultDatabase = 7;

  67.             csredisDb7Con.Prefix = "TELPO";

  68.             var csredisDb7 = new CSRedis.CSRedisClient(csredisDb7Con.ToString());

  69.             RedisHelperDb7.Initialization(csredisDb7);

  70. 

  71.             //  

  72.             RedisConfig csredisDb0Con = optConfigRedis.Value;

  73.             csredisDb0Con.DefaultDatabase = 0;

  74.             csredisDb0Con.Prefix = "";

  75.             var csredisDb0 = new CSRedis.CSRedisClient(csredisDb0Con.ToString());

  76.             RedisHelperDb0.Initialization(csredisDb0);

  77.             #endregion

  78. 

  79.             #region 共享维智定位缓存 

  80.             // 维智定位缓存Redis Server

  81.             // 172.19.42.56,47.100.200.105

  82.             RedisConfig csredisWayzCon = optConfigRedis.Value;

  83.             csredisWayzCon.DefaultDatabase = 10;

  84.             csredisDb7Con.Prefix = "_GW_";

  85. #if DEBUG

  86.             csredisWayzCon.Server = "192.168.12.127:8090";

  87. #else

  88.             csredisWayzCon.Server = "172.19.42.56:8090";

  89. #endif

  90.             var csredisWayz = new CSRedis.CSRedisClient(csredisWayzCon.ToString());

  91.             RedisHelperWayz.Initialization(csredisWayz);

  92. #endregion

  93.         }

  94.         public override Task StartAsync(CancellationToken cancellationToken)

  95.         {

  96.             _logger.LogInformation("------StartAsync");

  97.             _tokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);

  98.             // 创建消费者

  99.             return base.StartAsync(cancellationToken);

  100.         }

  101. 

  102.         public override Task StopAsync(CancellationToken cancellationToken)

  103.         {

  104.             _logger.LogInformation("------StopAsync");

  105.             _tokenSource.Cancel();

  106.             return base.StopAsync(cancellationToken);

  107.         }

  108.         protected override async Task ExecuteAsync(CancellationToken stoppingToken)

  109.         {

  110.             var kafkaConsumer = CreateKafkaConsumer();

  111.             kafkaConsumer.Subscribe("topics.storage.iot.postion");

  112.             var tasks = new List<Task>();

  113.             //int i = 0;

  114.             try

  115.             {

  116.                 while (!stoppingToken.IsCancellationRequested)

  117.                 {

  118.                     #region 生产

  119. 

  120.                     // await productPostionAsync();

  121. 

  122.                     #endregion

  123. 

  124.                     #region  消费

  125.                     // 并发处理 1000 个消息

  126.                     var concurrencyId = Guid.NewGuid().ToString("N")[^4..];

  127.                     using (_logger.BeginScope(new Dictionary<string, object> { ["RequestId"] = concurrencyId }))

  128.                     using (new CustomizeStopWatch(nameof(Worker), _logger))

  129.                     {

  130.                         //for (int i = 0; i < 2000; i++)

  131.                         //{

  132.                         //    //var consumeResult = kafkaConsumer.Consume(stoppingToken);

  133.                         //    var consumeResult = await Task.Run(() => kafkaConsumer.Consume(stoppingToken));

  134.                         //    if (consumeResult != null)

  135.                         //    {

  136.                         //        tasks.Add(ProcessMessageAsync(consumeResult, kafkaConsumer));

  137.                         //    }

  138.                         //}

  139.                         ////var consumeResult = kafkaConsumer.Consume(stoppingToken);

  140.                         ////if (consumeResult != null)

  141.                         ////{

  142.                         ////    tasks.Add(ProcessMessageAsync(consumeResult, kafkaConsumer));

  143.                         ////}

  144.                         //// 并行等待所有任务完成

  145.                         //await Task.WhenAll(tasks);

  146.                         //tasks.Clear(); // 清空任务列表，以便下一个循环重新填充

  147.                         //Console.WriteLine("----------------------------------------------------------");

  148. 

  149.                         var consumeResult = await Task.Run(() => kafkaConsumer.Consume(stoppingToken));

  150.                         if (consumeResult != null)

  151.                         {

  152.                             tasks.Add(ProcessMessageAsync(consumeResult, kafkaConsumer, concurrencyId));

  153.                         }

  154. 

  155.                         if (tasks.Count>=2000)

  156.                         {

  157.                             // 并行等待所有任务完成

  158.                             await Task.WhenAll(tasks);

  159.                             tasks.Clear(); // 清空任务列表，以便下一个循环重新填充

  160.                             Console.WriteLine("----------------------------------------------------------");

  161.                         }

  162. 

  163. 

  164. 

  165.                     }

  166.                     //var consumeResult = kafkaConsumer.Consume(stoppingToken);

  167.                     //if (consumeResult != null)

  168.                     //{

  169.                     //    i++;

  170.                     //    tasks.Add(ProcessMessageAsync(consumeResult, kafkaConsumer, i));

  171.                     //}

  172.                     //await Task.WhenAll(tasks);

  173.                     #endregion

  174.                 }

  175.             }

  176.             catch (OperationCanceledException)

  177.             {

  178.                 _logger.LogWarning("Worker exit");

  179.             }

  180.             catch (Exception ex)

  181.             {

  182.                 _logger.LogError($"An error occurred: {ex.Message}");

  183.             }

  184. 

  185.            // await Task.WhenAll(tasks);

  186.         }

  187. 

  188.         private async Task ProductPostionAsync()

  189.         {

  190.             Object wifi = JObject.Parse(@"{

  191. 		'MessageId': '1775412145908567552',

  192. 		'BusinessId': '862622050169778',

  193. 		'ReceiveTime': '2024-04-03 14:36:55',

  194. 		'TopicType': 512,

  195. 		'MsgType': 517,

  196. 		'TopicInfo': {

  197. 			'deviceType': 'CustomCategory',

  198. 			'iotId': '7EBPsoxlNmILKsUpKl0L000000',

  199. 			'requestId': '987',

  200. 			'productKey': 'a18mXM6Cvx8',

  201. 			'gmtCreate': 1712126215803,

  202. 			'deviceName': '862622050169778',

  203. 			'items': {

  204. 				'WIFI': {

  205. 					'value': {

  206. 						'mmac': '04:6b:25:8e:a1:81,-68,',

  207. 						'macs': 'f4:2a:7d:ca:c8:c9,-72,|04:6b:25:8e:0a:91,-75,|54:84:dc:37:e8:70,-78,|54:84:dc:37:e8:18,-82,',

  208. 						'IDType': 0,

  209. 						'IDNumber': ''

  210. 					},

  211. 					'time': 1712126215802

  212. 				}

  213. 			}

  214. 		},

  215. 		'DetailData': {

  216. 			'time': 1712126215802,

  217. 			'value': {

  218. 				'mmac': '04:6b:25:8e:a1:81,-68,',

  219. 				'macs': 'f4:2a:7d:ca:c8:c9,-72,|04:6b:25:8e:0a:91,-75,|54:84:dc:37:e8:70,-78,|54:84:dc:37:e8:18,-82,',

  220. 				'IDType': 0,

  221. 				'IDNumber': ''

  222. 			}

  223. 		}

  224. 	}");

  225. 

  226.             Object wifiplus2 = JObject.Parse(@"{

  227. 		'MessageId': '1775412188354926080',

  228. 		'BusinessId': '861281060094145',

  229. 		'ReceiveTime': '2024-04-03 14:37:05',

  230. 		'TopicType': 512,

  231. 		'MsgType': 514,

  232. 		'TopicInfo': {

  233. 			'deviceType': 'CustomCategory',

  234. 			'iotId': 'yQw7bqBl8gSltErYIRVBg1gs00',

  235. 			'requestId': '105',

  236. 			'productKey': 'a18mXM6Cvx8',

  237. 			'gmtCreate': 1712126225922,

  238. 			'deviceName': '861281060094145',

  239. 			'items': {

  240. 				'WIFIPLUS2_0': {

  241. 					'time': 1712124640009,

  242. 					'value': [

  243. 						{

  244. 							'bts': '460,00,30977,247196817,-41',

  245. 							'smac': '',

  246. 							'nearbts': '|460,00,30977,130,-17|460,00,30977,436,-12|460,00,30977,414,-8|460,00,30977,500,-41|460,00,30977,294,-47|460,00,30977,2,-39',

  247. 							'imsi': '460077277322956',

  248. 							'steps': '0',

  249. 							'IDType': 0,

  250. 							'network': 'LTE',

  251. 							'mmac': '08:10:7b:df:3d:c8,-62,',

  252. 							'datetime': '1712124640001',

  253. 							'macs': 'ee:b9:70:83:8d:d2,-68,|3c:06:a7:16:a7:21,-68,|3c:06:a7:80:68:1a,-70,|3e:06:a7:a0:68:1a,-70,|08:10:78:91:72:4a,-77,|68:77:24:aa:40:c3,-81,|a8:e5:44:a5:c4:20,-84,',

  254. 							'imei': '861281060094145',

  255. 							'cdma': 0,

  256. 							'IDNumber': ''

  257. 						}

  258. 					]

  259. 				}

  260. 			}

  261. 		},

  262. 		'DetailData': {

  263. 			'time': 1712124640009,

  264. 			'value': [

  265. 				{

  266. 					'Mmac': '08:10:7b:df:3d:c8,-62,',

  267. 					'Macs': 'ee:b9:70:83:8d:d2,-68,|3c:06:a7:16:a7:21,-68,|3c:06:a7:80:68:1a,-70,|3e:06:a7:a0:68:1a,-70,|08:10:78:91:72:4a,-77,|68:77:24:aa:40:c3,-81,|a8:e5:44:a5:c4:20,-84,',

  268. 					'IdType': 0,

  269. 					'IdNumber': '',

  270. 					'Imei': '861281060094145',

  271. 					'Smac': '',

  272. 					'Imsi': '460077277322956',

  273. 					'NearBts': '|460,00,30977,130,-17|460,00,30977,436,-12|460,00,30977,414,-8|460,00,30977,500,-41|460,00,30977,294,-47|460,00,30977,2,-39',

  274. 					'Bts': '460,00,30977,247196817,-41',

  275. 					'Cdma': 0,

  276. 					'Network': 'LTE',

  277. 					'Steps': '0',

  278. 					'Datetime': '1712124640001'

  279. 				}

  280. 			]

  281. 		}

  282. 	}");

  283. 

  284.             //Object lbs = JObject.Parse(@');

  285. 

  286.             //Object gps = JObject.Parse(@');

  287.             for (int i = 0; i < 1000; i++)

  288.             {

  289.                 if (i % 2 == 1)

  290.                 {

  291.                     var serialno = "862622050169778";

  292.                     var messageId = Guid.NewGuid().ToString();

  293.                     await _serviceMqProcess.ProcessIotPositionAsync(messageId, serialno, wifi);

  294.                 }

  295.                 else

  296.                 {

  297.                     var serialno = "861281060094145";

  298.                     var messageId = Guid.NewGuid().ToString();

  299.                     await _serviceMqProcess.ProcessIotPositionAsync(messageId, serialno, wifiplus2);

  300.                 }

  301.             }

  302.         }

  303.         private async Task ProcessMessageAsync(ConsumeResult<Ignore, string> consumeResult, IConsumer<Ignore, string> kafkaConsumer,string concurrencyId)

  304.         {

  305.             var msg = JsonConvert.DeserializeObject<MessageModel>(consumeResult.Message.Value);

  306.             string msgId = msg!.Data["MessageId"]?.ToObject<string>()!;

  307.             string bizId = msg!.Data["BusinessId"]?.ToObject<string>()!;

  308.             IotTopicType topic = msg!.Data["TopicType"]!.ToObject<IotTopicType>();

  309.             int type = msg!.Data["MsgType"]!.ToObject<int>();

  310.             object topicInfo = msg!.Data["TopicInfo"]!.ToObject<object>()!;

  311.             object detailData = msg!.Data["DetailData"]!.ToObject<object>()!;

  312.             var packge = new PackageMsgModel(msgId, bizId, topic, type, topicInfo, detailData);

  313. 

  314.             try

  315.             {

  316.                 using (_logger.BeginScope(new Dictionary<string, object> { ["RequestId"] =$"{concurrencyId}-{msg!.MessageId}"  }))

  317.                 using (new CustomizeStopWatch(IotMessageTypeUtils.TryToDescription(type), _logger))

  318.                 {

  319.                     var dict = JsonConvert.DeserializeObject<Dictionary<string, object>>(msg!.Data["TopicInfo"]?["items"]?.ToString()!);

  320.                     var key = dict?.Keys?.FirstOrDefault() + "";

  321.                     switch (key)

  322.                     {

  323.                         case "GPS":     //处理gps位置信息上报

  324.                             {

  325.                                 //Console.WriteLine("GPS");

  326.                                 var data = JsonConvert.DeserializeObject<PropertyItemModel<GpsInfoModel>>(dict?["GPS"] + "");

  327.                                 //if (Utils.GetTimeDeviationMilliseconds(data!.Time) > _configAppsettings.DeviceTimeDeviationMillieconds) throw new ArgumentException($"设备消息时间戳异常[{data.Time}]");

  328.                                 _resolverGpsPosition.SetResolveInfo(packge);

  329.                                 await _resolverGpsPosition.ExecuteMessageAsync().ConfigureAwait(false);

  330.                                 kafkaConsumer.Commit(consumeResult);

  331.                                 break;

  332.                                 //return new PackageMsgModel(msg.MessageId, topicInfo.DeviceName, IotTopicType.Property, (int)IotMessagePropertyType.GpsPosition, topicInfo, data);

  333.                             }

  334. 

  335.                         case "WIFI":     //处理gps位置信息上报

  336.                             {

  337.                                 var data = JsonConvert.DeserializeObject<PropertyItemModel<WifiInfoModel>>(dict?["WIFI"] + "");

  338.                                 //if (Utils.GetTimeDeviationMilliseconds(data!.Time) > _configAppsettings.DeviceTimeDeviationMillieconds) throw new ArgumentException($"设备消息时间戳异常[{data.Time}]");

  339.                                 _resolverWifiPosition.SetResolveInfo(packge);

  340.                                 await _resolverWifiPosition.ExecuteMessageAsync().ConfigureAwait(false);

  341.                                 kafkaConsumer.Commit(consumeResult);

  342.                                 break;

  343.                             }

  344.                         case "WIFIPLUS2_0":     //处理gps位置信息上报

  345.                             {

  346.                                 var data = JsonConvert.DeserializeObject<PropertyItemModel<List<EachWifiPlus2Model>>>(dict?["WIFIPLUS2_0"] + "");

  347.                                 //if (Utils.GetTimeDeviationMilliseconds(data!.Time) > _configAppsettings.DeviceTimeDeviationMillieconds) throw new ArgumentException($"设备消息时间戳异常[{data.Time}]");

  348.                                 _resolverWifiPlus2Position.SetResolveInfo(packge);

  349.                                 await _resolverWifiPlus2Position.ExecuteMessageAsync().ConfigureAwait(false);

  350.                                 kafkaConsumer.Commit(consumeResult);

  351.                                 break;

  352.                             }

  353.                         case "LBS":     //处理gps位置信息上报

  354.                             {

  355.                                 var data = JsonConvert.DeserializeObject<PropertyItemModel<LbsInfoModel>>(dict?["LBS"] + "");

  356.                                //if (Utils.GetTimeDeviationMilliseconds(data!.Time) > _configAppsettings.DeviceTimeDeviationMillieconds) throw new ArgumentException($"设备消息时间戳异常[{data.Time}]");

  357. 

  358.                                 //if (data.Value.Cdma == 0)

  359.                                 //{

  360.                                 //    return new PackageMsgModel(msg.MessageId, topicInfo.DeviceName, IotTopicType.Property, (int)IotMessagePropertyType.LbsGsmPosition, topicInfo, data);

  361.                                 //}

  362.                                 //else if (data.Value.Cdma == 1)

  363.                                 //{

  364.                                 //    return new PackageMsgModel(msg.MessageId, topicInfo.DeviceName, IotTopicType.Property, (int)IotMessagePropertyType.LbsCdmaPosition, topicInfo, data);

  365.                                 //}

  366.                                 _resolverLbsGsmPosition.SetResolveInfo(packge);

  367.                                 await _resolverLbsGsmPosition.ExecuteMessageAsync().ConfigureAwait(false);

  368.                                 kafkaConsumer.Commit(consumeResult);

  369.                                 break;

  370.                             }

  371.                     }

  372. 

  373.                 }

  374.             }

  375.             catch (Exception ex)

  376.             {

  377.                 _logger.LogError($"解析定位消息发生异常 messageId: {msg!.MessageId}, Message: {ex.Message}, body: {JsonConvert.SerializeObject(msg)}");

  378.             }

  379. 

  380.             await Task.CompletedTask;

  381.         }

  382.         private async Task ProcessMessageAsync6(ConsumeResult<Ignore, string> consumeResult, IConsumer<Ignore, string> kafkaConsumer)

  383.         {

  384.             var msg = JsonConvert.DeserializeObject<MessageModel>(consumeResult.Message.Value);

  385.             string msgId = msg!.Data["MessageId"]?.ToObject<string>()!;

  386.             string bizId = msg!.Data["BusinessId"]?.ToObject<string>()!;

  387.             IotTopicType topic = msg!.Data["TopicType"]!.ToObject<IotTopicType>();

  388.             int type = msg!.Data["MsgType"]!.ToObject<int>();

  389.             object topicInfo = msg!.Data["TopicInfo"]!.ToObject<object>()!;

  390.             object detailData = msg!.Data["DetailData"]!.ToObject<object>()!;

  391.             var packge = new PackageMsgModel(msgId, bizId, topic, type, topicInfo, detailData);

  392. 

  393.             try

  394.             {

  395.                 using (_logger.BeginScope(new Dictionary<string, object> { ["RequestId"] = msg!.MessageId }))

  396.                 using (new CustomizeStopWatch(IotMessageTypeUtils.TryToDescription(type), _logger))

  397.                 {

  398.                     var dict = JsonConvert.DeserializeObject<Dictionary<string, object>>(msg!.Data["TopicInfo"]?["items"]?.ToString()!);

  399.                     var key = dict?.Keys?.FirstOrDefault() + "";

  400.                     switch (key)

  401.                     {

  402.                         case "GPS":     //处理gps位置信息上报

  403.                             {

  404.                                 //Console.WriteLine("GPS");

  405.                                 var data = JsonConvert.DeserializeObject<PropertyItemModel<GpsInfoModel>>(dict?["GPS"] + "");

  406.                                 //if (Utils.GetTimeDeviationMilliseconds(data!.Time) > _configAppsettings.DeviceTimeDeviationMillieconds) throw new ArgumentException($"设备消息时间戳异常[{data.Time}]");

  407.                                 _resolverGpsPosition.SetResolveInfo(packge);

  408.                                 await _resolverGpsPosition.ExecuteMessageAsync().ConfigureAwait(false);

  409.                                 break;

  410.                                 //return new PackageMsgModel(msg.MessageId, topicInfo.DeviceName, IotTopicType.Property, (int)IotMessagePropertyType.GpsPosition, topicInfo, data);

  411.                             }

  412. 

  413.                         case "WIFI":     //处理gps位置信息上报

  414.                             {

  415.                                 var data = JsonConvert.DeserializeObject<PropertyItemModel<WifiInfoModel>>(dict?["WIFI"] + "");

  416.                                 //if (Utils.GetTimeDeviationMilliseconds(data!.Time) > _configAppsettings.DeviceTimeDeviationMillieconds) throw new ArgumentException($"设备消息时间戳异常[{data.Time}]");

  417.                                 _resolverWifiPosition.SetResolveInfo(packge);

  418.                                 await _resolverWifiPosition.ExecuteMessageAsync().ConfigureAwait(false);

  419.                                 break;

  420.                             }

  421.                         case "WIFIPLUS2_0":     //处理gps位置信息上报

  422.                             {

  423.                                 var data = JsonConvert.DeserializeObject<PropertyItemModel<List<EachWifiPlus2Model>>>(dict?["WIFIPLUS2_0"] + "");

  424.                                 //if (Utils.GetTimeDeviationMilliseconds(data!.Time) > _configAppsettings.DeviceTimeDeviationMillieconds) throw new ArgumentException($"设备消息时间戳异常[{data.Time}]");

  425.                                 _resolverWifiPlus2Position.SetResolveInfo(packge);

  426.                                 await _resolverWifiPlus2Position.ExecuteMessageAsync().ConfigureAwait(false);

  427.                                 break;

  428.                             }

  429.                         case "LBS":     //处理gps位置信息上报

  430.                             {

  431.                                 var data = JsonConvert.DeserializeObject<PropertyItemModel<LbsInfoModel>>(dict?["LBS"] + "");

  432.                                 //if (Utils.GetTimeDeviationMilliseconds(data!.Time) > _configAppsettings.DeviceTimeDeviationMillieconds) throw new ArgumentException($"设备消息时间戳异常[{data.Time}]");

  433. 

  434.                                 //if (data.Value.Cdma == 0)

  435.                                 //{

  436.                                 //    return new PackageMsgModel(msg.MessageId, topicInfo.DeviceName, IotTopicType.Property, (int)IotMessagePropertyType.LbsGsmPosition, topicInfo, data);

  437.                                 //}

  438.                                 //else if (data.Value.Cdma == 1)

  439.                                 //{

  440.                                 //    return new PackageMsgModel(msg.MessageId, topicInfo.DeviceName, IotTopicType.Property, (int)IotMessagePropertyType.LbsCdmaPosition, topicInfo, data);

  441.                                 //}

  442.                                 _resolverLbsGsmPosition.SetResolveInfo(packge);

  443.                                 await _resolverLbsGsmPosition.ExecuteMessageAsync().ConfigureAwait(false);

  444.                                 break;

  445.                             }

  446.                     }

  447. 

  448.                 }

  449.             }

  450.             catch (Exception ex)

  451.             {

  452.                 _logger.LogError($"解析定位消息发生异常 messageId: {msg!.MessageId}, Message: {ex.Message}, body: {JsonConvert.SerializeObject(msg)}");

  453.             }

  454. 

  455.             await Task.CompletedTask;

  456.         }

  457. 

  458.         private async Task ProcessMessageAsync5(ConsumeResult<Ignore, string> consumeResult, IConsumer<Ignore, string> kafkaConsumer)

  459.         {

  460.             var msg = JsonConvert.DeserializeObject<MessageModel>(consumeResult.Message.Value);

  461.             try

  462.             {

  463.                 

  464.                 using (_logger.BeginScope(new Dictionary<string, object> { ["RequestId"] = msg!.MessageId }))

  465.                 //using (new CustomizeStopWatch(nameof(ProcessMessageAsync), _logger))

  466.                 {

  467.                     _factoryResolver.Resolver(msg);

  468.                 }

  469.             }

  470.             catch (Exception ex)

  471.             {

  472.                 _logger.LogError($"解析Property主题消息发生异常 messageId: {msg!.MessageId}, Message: {ex.Message}, body: {JsonConvert.SerializeObject(msg)}");

  473.             }

  474. 

  475.             await Task.CompletedTask;

  476.         }

  477.         private async Task ProcessMessageAsync4(ConsumeResult<Ignore, string> consumeResult, IConsumer<Ignore, string> kafkaConsumer)

  478.         {

  479.             var msg = JsonConvert.DeserializeObject<MessageModel>(consumeResult.Message.Value);

  480.             try

  481.             {

  482.                 

  483.                 using (_logger.BeginScope(new Dictionary<string, object> { ["RequestId"] = msg!.MessageId }))

  484.                 //using (new CustomizeStopWatch(nameof(ProcessMessageAsync), _logger))

  485.                 {

  486.                     //var positionType = msg!.Data;

  487.                     string msgId = msg!.Data["MessageId"]?.ToObject<string>()!;

  488.                     string bizId = msg!.Data["BusinessId"]?.ToObject<string>()!;

  489.                     IotTopicType topic = msg!.Data["TopicType"]!.ToObject<IotTopicType>();

  490.                     int type = msg!.Data["MsgType"]!.ToObject<int>();

  491.                     object topicInfo = msg!.Data["TopicInfo"]!.ToObject<object>()!;

  492.                     object detailData = msg!.Data["DetailData"]!.ToObject<object>()!;

  493.                     var  packge = new PackageMsgModel(msgId, bizId, topic, type, topicInfo, detailData);

  494. 

  495.                     //return new PackageMsgModel(msgId, bizId, topic, type, topicInfo, detailData);

  496.                     var dict = JsonConvert.DeserializeObject<Dictionary<string, object>>(msg!.Data["TopicInfo"]?["items"]?.ToString()!);

  497.                     var key = dict?.Keys?.FirstOrDefault() + "";

  498.                     switch (key)

  499.                     {

  500.                         case "GPS":     //处理gps位置信息上报

  501.                             {

  502.                                 Console.WriteLine("GPS");

  503.                                 var data = JsonConvert.DeserializeObject<PropertyItemModel<GpsInfoModel>>(dict?["GPS"] + "");

  504.                                 if (Utils.GetTimeDeviationMilliseconds(data!.Time) > _configAppsettings.DeviceTimeDeviationMillieconds) throw new ArgumentException($"设备消息时间戳异常[{data.Time}]");

  505.                                 break;

  506.                                 //return new PackageMsgModel(msg.MessageId, topicInfo.DeviceName, IotTopicType.Property, (int)IotMessagePropertyType.GpsPosition, topicInfo, data);

  507.                             }

  508. 

  509.                         case "WIFI":     //处理gps位置信息上报

  510.                             {

  511.                                 var data = JsonConvert.DeserializeObject<PropertyItemModel<WifiInfoModel>>(dict?["WIFI"] + "");

  512.                                 if (Utils.GetTimeDeviationMilliseconds(data!.Time) > _configAppsettings.DeviceTimeDeviationMillieconds) throw new ArgumentException($"设备消息时间戳异常[{data.Time}]");

  513. 

  514.                                 Console.WriteLine("WIFI");

  515. 

  516.                                 break;

  517.                             }

  518.                         case "WIFIPLUS2_0":     //处理gps位置信息上报

  519.                             {

  520.                                 var data = JsonConvert.DeserializeObject<PropertyItemModel<List<EachWifiPlus2Model>>>(dict?["WIFIPLUS2_0"] + "");

  521.                                 if (Utils.GetTimeDeviationMilliseconds(data!.Time) > _configAppsettings.DeviceTimeDeviationMillieconds) throw new ArgumentException($"设备消息时间戳异常[{data.Time}]");

  522. 

  523.                                 Console.WriteLine("WIFIPLUS2_0");

  524. 

  525.                                 break;

  526.                             }

  527.                         case "LBS":     //处理gps位置信息上报

  528.                             {

  529.                                 var data = JsonConvert.DeserializeObject<PropertyItemModel<LbsInfoModel>>(dict?["LBS"] + "");

  530.                                 if (Utils.GetTimeDeviationMilliseconds(data!.Time) > _configAppsettings.DeviceTimeDeviationMillieconds) throw new ArgumentException($"设备消息时间戳异常[{data.Time}]");

  531. 

  532.                                 //if (data.Value.Cdma == 0)

  533.                                 //{

  534.                                 //    return new PackageMsgModel(msg.MessageId, topicInfo.DeviceName, IotTopicType.Property, (int)IotMessagePropertyType.LbsGsmPosition, topicInfo, data);

  535.                                 //}

  536.                                 //else if (data.Value.Cdma == 1)

  537.                                 //{

  538.                                 //    return new PackageMsgModel(msg.MessageId, topicInfo.DeviceName, IotTopicType.Property, (int)IotMessagePropertyType.LbsCdmaPosition, topicInfo, data);

  539.                                 //}

  540.                                 Console.WriteLine("WIFIPLUS2_0");

  541. 

  542.                                 break;

  543.                             }

  544.                     }

  545.                     

  546.                 }

  547.             }

  548.             catch (Exception ex)

  549.             {

  550.                 _logger.LogError($"解析Property主题消息发生异常 messageId: {msg!.MessageId}, Message: {ex.Message}, body: {JsonConvert.SerializeObject(msg)}");

  551.             }

  552. 

  553.             await Task.CompletedTask;

  554.         }

  555. 

  556.         private async Task ProcessMessageAsync3(ConsumeResult<Ignore, string> consumeResult, IConsumer<Ignore, string> kafkaConsumer)

  557.         {

  558.             #region

  559. 

  560.             Object data = JObject.Parse(@"{

  561.      'timestamp': 1670231477000,

  562.      'id': '679e9043-14fe-44df-aba6-b2fd62592fea',

  563.      'asset': {

  564.          'id': '679e9043-14fe-44df-aba6-b2fd62592fea',

  565.          'manufacturer': 'xiaomi',

  566.          'imeiMd5': '1be14c6210b3115f',

  567.          'uniqueId': '8f95204d-a60e-40bb-b8be-45a1ab3bef2a'

  568.      },

  569.      'location': {

  570.          'Wifis': [

  571.              {

  572.                  'timestamp': 1515743846504,

  573.                  'signalStrength': 40,

  574.                  'macAddress': '6a:77:24:27:9c:04',

  575.                  'ssid': '',

  576.                  'frequency': 2412,

  577.                  'channel': 0,

  578.                  'connected': true

  579.              },

  580.              {

  581.                  'timestamp': 1515743846504,

  582.                  'signalStrength': 60,

  583.                  'macAddress': 'ec:f0:fe:1e:e1:a8',

  584.                  'ssid': 'Z VENTURES 9F',

  585.                  'frequency': 2457,

  586.                  'channel': 0

  587.              },

  588.              {

  589.                  'timestamp': 1515743846504,

  590.                  'signalStrength': 60,

  591.                  'macAddress': '54:a7:03:b5:a5:5e',

  592.                  'ssid': 'Z VENTURES 9F',

  593.                  'frequency': 2457,

  594.                  'channel': 0

  595.              },

  596.              {

  597.                  'timestamp': 1515743846504,

  598.                  'signalStrength': 60,

  599.                  'macAddress': 'ec:f8:eb:b6:89:81',

  600.                  'ssid': 'Z VENTURES 9F',

  601.                  'frequency': 2457,

  602.                  'channel': 0

  603.              },

  604.              {

  605.                  'timestamp': 1515743846504,

  606.                  'signalStrength': 60,

  607.                  'macAddress': 'a0:69:d9:c1:6c:8b',

  608.                  'ssid': 'Z VENTURES 9F',

  609.                  'frequency': 2457,

  610.                  'channel': 0

  611.              },

  612.              {

  613.                  'timestamp': 1515743846504,

  614.                  'signalStrength': 60,

  615.                  'macAddress': '64:64:4a:d5:f5:ca',

  616.                  'ssid': 'Z VENTURES 9F',

  617.                  'frequency': 2457,

  618.                  'channel': 0

  619.              },

  620.              {

  621.                  'timestamp': 1515743846504,

  622.                  'signalStrength': 60,

  623.                  'macAddress': 'c8:5b:a0:f8:f4:6c',

  624.                  'ssid': 'Z VENTURES 9F',

  625.                  'frequency': 2457,

  626.                  'channel': 0

  627.              },

  628.              {

  629.                  'timestamp': 1515743846504,

  630.                  'signalStrength': 60,

  631.                  'macAddress': '84:74:60:bf:fd:20',

  632.                  'ssid': 'Z VENTURES 9F',

  633.                  'frequency': 2457,

  634.                  'channel': 0

  635.              }

  636.          ]

  637.      }

  638. }");

  639.             // 将对象转换为 JSON 字符串

  640.             var jsonData = JsonConvert.SerializeObject(data);

  641. 

  642.             // 创建 StringContent 对象，将 JSON 字符串传递给它

  643.             var content = new StringContent(jsonData, Encoding.UTF8, "application/json");

  644.             #endregion

  645. 

  646.             var msg = JsonConvert.DeserializeObject<MessageModel>(consumeResult.Message.Value);

  647.             using (_logger.BeginScope(new Dictionary<string, object> { ["RequestId"] = msg!.MessageId }))

  648.             //using (new CustomizeStopWatch(nameof(ProcessMessageAsync), _logger))

  649.             {

  650.                 using var httpClient = new HttpClient();

  651.                 using var request = new HttpRequestMessage

  652.                 {

  653.                     RequestUri = new Uri($"https://api.newayz.com/location/hub/v1/track_points?access_key=OK73jLa48dU9V3EvNN1RLpdjcVOdIEcn&field_masks=location.position"),

  654.                     Method = HttpMethod.Post,

  655.                     // Version = new Version(2, 0)

  656.                     Content = content

  657.                 };

  658.                 using var tokenResponse = await httpClient.SendAsync(request);

  659.                 tokenResponse.EnsureSuccessStatusCode();

  660. 

  661.                 var token = await tokenResponse.Content.ReadAsStringAsync();

  662.                 //await Task.Delay(0);

  663.                 //Thread.Sleep(0);

  664.                 //Console.WriteLine($"{Guid.NewGuid().ToString()}-- {token[^10..]}");

  665.             }

  666. 

  667.         }

  668. 

  669.         private async Task ProcessMessageAsync2(ConsumeResult<Ignore, string> consumeResult, IConsumer<Ignore, string> kafkaConsumer)

  670.         {

  671.             // using (_logger.BeginScope(new Dictionary<string, object> { ["RequestId"] = i }))

  672.             /**

  673.              using (new CustomizeStopWatch(nameof(ProcessMessageAsync), _logger))

  674.              {

  675.                  //if (i==800)

  676.                  //{

  677.                  //    Console.WriteLine($"第{i}消息:{consumeResult.Message.Value.Substring(0, 10)}");

  678. 

  679.                  //}

  680.                  //await Task.CompletedTask;

  681.                  Object data = JObject.Parse(@"{

  682.      'timestamp': 1670231477000,

  683.      'id': '679e9043-14fe-44df-aba6-b2fd62592fea',

  684.      'asset': {

  685.          'id': '679e9043-14fe-44df-aba6-b2fd62592fea',

  686.          'manufacturer': 'xiaomi',

  687.          'imeiMd5': '1be14c6210b3115f',

  688.          'uniqueId': '8f95204d-a60e-40bb-b8be-45a1ab3bef2a'

  689.      },

  690.      'location': {

  691.          'Wifis': [

  692.              {

  693.                  'timestamp': 1515743846504,

  694.                  'signalStrength': 40,

  695.                  'macAddress': '6a:77:24:27:9c:04',

  696.                  'ssid': '',

  697.                  'frequency': 2412,

  698.                  'channel': 0,

  699.                  'connected': true

  700.              },

  701.              {

  702.                  'timestamp': 1515743846504,

  703.                  'signalStrength': 60,

  704.                  'macAddress': 'ec:f0:fe:1e:e1:a8',

  705.                  'ssid': 'Z VENTURES 9F',

  706.                  'frequency': 2457,

  707.                  'channel': 0

  708.              },

  709.              {

  710.                  'timestamp': 1515743846504,

  711.                  'signalStrength': 60,

  712.                  'macAddress': '54:a7:03:b5:a5:5e',

  713.                  'ssid': 'Z VENTURES 9F',

  714.                  'frequency': 2457,

  715.                  'channel': 0

  716.              },

  717.              {

  718.                  'timestamp': 1515743846504,

  719.                  'signalStrength': 60,

  720.                  'macAddress': 'ec:f8:eb:b6:89:81',

  721.                  'ssid': 'Z VENTURES 9F',

  722.                  'frequency': 2457,

  723.                  'channel': 0

  724.              },

  725.              {

  726.                  'timestamp': 1515743846504,

  727.                  'signalStrength': 60,

  728.                  'macAddress': 'a0:69:d9:c1:6c:8b',

  729.                  'ssid': 'Z VENTURES 9F',

  730.                  'frequency': 2457,

  731.                  'channel': 0

  732.              },

  733.              {

  734.                  'timestamp': 1515743846504,

  735.                  'signalStrength': 60,

  736.                  'macAddress': '64:64:4a:d5:f5:ca',

  737.                  'ssid': 'Z VENTURES 9F',

  738.                  'frequency': 2457,

  739.                  'channel': 0

  740.              },

  741.              {

  742.                  'timestamp': 1515743846504,

  743.                  'signalStrength': 60,

  744.                  'macAddress': 'c8:5b:a0:f8:f4:6c',

  745.                  'ssid': 'Z VENTURES 9F',

  746.                  'frequency': 2457,

  747.                  'channel': 0

  748.              },

  749.              {

  750.                  'timestamp': 1515743846504,

  751.                  'signalStrength': 60,

  752.                  'macAddress': '84:74:60:bf:fd:20',

  753.                  'ssid': 'Z VENTURES 9F',

  754.                  'frequency': 2457,

  755.                  'channel': 0

  756.              }

  757.          ]

  758.      }

  759. }");

  760.                  // 将对象转换为 JSON 字符串

  761.                  var jsonData = JsonConvert.SerializeObject(data);

  762. 

  763.                  // 创建 StringContent 对象，将 JSON 字符串传递给它

  764.                  var content = new StringContent(jsonData, Encoding.UTF8, "application/json");

  765.                  using var httpClient = new HttpClient();

  766.                  using var request = new HttpRequestMessage

  767.                  {

  768.                      RequestUri = new Uri($"https://api.newayz.com/location/hub/v1/track_points?access_key=OK73jLa48dU9V3EvNN1RLpdjcVOdIEcn&field_masks=location.position"),

  769.                      Method = HttpMethod.Post,

  770.                      // Version = new Version(2, 0)

  771.                      Content = content

  772.                  };

  773.                  using var tokenResponse = await httpClient.SendAsync(request);

  774.                  tokenResponse.EnsureSuccessStatusCode();

  775. 

  776.                  var token = await tokenResponse.Content.ReadAsStringAsync();

  777.                  Console.WriteLine($"Successfully authenticated. token {token}");

  778.              }

  779.              */

  780. 

  781.             Object data = JObject.Parse(@"{

  782. 	'timestamp': 1670231477000,

  783. 	'id': '679e9043-14fe-44df-aba6-b2fd62592fea',

  784. 	'asset': {

  785. 		'id': '679e9043-14fe-44df-aba6-b2fd62592fea',

  786. 		'manufacturer': 'xiaomi',

  787. 		'imeiMd5': '1be14c6210b3115f',

  788. 		'uniqueId': '8f95204d-a60e-40bb-b8be-45a1ab3bef2a'

  789. 	},

  790. 	'location': {

  791. 		'Wifis': [

  792. 			{

  793. 				'timestamp': 1515743846504,

  794. 				'signalStrength': 40,

  795. 				'macAddress': '6a:77:24:27:9c:04',

  796. 				'ssid': '',

  797. 				'frequency': 2412,

  798. 				'channel': 0,

  799. 				'connected': true

  800. 			},

  801. 			{

  802. 				'timestamp': 1515743846504,

  803. 				'signalStrength': 60,

  804. 				'macAddress': 'ec:f0:fe:1e:e1:a8',

  805. 				'ssid': 'Z VENTURES 9F',

  806. 				'frequency': 2457,

  807. 				'channel': 0

  808. 			},

  809. 			{

  810. 				'timestamp': 1515743846504,

  811. 				'signalStrength': 60,

  812. 				'macAddress': '54:a7:03:b5:a5:5e',

  813. 				'ssid': 'Z VENTURES 9F',

  814. 				'frequency': 2457,

  815. 				'channel': 0

  816. 			},

  817. 			{

  818. 				'timestamp': 1515743846504,

  819. 				'signalStrength': 60,

  820. 				'macAddress': 'ec:f8:eb:b6:89:81',

  821. 				'ssid': 'Z VENTURES 9F',

  822. 				'frequency': 2457,

  823. 				'channel': 0

  824. 			},

  825. 			{

  826. 				'timestamp': 1515743846504,

  827. 				'signalStrength': 60,

  828. 				'macAddress': 'a0:69:d9:c1:6c:8b',

  829. 				'ssid': 'Z VENTURES 9F',

  830. 				'frequency': 2457,

  831. 				'channel': 0

  832. 			},

  833. 			{

  834. 				'timestamp': 1515743846504,

  835. 				'signalStrength': 60,

  836. 				'macAddress': '64:64:4a:d5:f5:ca',

  837. 				'ssid': 'Z VENTURES 9F',

  838. 				'frequency': 2457,

  839. 				'channel': 0

  840. 			},

  841. 			{

  842. 				'timestamp': 1515743846504,

  843. 				'signalStrength': 60,

  844. 				'macAddress': 'c8:5b:a0:f8:f4:6c',

  845. 				'ssid': 'Z VENTURES 9F',

  846. 				'frequency': 2457,

  847. 				'channel': 0

  848. 			},

  849. 			{

  850. 				'timestamp': 1515743846504,

  851. 				'signalStrength': 60,

  852. 				'macAddress': '84:74:60:bf:fd:20',

  853. 				'ssid': 'Z VENTURES 9F',

  854. 				'frequency': 2457,

  855. 				'channel': 0

  856. 			}

  857. 		]

  858. 	}

  859. }");

  860.             // 将对象转换为 JSON 字符串

  861.             var jsonData = JsonConvert.SerializeObject(data);

  862. 

  863.             // 创建 StringContent 对象，将 JSON 字符串传递给它

  864.             var content = new StringContent(jsonData, Encoding.UTF8, "application/json");

  865.             using var httpClient = new HttpClient();

  866.             using var request = new HttpRequestMessage

  867.             {

  868.                 RequestUri = new Uri($"https://api.newayz.com/location/hub/v1/track_points?access_key=OK73jLa48dU9V3EvNN1RLpdjcVOdIEcn&field_masks=location.position"),

  869.                 Method = HttpMethod.Post,

  870.                 // Version = new Version(2, 0)

  871.                 Content = content

  872.             };

  873.             using var tokenResponse = await httpClient.SendAsync(request);

  874.             tokenResponse.EnsureSuccessStatusCode();

  875. 

  876.             var token = await tokenResponse.Content.ReadAsStringAsync();

  877.             Console.WriteLine($"{Guid.NewGuid().ToString()}-- {token[^10..]}");

  878. 

  879.         }

  880.         

  881.         private async Task ProcessMessageAsync(ConsumeResult<Ignore, string> consumeResult, IConsumer<Ignore, string> kafkaConsumer,int i)

  882.         {

  883.             using (_logger.BeginScope(new Dictionary<string, object> { ["RequestId"] = i }))

  884.             using (new CustomizeStopWatch(nameof(ProcessMessageAsync), _logger))

  885.             {

  886.                 //if (i==800)

  887.                 //{

  888.                 //    Console.WriteLine($"第{i}消息:{consumeResult.Message.Value.Substring(0, 10)}");

  889. 

  890.                 //}

  891.                 //await Task.CompletedTask;

  892.                 Object data = JObject.Parse(@"{

  893. 	'timestamp': 1670231477000,

  894. 	'id': '679e9043-14fe-44df-aba6-b2fd62592fea',

  895. 	'asset': {

  896. 		'id': '679e9043-14fe-44df-aba6-b2fd62592fea',

  897. 		'manufacturer': 'xiaomi',

  898. 		'imeiMd5': '1be14c6210b3115f',

  899. 		'uniqueId': '8f95204d-a60e-40bb-b8be-45a1ab3bef2a'

  900. 	},

  901. 	'location': {

  902. 		'Wifis': [

  903. 			{

  904. 				'timestamp': 1515743846504,

  905. 				'signalStrength': 40,

  906. 				'macAddress': '6a:77:24:27:9c:04',

  907. 				'ssid': '',

  908. 				'frequency': 2412,

  909. 				'channel': 0,

  910. 				'connected': true

  911. 			},

  912. 			{

  913. 				'timestamp': 1515743846504,

  914. 				'signalStrength': 60,

  915. 				'macAddress': 'ec:f0:fe:1e:e1:a8',

  916. 				'ssid': 'Z VENTURES 9F',

  917. 				'frequency': 2457,

  918. 				'channel': 0

  919. 			},

  920. 			{

  921. 				'timestamp': 1515743846504,

  922. 				'signalStrength': 60,

  923. 				'macAddress': '54:a7:03:b5:a5:5e',

  924. 				'ssid': 'Z VENTURES 9F',

  925. 				'frequency': 2457,

  926. 				'channel': 0

  927. 			},

  928. 			{

  929. 				'timestamp': 1515743846504,

  930. 				'signalStrength': 60,

  931. 				'macAddress': 'ec:f8:eb:b6:89:81',

  932. 				'ssid': 'Z VENTURES 9F',

  933. 				'frequency': 2457,

  934. 				'channel': 0

  935. 			},

  936. 			{

  937. 				'timestamp': 1515743846504,

  938. 				'signalStrength': 60,

  939. 				'macAddress': 'a0:69:d9:c1:6c:8b',

  940. 				'ssid': 'Z VENTURES 9F',

  941. 				'frequency': 2457,

  942. 				'channel': 0

  943. 			},

  944. 			{

  945. 				'timestamp': 1515743846504,

  946. 				'signalStrength': 60,

  947. 				'macAddress': '64:64:4a:d5:f5:ca',

  948. 				'ssid': 'Z VENTURES 9F',

  949. 				'frequency': 2457,

  950. 				'channel': 0

  951. 			},

  952. 			{

  953. 				'timestamp': 1515743846504,

  954. 				'signalStrength': 60,

  955. 				'macAddress': 'c8:5b:a0:f8:f4:6c',

  956. 				'ssid': 'Z VENTURES 9F',

  957. 				'frequency': 2457,

  958. 				'channel': 0

  959. 			},

  960. 			{

  961. 				'timestamp': 1515743846504,

  962. 				'signalStrength': 60,

  963. 				'macAddress': '84:74:60:bf:fd:20',

  964. 				'ssid': 'Z VENTURES 9F',

  965. 				'frequency': 2457,

  966. 				'channel': 0

  967. 			}

  968. 		]

  969. 	}

  970. }");

  971.                 // 将对象转换为 JSON 字符串

  972.                 var jsonData = JsonConvert.SerializeObject(data);

  973. 

  974.                 // 创建 StringContent 对象，将 JSON 字符串传递给它

  975.                 var content = new StringContent(jsonData, Encoding.UTF8, "application/json");

  976.                 using var httpClient = new HttpClient();

  977.                 using var request = new HttpRequestMessage

  978.                 {

  979.                     RequestUri = new Uri($"https://api.newayz.com/location/hub/v1/track_points?access_key=OK73jLa48dU9V3EvNN1RLpdjcVOdIEcn&field_masks=location.position"),

  980.                     Method = HttpMethod.Post,

  981.                    // Version = new Version(2, 0)

  982.                     Content= content

  983.                 };

  984.                 using var tokenResponse = await httpClient.SendAsync(request);

  985.                 tokenResponse.EnsureSuccessStatusCode();

  986. 

  987.                 var token = await tokenResponse.Content.ReadAsStringAsync();

  988.                 Console.WriteLine($"Successfully authenticated. token {token}");

  989.             }

  990. 

  991.         }

  992.         private IConsumer<Ignore, string> CreateKafkaConsumer()

  993.         {

  994.             //var partitionIndex = 0;

  995.             var collection = new List<int>();

  996.             // 获取注册的kafka消费者

  997. 

  998.             var consumerConfig = new ConsumerConfig

  999.             {

  1000.                 GroupId = "iot.position",

  1001.                 BootstrapServers = _configService.MqServerAddress,

  1002.                 AutoOffsetReset = AutoOffsetReset.Earliest,

  1003.                 EnableAutoCommit = false, // 关闭自动提交偏移量

  1004.                 CancellationDelayMaxMs = 1//set CancellationDelayMaxMs

  1005.             };

  1006. 

  1007.             return new ConsumerBuilder<Ignore, string>(consumerConfig)

  1008.                      .SetErrorHandler((_, e) =>

  1009.                      {

  1010.                          Console.WriteLine($"消费者创建出错,代码:{e.Code} |原因: {e.Reason}");

  1011.                          _logger.LogInformation($"消费者创建出错,代码:{e.Code} |原因: {e.Reason}");

  1012.                      })

  1013.                      .SetPartitionsAssignedHandler((c, partitions) =>

  1014.                      {

  1015.                          //// 在这里手动指定要消费的分区

  1016.                          //var partitionsToConsume = new List<TopicPartitionOffset>

  1017.                          //{

  1018.                          //   new TopicPartitionOffset("topics.storage.test_env_db", partitionIndex, Offset.Unset)

  1019.                          //};

  1020. 

  1021.                          ////c.Assign(partitionsToConsume);

  1022.                          //Console.WriteLine($"Assigned partitions: {string.Join(", ", partitionsToConsume)}");

  1023.                          //return partitionsToConsume;

  1024.                      })

  1025.                      .SetPartitionsRevokedHandler((c, partitions) =>

  1026.                      {

  1027. 

  1028.                      })

  1029.                      .Build();

  1030.         }

  1031. 

  1032.     }

  1033. }