提交kafka

GpsCardGatewayPosition.Postion/Worker.cs

@@ -83,9 +83,9 @@ namespace GpsCardGatewayPosition.Postion
             csredisWayzCon.DefaultDatabase = 10;
             csredisDb7Con.Prefix = "_GW_";
 #if DEBUG
-            csredisWayzCon.Server = "192.168.2.121:8090";
+            csredisWayzCon.Server = "192.168.12.127:8090";
 #else
-            csredisWayzCon.Server = "172.19.42.56:8090";
+            csredisWayzCon.Server = "192.168.12.127:8090";
 #endif
             var csredisWayz = new CSRedis.CSRedisClient(csredisWayzCon.ToString());
             RedisHelperWayz.Initialization(csredisWayz);
@@ -327,6 +327,7 @@ namespace GpsCardGatewayPosition.Postion
                                 //if (Utils.GetTimeDeviationMilliseconds(data!.Time) > _configAppsettings.DeviceTimeDeviationMillieconds) throw new ArgumentException($"设备消息时间戳异常[{data.Time}]");
                                 _resolverGpsPosition.SetResolveInfo(packge);
                                 await _resolverGpsPosition.ExecuteMessageAsync().ConfigureAwait(false);
+                                kafkaConsumer.Commit(consumeResult);
                                 break;
                                 //return new PackageMsgModel(msg.MessageId, topicInfo.DeviceName, IotTopicType.Property, (int)IotMessagePropertyType.GpsPosition, topicInfo, data);
                             }
@@ -337,6 +338,7 @@ namespace GpsCardGatewayPosition.Postion
                                 //if (Utils.GetTimeDeviationMilliseconds(data!.Time) > _configAppsettings.DeviceTimeDeviationMillieconds) throw new ArgumentException($"设备消息时间戳异常[{data.Time}]");
                                 _resolverWifiPosition.SetResolveInfo(packge);
                                 await _resolverWifiPosition.ExecuteMessageAsync().ConfigureAwait(false);
+                                kafkaConsumer.Commit(consumeResult);
                                 break;
                             }
                         case "WIFIPLUS2_0":     //处理gps位置信息上报
@@ -345,6 +347,7 @@ namespace GpsCardGatewayPosition.Postion
                                 //if (Utils.GetTimeDeviationMilliseconds(data!.Time) > _configAppsettings.DeviceTimeDeviationMillieconds) throw new ArgumentException($"设备消息时间戳异常[{data.Time}]");
                                 _resolverWifiPlus2Position.SetResolveInfo(packge);
                                 await _resolverWifiPlus2Position.ExecuteMessageAsync().ConfigureAwait(false);
+                                kafkaConsumer.Commit(consumeResult);
                                 break;
                             }
                         case "LBS":     //处理gps位置信息上报
@@ -362,6 +365,7 @@ namespace GpsCardGatewayPosition.Postion
                                 //}
                                 _resolverLbsGsmPosition.SetResolveInfo(packge);
                                 await _resolverLbsGsmPosition.ExecuteMessageAsync().ConfigureAwait(false);
+                                kafkaConsumer.Commit(consumeResult);
                                 break;
                             }
                     }