第一次提交

.gitignore

@@ -0,0 +1,340 @@
+## Ignore Visual Studio temporary files, build results, and
+## files generated by popular Visual Studio add-ons.
+##
+## Get latest from https://github.com/github/gitignore/blob/master/VisualStudio.gitignore
+
+# User-specific files
+*.rsuser
+*.suo
+*.user
+*.userosscache
+*.sln.docstates
+
+# User-specific files (MonoDevelop/Xamarin Studio)
+*.userprefs
+
+# Build results
+[Dd]ebug/
+[Dd]ebugPublic/
+[Rr]elease/
+[Rr]eleases/
+x64/
+x86/
+[Aa][Rr][Mm]/
+[Aa][Rr][Mm]64/
+bld/
+[Bb]in/
+[Oo]bj/
+[Ll]og/
+
+# Visual Studio 2015/2017 cache/options directory
+.vs/
+# Uncomment if you have tasks that create the project's static files in wwwroot
+#wwwroot/
+
+# Visual Studio 2017 auto generated files
+Generated\ Files/
+
+# MSTest test Results
+[Tt]est[Rr]esult*/
+[Bb]uild[Ll]og.*
+
+# NUNIT
+*.VisualState.xml
+TestResult.xml
+
+# Build Results of an ATL Project
+[Dd]ebugPS/
+[Rr]eleasePS/
+dlldata.c
+
+# Benchmark Results
+BenchmarkDotNet.Artifacts/
+
+# .NET Core
+project.lock.json
+project.fragment.lock.json
+artifacts/
+
+# StyleCop
+StyleCopReport.xml
+
+# Files built by Visual Studio
+*_i.c
+*_p.c
+*_h.h
+*.ilk
+*.meta
+*.obj
+*.iobj
+*.pch
+*.pdb
+*.ipdb
+*.pgc
+*.pgd
+*.rsp
+*.sbr
+*.tlb
+*.tli
+*.tlh
+*.tmp
+*.tmp_proj
+*_wpftmp.csproj
+*.log
+*.vspscc
+*.vssscc
+.builds
+*.pidb
+*.svclog
+*.scc
+
+# Chutzpah Test files
+_Chutzpah*
+
+# Visual C++ cache files
+ipch/
+*.aps
+*.ncb
+*.opendb
+*.opensdf
+*.sdf
+*.cachefile
+*.VC.db
+*.VC.VC.opendb
+
+# Visual Studio profiler
+*.psess
+*.vsp
+*.vspx
+*.sap
+
+# Visual Studio Trace Files
+*.e2e
+
+# TFS 2012 Local Workspace
+$tf/
+
+# Guidance Automation Toolkit
+*.gpState
+
+# ReSharper is a .NET coding add-in
+_ReSharper*/
+*.[Rr]e[Ss]harper
+*.DotSettings.user
+
+# JustCode is a .NET coding add-in
+.JustCode
+
+# TeamCity is a build add-in
+_TeamCity*
+
+# DotCover is a Code Coverage Tool
+*.dotCover
+
+# AxoCover is a Code Coverage Tool
+.axoCover/*
+!.axoCover/settings.json
+
+# Visual Studio code coverage results
+*.coverage
+*.coveragexml
+
+# NCrunch
+_NCrunch_*
+.*crunch*.local.xml
+nCrunchTemp_*
+
+# MightyMoose
+*.mm.*
+AutoTest.Net/
+
+# Web workbench (sass)
+.sass-cache/
+
+# Installshield output folder
+[Ee]xpress/
+
+# DocProject is a documentation generator add-in
+DocProject/buildhelp/
+DocProject/Help/*.HxT
+DocProject/Help/*.HxC
+DocProject/Help/*.hhc
+DocProject/Help/*.hhk
+DocProject/Help/*.hhp
+DocProject/Help/Html2
+DocProject/Help/html
+
+# Click-Once directory
+publish/
+
+# Publish Web Output
+*.[Pp]ublish.xml
+*.azurePubxml
+# Note: Comment the next line if you want to checkin your web deploy settings,
+# but database connection strings (with potential passwords) will be unencrypted
+*.pubxml
+*.publishproj
+
+# Microsoft Azure Web App publish settings. Comment the next line if you want to
+# checkin your Azure Web App publish settings, but sensitive information contained
+# in these scripts will be unencrypted
+PublishScripts/
+
+# NuGet Packages
+*.nupkg
+# The packages folder can be ignored because of Package Restore
+**/[Pp]ackages/*
+# except build/, which is used as an MSBuild target.
+!**/[Pp]ackages/build/
+# Uncomment if necessary however generally it will be regenerated when needed
+#!**/[Pp]ackages/repositories.config
+# NuGet v3's project.json files produces more ignorable files
+*.nuget.props
+*.nuget.targets
+
+# Microsoft Azure Build Output
+csx/
+*.build.csdef
+
+# Microsoft Azure Emulator
+ecf/
+rcf/
+
+# Windows Store app package directories and files
+AppPackages/
+BundleArtifacts/
+Package.StoreAssociation.xml
+_pkginfo.txt
+*.appx
+
+# Visual Studio cache files
+# files ending in .cache can be ignored
+*.[Cc]ache
+# but keep track of directories ending in .cache
+!?*.[Cc]ache/
+
+# Others
+ClientBin/
+~$*
+*~
+*.dbmdl
+*.dbproj.schemaview
+*.jfm
+*.pfx
+*.publishsettings
+orleans.codegen.cs
+
+# Including strong name files can present a security risk
+# (https://github.com/github/gitignore/pull/2483#issue-259490424)
+#*.snk
+
+# Since there are multiple workflows, uncomment next line to ignore bower_components
+# (https://github.com/github/gitignore/pull/1529#issuecomment-104372622)
+#bower_components/
+
+# RIA/Silverlight projects
+Generated_Code/
+
+# Backup & report files from converting an old project file
+# to a newer Visual Studio version. Backup files are not needed,
+# because we have git ;-)
+_UpgradeReport_Files/
+Backup*/
+UpgradeLog*.XML
+UpgradeLog*.htm
+ServiceFabricBackup/
+*.rptproj.bak
+
+# SQL Server files
+*.mdf
+*.ldf
+*.ndf
+
+# Business Intelligence projects
+*.rdl.data
+*.bim.layout
+*.bim_*.settings
+*.rptproj.rsuser
+*- Backup*.rdl
+
+# Microsoft Fakes
+FakesAssemblies/
+
+# GhostDoc plugin setting file
+*.GhostDoc.xml
+
+# Node.js Tools for Visual Studio
+.ntvs_analysis.dat
+node_modules/
+
+# Visual Studio 6 build log
+*.plg
+
+# Visual Studio 6 workspace options file
+*.opt
+
+# Visual Studio 6 auto-generated workspace file (contains which files were open etc.)
+*.vbw
+
+# Visual Studio LightSwitch build output
+**/*.HTMLClient/GeneratedArtifacts
+**/*.DesktopClient/GeneratedArtifacts
+**/*.DesktopClient/ModelManifest.xml
+**/*.Server/GeneratedArtifacts
+**/*.Server/ModelManifest.xml
+_Pvt_Extensions
+
+# Paket dependency manager
+.paket/paket.exe
+paket-files/
+
+# FAKE - F# Make
+.fake/
+
+# JetBrains Rider
+.idea/
+*.sln.iml
+
+# CodeRush personal settings
+.cr/personal
+
+# Python Tools for Visual Studio (PTVS)
+__pycache__/
+*.pyc
+
+# Cake - Uncomment if you are using it
+# tools/**
+# !tools/packages.config
+
+# Tabs Studio
+*.tss
+
+# Telerik's JustMock configuration file
+*.jmconfig
+
+# BizTalk build output
+*.btp.cs
+*.btm.cs
+*.odx.cs
+*.xsd.cs
+
+# OpenCover UI analysis results
+OpenCover/
+
+# Azure Stream Analytics local run output
+ASALocalRun/
+
+# MSBuild Binary and Structured Log
+*.binlog
+
+# NVidia Nsight GPU debugger configuration file
+*.nvuser
+
+# MFractors (Xamarin productivity tool) working folder
+.mfractor/
+
+# Local History for Visual Studio
+.localhistory/
+
+# BeatPulse healthcheck temp database
+healthchecksdb

TelpoPush.WanJiaAn.Worker/Common/HttpHelperAsync.cs

@@ -0,0 +1,460 @@
+using Microsoft.Extensions.Logging;
+using Newtonsoft.Json;
+using System;
+using System.Collections.Generic;
+using System.Net.Http;
+using System.Net.Http.Headers;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace TelpoPush.WanJiaAn.Worker.Common
+{ 
+    /// <summary>
+    /// HTTP帮助类
+    /// </summary>
+    public class HttpHelperAsync
+    {
+        private IHttpClientFactory _httpClientFactory;
+        private readonly ILogger<HttpHelperAsync> _logger;
+        public HttpHelperAsync(IHttpClientFactory httpClientFactory, ILogger<HttpHelperAsync> logger)
+        {
+            _httpClientFactory = httpClientFactory;
+            _logger = logger;
+        }
+
+
+        #region  异步
+        /// <summary>
+        /// 发起POST异步请求表单
+        /// </summary>
+        /// <param name="url">请求地址</param>
+        /// <param name="body">POST提交的内容</param>
+        /// <param name="bodyMediaType">POST内容的媒体类型，如：application/xml、application/json</param>
+        /// <param name="responseContentType">HTTP响应上的content-type内容头的值,如：application/xml、application/json、application/text、application/x-www-form-urlencoded等</param>
+        /// <param name="headers">请求头信息</param>
+        /// <param name="timeOut">请求超时时间，单位秒</param>
+        /// <returns>返回string</returns>
+        public async Task<string> PostFormAsync(string url, MultipartFormDataContent content,
+            Dictionary<string, string> headers = null,
+            int timeOut = 5)
+        {
+            try
+            {
+                var hostName = GetHostName(url);
+                using (HttpClient client = _httpClientFactory.CreateClient(hostName))
+                {
+                    client.Timeout = TimeSpan.FromSeconds(timeOut);
+                    if (headers?.Count > 0)
+                    {
+                        foreach (string key in headers.Keys)
+                        {
+                            client.DefaultRequestHeaders.Add(key, headers[key]);
+                        }
+                    }
+                    content.Headers.Add("ContentType", "multipart/form-data");//声明头部
+                    using (HttpResponseMessage response = await client.PostAsync(url, content))
+                    {
+                        return JsonConvert.SerializeObject(new { response.IsSuccessStatusCode, response.StatusCode });
+                        //if (response.IsSuccessStatusCode)
+                        //{
+                        //    string responseString = await response.Content.ReadAsStringAsync();
+                        //    return responseString;
+                        //}
+                        //else
+                        //{
+                        //    return string.Empty;
+                        //}
+                    }
+                }
+            }
+            catch (Exception ex)
+            {
+                return $"推送完成:请求响应超过{timeOut}秒,异常 {ex.Message}";
+            }
+        }
+
+
+
+        /// <summary>
+        /// 发起GET异步请求
+        /// </summary>
+        /// <typeparam name="T">返回类型</typeparam>
+        /// <param name="url">请求地址</param>
+        /// <param name="headers">请求头信息</param>
+        /// <param name="timeOut">请求超时时间，单位秒</param>
+        /// <returns>返回string</returns>
+        public async Task<string> GetAsync(string url, Dictionary<string, string> headers = null, int timeOut = 30)
+        {
+            var hostName = GetHostName(url);
+            using (HttpClient client = _httpClientFactory.CreateClient(hostName))
+            {
+                client.Timeout = TimeSpan.FromSeconds(timeOut);
+                if (headers?.Count > 0)
+                {
+                    foreach (string key in headers.Keys)
+                    {
+                        client.DefaultRequestHeaders.Add(key, headers[key]);
+                    }
+                }
+                using (HttpResponseMessage response = await client.GetAsync(url))
+                {
+                    if (response.IsSuccessStatusCode)
+                    {
+                        string responseString = await response.Content.ReadAsStringAsync();
+                        return responseString;
+                    }
+                    else
+                    {
+                        return string.Empty;
+                    }
+                }
+            }
+        }
+
+
+        /// <summary>
+        /// 发起POST异步请求
+        /// </summary>
+        /// <param name="url">请求地址</param>
+        /// <param name="body">POST提交的内容</param>
+        /// <param name="bodyMediaType">POST内容的媒体类型，如：application/xml、application/json</param>
+        /// <param name="responseContentType">HTTP响应上的content-type内容头的值,如：application/xml、application/json、application/text、application/x-www-form-urlencoded等</param>
+        /// <param name="headers">请求头信息</param>
+        /// <param name="timeOut">请求超时时间，单位秒</param>
+        /// <returns>返回string</returns>
+        public async Task<string> PostAsync(string url, string body, 
+            Dictionary<string, string> headers = null, 
+            int timeOut = 30, 
+            string bodyMediaType = "application/json",
+            string responseContentType = "application/json;charset=utf-8")
+        {
+
+            //content.Headers.ContentType = new MediaTypeHeaderValue("application/json");
+
+            //var clientHandler = new HttpClientHandler
+            //{
+            //    ServerCertificateCustomValidationCallback = (message, certificate2, arg3, arg4) => true
+            //};
+            //using (var client = new HttpClient(clientHandler))
+            //{
+            //    client.DefaultRequestHeaders.Accept.Add(new MediaTypeWithQualityHeaderValue("application/json"));
+
+            try
+            {
+                var hostName = GetHostName(url);
+                using (HttpClient client = _httpClientFactory.CreateClient(hostName))
+                {
+                    client.Timeout = TimeSpan.FromSeconds(timeOut);
+                    if (headers?.Count > 0)
+                    {
+                        foreach (string key in headers.Keys)
+                        {
+                            client.DefaultRequestHeaders.Add(key, headers[key]);
+                        }
+                    }
+                    StringContent content = new StringContent(body, System.Text.Encoding.UTF8, mediaType: bodyMediaType);
+                    if (!string.IsNullOrWhiteSpace(responseContentType))
+                    {
+                        content.Headers.ContentType = System.Net.Http.Headers.MediaTypeHeaderValue.Parse(responseContentType);
+                    }
+                    using (HttpResponseMessage response = await client.PostAsync(url, content))
+                    {
+                        if (response.IsSuccessStatusCode)
+                        {
+                            string responseString = await response.Content.ReadAsStringAsync();
+                            return responseString;
+                        }
+                        else
+                            return $"请求异常:{response.IsSuccessStatusCode},{response.StatusCode}";
+                    }
+                }
+            }
+            catch(Exception ex)
+            {
+                return $"请求异常:{ex.Message}";
+            }
+        }
+
+        /// <summary>
+        /// 发起POST异步请求
+        /// </summary>
+        /// <param name="url">请求地址</param>
+        /// <param name="body">POST提交的内容</param>
+        /// <param name="bodyMediaType">POST内容的媒体类型，如：application/xml、application/json</param>
+        /// <param name="responseContentType">HTTP响应上的content-type内容头的值,如：application/xml、application/json、application/text、application/x-www-form-urlencoded等</param>
+        /// <param name="headers">请求头信息</param>
+        /// <param name="timeOut">请求超时时间，单位秒</param>
+        /// <returns>返回string</returns>
+        public async Task<string> PutAsync(string url, string body,
+            string bodyMediaType = null,
+            string responseContentType = null,
+            Dictionary<string, string> headers = null,
+            int timeOut = 30)
+        {
+            var hostName = GetHostName(url);
+            using (HttpClient client = _httpClientFactory.CreateClient(hostName))
+            {
+                client.Timeout = TimeSpan.FromSeconds(timeOut);
+                if (headers?.Count > 0)
+                {
+                    foreach (string key in headers.Keys)
+                    {
+                        client.DefaultRequestHeaders.Add(key, headers[key]);
+                    }
+                }
+                StringContent content = new StringContent(body, System.Text.Encoding.UTF8, mediaType: bodyMediaType);
+                if (!string.IsNullOrWhiteSpace(responseContentType))
+                {
+                    content.Headers.ContentType = System.Net.Http.Headers.MediaTypeHeaderValue.Parse(responseContentType);
+                }
+                using (HttpResponseMessage response = await client.PutAsync(url, content))
+                {
+                    if (response.IsSuccessStatusCode)
+                    {
+                        string responseString = await response.Content.ReadAsStringAsync();
+                        return responseString;
+                    }
+                    else
+                    {
+                        return string.Empty;
+                    }
+                }
+            }
+        }
+
+        /// <summary>
+        /// 发起GET异步请求
+        /// </summary>
+        /// <typeparam name="T">返回类型</typeparam>
+        /// <param name="url">请求地址</param>
+        /// <param name="headers">请求头信息</param>
+        /// <param name="timeOut">请求超时时间，单位秒</param>
+        /// <returns>返回string</returns>
+        public async Task<string> DeleteAsync(string url, Dictionary<string, string> headers = null, int timeOut = 30)
+        {
+            var hostName = GetHostName(url);
+            using (HttpClient client = _httpClientFactory.CreateClient(hostName))
+            {
+                client.Timeout = TimeSpan.FromSeconds(timeOut);
+                if (headers?.Count > 0)
+                {
+                    foreach (string key in headers.Keys)
+                    {
+                        client.DefaultRequestHeaders.Add(key, headers[key]);
+                    }
+                }
+                using (HttpResponseMessage response = await client.DeleteAsync(url))
+                {
+                    if (response.IsSuccessStatusCode)
+                    {
+                        string responseString = await response.Content.ReadAsStringAsync();
+                        return responseString;
+                    }
+                    else
+                    {
+                        return string.Empty;
+                    }
+                }
+            }
+        }
+
+        /// <summary>
+        /// 发起GET异步请求
+        /// </summary>
+        /// <typeparam name="T">返回类型</typeparam>
+        /// <param name="url">请求地址</param>
+        /// <param name="headers">请求头信息</param>
+        /// <param name="timeOut">请求超时时间，单位秒</param>
+        /// <returns>返回T</returns>
+        public async Task<T> GetAsync<T>(string url, Dictionary<string, string> headers = null, int timeOut = 30) where T : new()
+        {
+            string responseString = await GetAsync(url, headers, timeOut);
+            if (!string.IsNullOrWhiteSpace(responseString))
+            {
+                return JsonConvert.DeserializeObject<T>(responseString);
+            }
+            else
+            {
+                return default(T);
+            }
+        }
+
+
+        /// <summary>
+        /// 发起POST异步请求
+        /// </summary>
+        /// <typeparam name="T">返回类型</typeparam>
+        /// <param name="url">请求地址</param>
+        /// <param name="body">POST提交的内容</param>
+        /// <param name="bodyMediaType">POST内容的媒体类型，如：application/xml、application/json</param>
+        /// <param name="responseContentType">HTTP响应上的content-type内容头的值,如：application/xml、application/json、application/text、application/x-www-form-urlencoded等</param>
+        /// <param name="headers">请求头信息</param>
+        /// <param name="timeOut">请求超时时间，单位秒</param>
+        /// <returns>返回T</returns>
+        public async Task<T> PostAsync<T>(string url, string body,
+            Dictionary<string, string> headers = null, 
+            int timeOut = 30, 
+            string bodyMediaType = "application/json", 
+            string responseContentType = "application/json;charset=utf-8"
+        ) where T : new()
+        {
+            string responseString = await PostAsync(url, body, headers, timeOut, bodyMediaType, responseContentType);
+            if (!string.IsNullOrWhiteSpace(responseString))
+            {
+                return JsonConvert.DeserializeObject<T>(responseString);
+            }
+            else
+            {
+                return default(T);
+            }
+        }
+
+        /// <summary>
+        /// 发起PUT异步请求
+        /// </summary>
+        /// <typeparam name="T">返回类型</typeparam>
+        /// <param name="url">请求地址</param>
+        /// <param name="body">POST提交的内容</param>
+        /// <param name="bodyMediaType">POST内容的媒体类型，如：application/xml、application/json</param>
+        /// <param name="responseContentType">HTTP响应上的content-type内容头的值,如：application/xml、application/json、application/text、application/x-www-form-urlencoded等</param>
+        /// <param name="headers">请求头信息</param>
+        /// <param name="timeOut">请求超时时间，单位秒</param>
+        /// <returns>返回T</returns>
+        public async Task<T> PutAsync<T>(string url, string body,
+            string bodyMediaType = null,
+            string responseContentType = null,
+            Dictionary<string, string> headers = null,
+            int timeOut = 30) where T : new()
+        {
+            string responseString = await PutAsync(url, body, bodyMediaType, responseContentType, headers, timeOut);
+            if (!string.IsNullOrWhiteSpace(responseString))
+            {
+                return JsonConvert.DeserializeObject<T>(responseString);
+            }
+            else
+            {
+                return default(T);
+            }
+        }
+
+        /// <summary>
+        /// 发起DELETE异步请求
+        /// </summary>
+        /// <typeparam name="T">返回类型</typeparam>
+        /// <param name="url">请求地址</param>
+        /// <param name="headers">请求头信息</param>
+        /// <param name="timeOut">请求超时时间，单位秒</param>
+        /// <returns>返回T</returns>
+        public async Task<T> DeleteAsync<T>(string url, Dictionary<string, string> headers = null, int timeOut = 30) where T : new()
+        {
+            string responseString = await DeleteAsync(url, headers, timeOut);
+            if (!string.IsNullOrWhiteSpace(responseString))
+            {
+                return JsonConvert.DeserializeObject<T>(responseString);
+            }
+            else
+            {
+                return default(T);
+            }
+        }
+        #region 私有函数
+
+        /// <summary>
+        /// 获取请求的主机名
+        /// </summary>
+        /// <param name="url"></param>
+        /// <returns></returns>
+        private static string GetHostName(string url)
+        {
+            if (!string.IsNullOrWhiteSpace(url))
+            {
+                return url.Replace("https://", "").Replace("http://", "").Split('/')[0];
+            }
+            else
+            {
+                return "AnyHost";
+            }
+        }
+
+        #endregion
+
+
+        #endregion
+
+        #region  同步
+
+        /// <summary>
+        /// 发起GET同步请求
+        /// </summary>
+        /// <param name="url"></param>
+        /// <param name="headers"></param>
+        /// <param name="contentType"></param>
+        /// <returns></returns>
+        /// 
+
+
+        public string HttpGet(string url, Dictionary<string, string> headers = null, string contentType = "application/json;charset=utf-8")
+        {
+            if (string.IsNullOrEmpty(url)) return "";
+            try
+            {
+                using (HttpClient client = new HttpClient())
+                {
+                    if (contentType != null)
+                        client.DefaultRequestHeaders.Add("ContentType", contentType);
+                    if (headers != null)
+                    {
+                        foreach (var header in headers)
+                            client.DefaultRequestHeaders.Add(header.Key, header.Value);
+                    }
+                    HttpResponseMessage response = client.GetAsync(url).Result;
+                    return response.Content.ReadAsStringAsync().Result;
+                }
+            }
+            catch (Exception ex)
+            {
+                _logger.LogDebug($"HttpGet/URL:{url},headers:{JsonConvert.SerializeObject(headers)},异常:{ex.Message}|{ex.Source}|{ex.StackTrace}");
+            }
+            return "";
+        }
+        /// <summary>
+        /// 发起POST同步请求
+        /// </summary>
+        /// <param name="url"></param>
+        /// <param name="postData"></param>
+        /// <param name="contentType">application/xml、application/json、application/text、application/x-www-form-urlencoded</param>
+        /// <param name="headers">填充消息头</param>        
+        /// <returns></returns>
+        public string HttpPost(string url, string postData = null, Dictionary<string, string> headers = null, string contentType = "application/json")
+        {
+            if (string.IsNullOrEmpty(url)) return "";
+            try
+            {
+                postData = postData ?? "";
+                using (HttpClient client = new HttpClient())
+                {
+                    if (headers != null)
+                    {
+                        foreach (var header in headers)
+                            client.DefaultRequestHeaders.Add(header.Key, header.Value);
+                    }
+                    using (HttpContent httpContent = new StringContent(postData, Encoding.UTF8))
+                    {
+                        if (contentType != null)
+                            httpContent.Headers.ContentType = new MediaTypeHeaderValue(contentType);
+
+                        HttpResponseMessage response = client.PostAsync(url, httpContent).Result;
+                        return response.Content.ReadAsStringAsync().Result;
+                    }
+                }
+            }
+            catch  (Exception ex){
+                _logger.LogDebug($"HttpPost/URL:{url},postStr:{postData},headers:{JsonConvert.SerializeObject(headers)},异常:{ex.Message}|{ex.Source}|{ex.StackTrace}");
+            }
+            return "";
+        }
+ 
+
+        #endregion
+    }
+}

TelpoPush.WanJiaAn.Worker/Common/LimitedConcurrencyLevelTaskScheduler.cs

@@ -0,0 +1,137 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace TelpoPush.WanJiaAn.Worker.Common
+{
+    /// <summary>
+    /// Provides a task scheduler that ensures a maximum concurrency level while
+    /// running on top of the ThreadPool.
+    /// </summary>
+    public class LimitedConcurrencyLevelTaskScheduler : TaskScheduler
+    {
+        /// <summary>Whether the current thread is processing work items.</summary>
+        [ThreadStatic]
+        private static bool _currentThreadIsProcessingItems;
+        /// <summary>The list of tasks to be executed.</summary>
+        private readonly LinkedList<Task> _tasks = new LinkedList<Task>(); // protected by lock(_tasks)
+        /// <summary>The maximum concurrency level allowed by this scheduler.</summary>
+        private readonly int _maxDegreeOfParallelism;
+        /// <summary>Whether the scheduler is currently processing work items.</summary>
+        private int _delegatesQueuedOrRunning = 0; // protected by lock(_tasks)
+
+        /// <summary>
+        /// Initializes an instance of the LimitedConcurrencyLevelTaskScheduler class with the
+        /// specified degree of parallelism.
+        /// </summary>
+        /// <param name="maxDegreeOfParallelism">The maximum degree of parallelism provided by this scheduler.</param>
+        public LimitedConcurrencyLevelTaskScheduler(int maxDegreeOfParallelism)
+        {
+            if (maxDegreeOfParallelism < 1) throw new ArgumentOutOfRangeException("maxDegreeOfParallelism");
+            _maxDegreeOfParallelism = maxDegreeOfParallelism;
+        }
+
+        /// <summary>Queues a task to the scheduler.</summary>
+        /// <param name="task">The task to be queued.</param>
+        protected sealed override void QueueTask(Task task)
+        {
+            // Add the task to the list of tasks to be processed.  If there aren't enough
+            // delegates currently queued or running to process tasks, schedule another.
+            lock (_tasks)
+            {
+                _tasks.AddLast(task);
+                if (_delegatesQueuedOrRunning < _maxDegreeOfParallelism)
+                {
+                    ++_delegatesQueuedOrRunning;
+                    NotifyThreadPoolOfPendingWork();
+                }
+            }
+        }
+
+        /// <summary>
+        /// Informs the ThreadPool that there's work to be executed for this scheduler.
+        /// </summary>
+        private void NotifyThreadPoolOfPendingWork()
+        {
+            ThreadPool.UnsafeQueueUserWorkItem(_ =>
+            {
+                // Note that the current thread is now processing work items.
+                // This is necessary to enable inlining of tasks into this thread.
+                _currentThreadIsProcessingItems = true;
+                try
+                {
+                    // Process all available items in the queue.
+                    while (true)
+                    {
+                        Task item;
+                        lock (_tasks)
+                        {
+                            // When there are no more items to be processed,
+                            // note that we're done processing, and get out.
+                            if (_tasks.Count == 0)
+                            {
+                                --_delegatesQueuedOrRunning;
+                                break;
+                            }
+
+                            // Get the next item from the queue
+                            item = _tasks.First.Value;
+                            _tasks.RemoveFirst();
+                        }
+
+                        // Execute the task we pulled out of the queue
+                        base.TryExecuteTask(item);
+                    }
+                }
+                // We're done processing items on the current thread
+                finally { _currentThreadIsProcessingItems = false; }
+            }, null);
+        }
+
+        /// <summary>Attempts to execute the specified task on the current thread.</summary>
+        /// <param name="task">The task to be executed.</param>
+        /// <param name="taskWasPreviouslyQueued"></param>
+        /// <returns>Whether the task could be executed on the current thread.</returns>
+        protected sealed override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
+        {
+            // If this thread isn't already processing a task, we don't support inlining
+            if (!_currentThreadIsProcessingItems) return false;
+
+            // If the task was previously queued, remove it from the queue
+            if (taskWasPreviouslyQueued) TryDequeue(task);
+
+            // Try to run the task.
+            return base.TryExecuteTask(task);
+        }
+
+        /// <summary>Attempts to remove a previously scheduled task from the scheduler.</summary>
+        /// <param name="task">The task to be removed.</param>
+        /// <returns>Whether the task could be found and removed.</returns>
+        protected sealed override bool TryDequeue(Task task)
+        {
+            lock (_tasks) return _tasks.Remove(task);
+        }
+
+        /// <summary>Gets the maximum concurrency level supported by this scheduler.</summary>
+        public sealed override int MaximumConcurrencyLevel { get { return _maxDegreeOfParallelism; } }
+
+        /// <summary>Gets an enumerable of the tasks currently scheduled on this scheduler.</summary>
+        /// <returns>An enumerable of the tasks currently scheduled.</returns>
+        protected sealed override IEnumerable<Task> GetScheduledTasks()
+        {
+            bool lockTaken = false;
+            try
+            {
+                Monitor.TryEnter(_tasks, ref lockTaken);
+                if (lockTaken) return _tasks.ToArray();
+                else throw new NotSupportedException();
+            }
+            finally
+            {
+                if (lockTaken) Monitor.Exit(_tasks);
+            }
+        }
+    }
+}

TelpoPush.WanJiaAn.Worker/Common/TimeHelper.cs

@@ -0,0 +1,33 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace TelpoPush.WanJiaAn.Worker.Common
+{
+    public class TimeHelper
+    {
+        /// <summary>
+        /// 时间戳转成时间类型
+        /// </summary>
+        /// <param name="timeStamp"></param>
+        /// <returns></returns>
+        public static DateTime ConvertToLocalDateTime(string timeStamp)
+        {
+            DateTime dtStart = TimeZoneInfo.ConvertTime(new DateTime(1970, 1, 1), TimeZoneInfo.Utc, TimeZoneInfo.Local);
+            if (timeStamp.Length == 13)
+            {
+                long lTime = long.Parse(timeStamp + "0000");
+                TimeSpan toNow = new TimeSpan(lTime);
+                return dtStart.Add(toNow);
+            }
+            return dtStart.AddSeconds(long.Parse(timeStamp));
+        }
+
+        public static string ToDateTimeStr(DateTime dt)
+        {
+            return dt.ToString("yyyy-MM-dd HH:mm:ss");
+        }
+    }
+}

TelpoPush.WanJiaAn.Worker/Common/Utils.cs

@@ -0,0 +1,40 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using System.Security.Cryptography;
+
+namespace TelpoPush.WanJiaAn.Worker.Common
+{
+    public class Utils
+    {
+        public static MultipartFormDataContent GetMultipartFormDataContent(Dictionary<string, object> dic, string appId, ref Dictionary<string, object> outDic)
+        {
+            MultipartFormDataContent mfdc = new MultipartFormDataContent();
+            StringBuilder sb = new StringBuilder();
+            if (dic != null && dic.Count > 0)
+            {
+                var dicOrderBy = dic.OrderBy(z => z.Key);
+                foreach (KeyValuePair<string, object> kv in dicOrderBy)
+                {
+                    sb.Append($"{kv.Key}={kv.Value.ToString()}&");
+                    mfdc.Add(new StringContent(kv.Value.ToString()), kv.Key);//参数, 内容在前,参数名称在后
+                }
+            }
+            string signStr = $"{sb.ToString().Trim('&')}{appId}";
+            byte[] bytes = Encoding.UTF8.GetBytes(signStr);
+            byte[] hash = SHA256.Create().ComputeHash(bytes);
+            StringBuilder builder = new StringBuilder();
+            for (int i = 0; i < hash.Length; i++)
+            {
+                builder.Append(hash[i].ToString("X2"));
+            }
+            string sign = builder.ToString().ToLower();
+            dic.Add("sign", sign);
+            mfdc.Add(new StringContent(sign), "sign");//参数, 内容在前,参数名称在后
+            outDic = dic;
+            return mfdc;
+        }
+    }
+}

TelpoPush.WanJiaAn.Worker/Dockerfile

@@ -0,0 +1,33 @@
+#See https://aka.ms/customizecontainer to learn how to customize your debug container and how Visual Studio uses this Dockerfile to build your images for faster debugging.
+
+FROM mcr.microsoft.com/dotnet/sdk:8.0-alpine AS base
+WORKDIR /app
+
+FROM mcr.microsoft.com/dotnet/sdk:8.0-alpine AS build
+WORKDIR /src
+COPY ["TelpoPush.WanJiaAn.Worker/TelpoPush.WanJiaAn.Worker.csproj", "TelpoPush.WanJiaAn.Worker/"]
+
+
+COPY ["nuget.config","."]
+RUN dotnet nuget remove source nuget.org
+RUN dotnet nuget add source https://repo.huaweicloud.com/repository/nuget/v3/index.json -n huaweicloud_nuget
+RUN dotnet restore "TelpoPush.WanJiaAn.Worker/TelpoPush.WanJiaAn.Worker.csproj"
+COPY . .
+WORKDIR "/src/TelpoPush.WanJiaAn.Worker"
+
+
+FROM build AS publish
+RUN dotnet publish "TelpoPush.WanJiaAn.Worker.csproj" -c Release -o /app/publish
+
+
+FROM base AS final
+WORKDIR /app
+COPY --from=publish /app/publish .
+#COPY pem /app/pem
+ENV environment=Development
+RUN apk add --no-cache tzdata
+ENV TimeZone=Asia/Shanghai
+ENV LANG C.UTF-8
+ENV ASPNETCORE_ENVIRONMENT_LOG=Development
+RUN ln -snf /usr/share/zoneinfo/$TimeZone /etc/localtime && echo $TimeZone > /etc/timezone
+ENTRYPOINT dotnet TelpoPush.WanJiaAn.Worker.dll --environment=$environment

TelpoPush.WanJiaAn.Worker/Handlers/KafkaSubscribe.cs

@@ -0,0 +1,62 @@
+using Confluent.Kafka;
+using TelpoPush.WanJiaAn.Worker.Common;
+using TelpoPush.WanJiaAn.Worker.Service.Mq;
+
+namespace TelpoPush.WanJiaAn.Worker.Handlers
+{
+    public class KafkaSubscribe
+    {
+        private readonly ILogger<KafkaSubscribe> _logger;
+        private readonly IHostEnvironment _env;
+        private readonly IKafkaService _kafkaService;
+        private readonly WanJiaAnProcess _WanJiaAnProcess;
+
+
+        public KafkaSubscribe(
+            ILogger<KafkaSubscribe> logger, IHostEnvironment env,
+            IKafkaService kafkaService,
+            WanJiaAnProcess WanJiaAnProcess)
+        {
+            _logger = logger;
+            _env = env;
+            _kafkaService = kafkaService;
+            _WanJiaAnProcess = WanJiaAnProcess;
+        }
+        public async Task SubscribeAsync()
+        {
+#if DEBUG
+
+            _logger.LogInformation("11312");
+
+            var temp = new Headers();
+            string topic = "topic.push.WanJiaAn";
+            //temp.Add(new Header("DataType", new byte[] { 0, 0, 0, 0 }));
+            //temp.Add(new Header("AlarmType", new byte[] { 2, 0, 0, 0 }));
+            //string psych = "{\"messageId\":\"1790941606816612864\",\"topic\":\"topic.push.third\",\"time\":\"2024-05-16 11:05:27\",\"data\":{\"imei\":\"861281060093147\",\"atteryLowId\":\"861281060093147664577f9\",\"info\":\"设备电量低于15%\"}}";
+            //await _WanJiaAnProcess.SendWanJiaAn(psych, topic, temp);
+
+            //// await  _kafkaService.SubscribeAsync(DoReceive, CancellationToken.None);
+
+#else
+      
+            LimitedConcurrencyLevelTaskScheduler lcts = new LimitedConcurrencyLevelTaskScheduler(5);
+            TaskFactory factory = new TaskFactory(lcts);
+            try
+            {
+                await factory.StartNew(async () =>
+                {
+                 await  _kafkaService.SubscribeAsync(DoReceive, CancellationToken.None);
+                });
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError($"Subscribe 处理Kafka数据发生异常 {ex.Message}|{ex.Source}|{ex.StackTrace}");
+            }
+#endif
+        }
+        async void DoReceive(string topic, string message, Headers headers)
+        {
+            await _WanJiaAnProcess.SendWanJiaAn(message, topic, headers);
+        }
+    }
+}

TelpoPush.WanJiaAn.Worker/Handlers/WanJiaAnProcess.cs

@@ -0,0 +1,292 @@
+using Confluent.Kafka;
+using Microsoft.Extensions.Options;
+using Newtonsoft.Json;
+using TelpoPush.WanJiaAn.Worker.Common;
+using TelpoPush.WanJiaAn.Worker.Models.Config;
+using TelpoPush.WanJiaAn.Worker.Models.Enum;
+using TelpoPush.WanJiaAn.Worker.Models.MqTemplates;
+using TelpoPush.WanJiaAn.Worker.Models.PushTemplates;
+using TelpoPush.WanJiaAn.Worker.Service.Cache;
+using TelpoPush.WanJiaAn.Worker.Service.Mq;
+
+namespace TelpoPush.WanJiaAn.Worker.Handlers
+{
+    public class WanJiaAnProcess
+    {
+        private readonly static object _syncLocker = new object();
+        private readonly IHostEnvironment _env;
+        private readonly ILogger<WanJiaAnProcess> _logger;
+        private readonly HttpHelperAsync _httpHelper;
+        private readonly RedisUtil _redis;
+        private readonly MqProcessMessage _serviceMqProcess;
+        private readonly WanJiaAnConfig _WanJiaAnConfig;
+
+        public WanJiaAnProcess(
+            IHostEnvironment env,
+            ILogger<WanJiaAnProcess> logger,
+            HttpHelperAsync httpHelper,
+            RedisUtil redis,
+             MqProcessMessage serviceMqProcess,
+              IOptions<WanJiaAnConfig> WanJiaAnConfig
+           )
+        {
+            _env = env;
+            _logger = logger;
+            _httpHelper = httpHelper;
+            _redis = redis;
+            _WanJiaAnConfig = WanJiaAnConfig.Value;
+            _serviceMqProcess = serviceMqProcess;
+
+        }
+
+        public async Task SendWanJiaAn(string? message, string topic, Headers headers)
+        {
+            #region 数据初始验证
+            bool isHandle = true;
+            BaseModel model = null;
+            string imei = "";
+            if (!string.IsNullOrEmpty(message))
+            {
+                model = JsonConvert.DeserializeObject<BaseModel>(message);
+                if (model != null)
+                {
+                    var Jo = JsonConvert.DeserializeObject<Dictionary<string, object>>(model.data.ToString());
+                    if (Jo.ContainsKey("device_id"))
+                        imei = Jo["device_id"].ToString();
+   
+                    if (string.IsNullOrEmpty(imei))
+                    {
+                        _logger.LogInformation($"[数据信息不完整] imei信息不存在：{message}");
+                        isHandle = false;
+                    }
+                    else
+                        await _redis.GetGpsDevice(imei);
+                }
+                else
+                {
+                    _logger.LogInformation($"[数据信息不完整] 数据解析异常：{message}");
+                    isHandle = false;
+                }
+            }
+            else
+            {
+                _logger.LogInformation($"[数据信息不完整] message数据异常：{message}");
+                isHandle = false;
+            }
+            #endregion
+            if (isHandle)
+            {
+                lock (_syncLocker)
+                {
+                    //Headers 解析
+                    HeadersDto headersDto = new HeadersDto();
+                    try
+                    {
+                        foreach (var item in headers)
+                        {
+                            if (item.Key == KafkaHeader.DataType)
+                                headersDto.DataType = BitConverter.ToInt32(item.GetValueBytes(), 0);
+                            else if (item.Key == KafkaHeader.AlarmType)
+                                headersDto.AlarmType = BitConverter.ToInt32(item.GetValueBytes(), 0);
+                            else if (item.Key == KafkaHeader.OperType)
+                                headersDto.OperType = BitConverter.ToInt32(item.GetValueBytes(), 0);
+                        }
+                    }
+                    catch (Exception ex)
+                    {
+                        _logger.LogError($"当前工作线程Headers异常,{ex.Message}|{ex.Source}|{ex.StackTrace}");
+                    }
+                    try
+                    {
+                        #region  注释
+                        //string dataType = headersDto.DataType != null ? "_" + headersDto.DataType : "";
+                        //string alarmType = headersDto.AlarmType != null ? "_" + headersDto.AlarmType : "";
+                        //string operType = headersDto.OperType != null ? "_" + headersDto.OperType : "";
+                        //string key = dataType + alarmType + operType;
+                        //var dataStatus = _redis.IsDateStatus(model, imei, key).Result;
+                        //过滤
+                        //if (headersDto.DataType == (int)MqDataType.TemperatureInfo
+                        //    || headersDto.DataType == (int)MqDataType.Temperature1Info
+                        //    || headersDto.DataType == (int)MqDataType.BindDevice
+                        //    || headersDto.DataType == (int)MqDataType.WanJiaAnInfo
+                        //    || headersDto.DataType == (int)MqDataType.HeartRateInfo
+                        //    || headersDto.DataType == (int)MqDataType.HeartRate1Info
+                        //    || headersDto.DataType == (int)MqDataType.Spo2Info
+                        //    || headersDto.DataType == (int)MqDataType.Spo21Info
+                        //    || headersDto.DataType == (int)MqDataType.BloodPressInfo
+                        //    || headersDto.DataType == (int)MqDataType.BloodPress1Info
+                        //    || headersDto.DataType == (int)MqDataType.SportResult
+                        //    || headersDto.DataType == (int)MqDataType.BloodSugar
+                        //    )
+                        //    dataStatus.isPush = true;
+                        //dataStatus.isPush = true;
+                        //if (dataStatus.isPush)
+                        //{
+                        //switch (topic)
+                        //{
+                        //    case "topic.push.third":
+                        //        switch (headersDto.DataType)
+                        //        {
+                        //            case (int)MqDataType.AlarmInfo:  //报警消息
+                        //                break;
+                        //            case (int)MqDataType.TemperatureInfo: //体温消息
+                        //                break;
+                        //            case (int)MqDataType.WanJiaAnInfo: //定位消息
+                        //               // DataServicePusWanJiaAn(model, imei);
+                        //                break;
+                        //            case (int)MqDataType.StepInfo: //步数消息
+
+                        //                break;
+                        //            case (int)MqDataType.BatteryLevelInfo:  //电量消息
+                        //                break;
+                        //            case (int)MqDataType.DeviceCallLog:  //设备通话记录
+
+                        //                break;
+                        //            case (int)MqDataType.DeviceSmsLog:  //设备短信记录
+
+                        //                break;
+                        //            case (int)MqDataType.DeviceConfigInfo:  //设备配置信息
+
+                        //                break;
+                        //            case (int)MqDataType.Status: //设备状态（offline,online）
+
+                        //                break;
+                        //            case (int)MqDataType.Active:  //设备激活状态
+                        //                break;
+                        //            case (int)MqDataType.reply:  //指令回调
+
+                        //                break;
+                        //            case (int)MqDataType.Weather:  //天气查询
+
+                        //                break;
+                        //            case (int)MqDataType.ReadMsg:  //短消息阅读
+
+                        //                break;
+                        //            case (int)MqDataType.StudyAINotifyStatusUpload:  //学习能力状态
+
+                        //                break;
+                        //            case (int)MqDataType.HeartRateInfo:  //心率
+
+                        //                break;
+                        //            case (int)MqDataType.HeartRate1Info:  //周期性心率
+
+                        //                break;
+                        //            case (int)MqDataType.Spo2Info:  //血氧
+
+                        //                break;
+                        //            case (int)MqDataType.Spo21Info:  //周期性血氧
+
+                        //                break;
+                        //            case (int)MqDataType.Temperature1Info:  //周期性报体温数据
+
+                        //                break;
+                        //            case (int)MqDataType.DrownReportInfo:  //防溺水告警 
+
+                        //                break;
+                        //            case (int)MqDataType.WearStatusInfo:  //手表未佩戴告警 
+
+                        //                break;
+                        //            case (int)MqDataType.BloodPressInfo:  //血压 
+
+                        //                break;
+                        //            case (int)MqDataType.BloodPress1Info:  //周期性血压 
+
+                        //                break;
+                        //            case (int)MqDataType.PsychInfo:  //心理监测
+
+                        //                break;
+                        //            case (int)MqDataType.AiCallResult:  //AI呼叫结果回调
+
+                        //            case (int)MqDataType.CrossBorder:  //越界上报(围栏进出告警)
+
+                        //                break;
+                        //            case (int)MqDataType.SportResult:  //运动数据上报
+
+                        //                break;
+                        //            case (int)MqDataType.BindDevice:  //绑定业务
+
+                        //                break;
+                        //            case (int)MqDataType.BloodSugar:  //血糖业务
+
+                        //                break;
+                        //            default:
+                        //                break;
+                        //        }
+                        //        break;
+                        //    default:
+                        //        break;
+                        //}
+                        //}
+                        //else
+                        //    _logger.LogInformation($"数据未处理(历史数据):{JsonConvert.SerializeObject(dataStatus)}");
+
+                        #endregion
+
+                        switch (topic)
+                        {
+                            case "topic.wanjiaan.push.telpo":
+                                switch (headersDto.DataType)
+                                {
+                                    case (int)MqDataType.WanJiaAnInfo: //定位消息
+                                        DataServicePusWanJiaAn(model, imei);
+                                        break;
+                                    default:
+                                        break;
+                                }
+                                break;
+                            default:
+                                break;
+                        }
+
+                    }
+                    catch (Exception ex)
+                    {
+                        _logger.LogError($"当前工作线程异常: {ex.Message}|{ex.Source}|{ex.StackTrace}");
+                    }
+                }
+            }
+        }
+
+        //位置
+        public async Task DataServicePusWanJiaAn(BaseModel model,string imei)
+        {
+            if (model.data != null)
+            {
+                var alarmEvent = JsonConvert.DeserializeObject<MqAlarmEventTemplate>(model.data.ToString());
+                if (alarmEvent?.alarm_event != null)
+                {
+                    var data = alarmEvent?.alarm_event;
+                    if (data?.event_type == 2 || data?.event_type == 52)//2:哭声检测;52:遮脸提醒
+                    {
+                        string event_id = data.event_id;
+                        string device_id = model.device_id;
+                        string user_id = model.user_id;
+                        int event_type = data.event_type;
+                        int report_type = data.report_type;
+                        DateTime event_time = TimeHelper.ConvertToLocalDateTime(model.event_time);
+                        DateTime event_start = TimeHelper.ConvertToLocalDateTime(data.event_start);
+                        if (!string.IsNullOrEmpty(data.event_end))
+                        {
+                            DateTime event_end = TimeHelper.ConvertToLocalDateTime(data.event_end);
+                        }
+                        if (!string.IsNullOrEmpty(data.video_start))
+                        {
+                            DateTime video_start = TimeHelper.ConvertToLocalDateTime(data.video_start);
+                        }
+                        if (!string.IsNullOrEmpty(data.video_end))
+                        {
+                            DateTime video_end = TimeHelper.ConvertToLocalDateTime(data.video_end);
+                        }
+                        string image_url = data.image;
+                        string video_url = data.url;
+                        string remark = JsonConvert.SerializeObject(model);
+                        DateTime create_time = DateTime.Now;
+
+                        var obj = new { };
+                    }
+                }
+
+            }
+        }
+    }
+}

TelpoPush.WanJiaAn.Worker/Models/CacheTemplates/DeviceInfoModel.cs

@@ -0,0 +1,19 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace TelpoPush.WanJiaAn.Worker.Models.CacheTemplates
+{
+    public class DeviceInfoModel
+    {
+        public string deviceId { get; set; }
+        public string imei { get; set; }
+        public string deviceName { get; set; }
+        public string orgId { get; set; }
+        public string apiUid { get; set; }
+        public string activeStatus { get; set; }
+        public string activeTime { get; set; }
+    }
+}

TelpoPush.WanJiaAn.Worker/Models/CacheTemplates/ManufactorPushSettingInfoModel.cs

@@ -0,0 +1,35 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace TelpoPush.WanJiaAn.Worker.Models.CacheTemplates
+{
+
+    public class ManufactorPushSettingInfoModel
+    {
+        public string id { get; set; }
+        public string manufactorId { get; set; }
+        public string manufactorName { get; set; }
+        public string imei { get; set; }
+        public List<PushSettingsItem> pushSettings { get; set; }
+        public int settingType { get; set; }
+    }
+    public class PushSettingsItem
+    {
+        public string dataName { get; set; }
+        public int dataType { get; set; }
+        public List<SettingInfosItem> settingInfos { get; set; }
+    }
+    public class SettingInfosItem
+    {
+        public bool pushFlag { get; set; }
+        public string pushStartTime { get; set; }
+        public string pushEndTime { get; set; }
+        public int pushType { get; set; }
+        public string pushUrl { get; set; }
+        public string remark { get; set; }
+    }
+
+}

TelpoPush.WanJiaAn.Worker/Models/CacheTemplates/PersonInfoModel.cs

@@ -0,0 +1,42 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace TelpoPush.WanJiaAn.Worker.Models.CacheTemplates
+{
+
+    public class PersonInfoModel
+    {
+        public PersonModel person { get; set; }
+        public string time { get; set; }
+    }
+    public class PersonModel
+    {
+        public string personId { get; set; }
+        public string serialno { get; set; }
+        public string personName { get; set; }
+        public string deviceId { get; set; }
+        public string nickName { get; set; }
+        public bool gender { get; set; }
+        public int height { get; set; }
+        public int weight { get; set; }
+        public string bornDate { get; set; }
+        public string school { get; set; }
+        public string grade { get; set; }
+        public string className { get; set; }
+        public string imagePath { get; set; }
+        public string imagePathSmall { get; set; }
+        public int age { get; set; }
+        public string createTime { get; set; }
+        public string remarks { get; set; }
+        public int ishypertension { get; set; }
+        public string emotion { get; set; }
+        public int profession { get; set; }
+        public int regularity { get; set; }
+        public int chronicDisease { get; set; }
+        public string apiUid { get; set; }
+        public string apiRemark { get; set; }
+    }
+}

TelpoPush.WanJiaAn.Worker/Models/Config/RedisConfig.cs

@@ -0,0 +1,94 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace TelpoPush.WanJiaAn.Worker.Models.Config
+{
+    /// <summary>
+    /// Redis配置模板类
+    /// </summary>
+    public class RedisConfig
+    {
+        public string Server { get; set; }
+        /// <summary>
+        /// Redis server password
+        /// </summary>
+        public string Password { get; set; }
+        /// <summary>
+        /// Redis server database, default 0
+        /// </summary>
+        public int? DefaultDatabase { get; set; }
+        /// <summary>
+        /// The asynchronous method automatically uses pipeline, and the 10W concurrent time is 450ms(welcome to feedback)
+        /// </summary>
+        public bool? AsyncPipeline { get; set; }
+        /// <summary>
+        /// Connection pool size, default 50
+        /// </summary>
+        public int? Poolsize { get; set; }
+        /// <summary>
+        /// Idle time of elements in the connection pool(MS), suitable for connecting to remote redis server, default 20000
+        /// </summary>
+        public int? IdleTimeout { get; set; }
+        /// <summary>
+        /// Connection timeout(MS), default 5000
+        /// </summary>
+        public int? ConnectTimeout { get; set; }
+        /// <summary>
+        /// Send / receive timeout(MS), default 10000
+        /// </summary>
+        public int? SyncTimeout { get; set; }
+        /// <summary>
+        /// Preheat connections, receive values such as preheat = 5 preheat 5 connections, default 5
+        /// </summary>
+        public int? Preheat { get; set; }
+        /// <summary>
+        /// Follow system exit event to release automatically, default true
+        /// </summary>
+        public bool? AutoDispose { get; set; }
+        /// <summary>
+        /// Enable encrypted transmission, default false
+        /// </summary>
+        public bool? Ssl { get; set; }
+        /// <summary>
+        /// 是否尝试集群模式，阿里云、腾讯云集群需要设置此选项为 false, default true
+        /// </summary>
+        public bool? Testcluster { get; set; }
+        /// <summary>
+        /// Execution error, retry attempts, default 0
+        /// </summary>
+        public int? Tryit { get; set; }
+        /// <summary>
+        /// Connection name, use client list command to view
+        /// </summary>
+        public string Name { get; set; }
+        /// <summary>
+        /// key前辍，所有方法都会附带此前辍，csredis.Set(prefix + "key", 111)
+        /// </summary>
+        public string Prefix { get; set; }
+
+        public override string ToString()
+        {
+            if (string.IsNullOrWhiteSpace(Server)) throw new ArgumentNullException(nameof(Server));
+            var sb = new StringBuilder(Server);
+            if (!string.IsNullOrWhiteSpace(Password)) sb.Append($",password={Password}");
+            if (DefaultDatabase.HasValue) sb.Append($",defaultDatabase={DefaultDatabase}");
+            if (AsyncPipeline.HasValue) sb.Append($",asyncPipeline={AsyncPipeline}");
+            if (Poolsize.HasValue) sb.Append($",poolsize={Poolsize}");
+            if (IdleTimeout.HasValue) sb.Append($",idleTimeout={IdleTimeout}");
+            if (ConnectTimeout.HasValue) sb.Append($",connectTimeout={ConnectTimeout}");
+            if (SyncTimeout.HasValue) sb.Append($",syncTimeout={0}");
+            if (Preheat.HasValue) sb.Append($",preheat={Preheat}");
+            if (AutoDispose.HasValue) sb.Append($",autoDispose={AutoDispose}");
+            if (Ssl.HasValue) sb.Append($",ssl={Ssl}");
+            if (Testcluster.HasValue) sb.Append($",testcluster={Testcluster}");
+            if (Tryit.HasValue) sb.Append($",tryit={Tryit}");
+            if (!string.IsNullOrWhiteSpace(Name)) sb.Append($",name={Name}");
+            if (!string.IsNullOrWhiteSpace(Prefix)) sb.Append($",prefix={Prefix}");
+
+            return sb.ToString();
+        }
+    }
+}

TelpoPush.WanJiaAn.Worker/Models/Config/ServiceConfig.cs

@@ -0,0 +1,31 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace TelpoPush.WanJiaAn.Worker.Models.Config
+{
+    public class ServiceConfig
+    {
+        /// <summary>
+        /// 数据服务Host Url
+        /// </summary>
+        public string TelpoDataUrl { get; set; }
+
+        /// <summary>
+        /// Kafka服务地址
+        /// </summary>
+        public string KafkaBootstrapServers { get; set; }
+        public List<string> KafkaTopics { get; set; }
+        public string KafkaGroupId { get; set; }
+        public string KafkaUserName { get; set; }
+        public string KafkaPassword { get; set; }
+        //public string KafkaSslCaLocation { get; set; }
+        /// <summary>
+        /// 默认缓存时间
+        /// </summary>
+        public int CacheDurationSeconds { get; set; }
+        public int CacheDurationSeconds10 { get; set; }
+    }
+}

TelpoPush.WanJiaAn.Worker/Models/Config/WanJiaAnConfig.cs

@@ -0,0 +1,19 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace TelpoPush.WanJiaAn.Worker.Models.Config
+{
+    public class WanJiaAnConfig
+    {
+        public string CoreServiceApiUrl { get; set; }
+        public string GpsWebApiUrl { get; set; }
+
+        public string RzlManufactorId { get; set; }
+        public string RzlPushTemperatureUrl { get; set; }
+        public string RzlPushTranspondUrl { get; set; }
+        public string RzlVoiceCallback { get; set; }
+    }
+}

TelpoPush.WanJiaAn.Worker/Models/Enum/HeadersDto.cs

@@ -0,0 +1,16 @@
+using Newtonsoft.Json;
+namespace TelpoPush.WanJiaAn.Worker.Models.Enum
+{
+    /// <summary>
+    /// 消息数据头
+    /// </summary>
+    public class HeadersDto
+    {
+        [JsonProperty(PropertyName = "DataType")]
+        public int? DataType { get; set; }
+        [JsonProperty(PropertyName = "AlarmType")]
+        public int? AlarmType { get; set; }
+        [JsonProperty(PropertyName = "OperType")]
+        public int? OperType { get; set; }
+    }
+}

TelpoPush.WanJiaAn.Worker/Models/Enum/MqDataTopic.cs

@@ -0,0 +1,19 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace TelpoPush.WanJiaAn.Worker.Models.Enum
+{
+
+    public enum MqDataTopic : int
+    {
+        /// <summary>
+        /// 中高实时心率
+        /// </summary>
+        ZkRealHRMonitorTopic = 1
+
+
+    }
+}

TelpoPush.WanJiaAn.Worker/Models/Enum/MqDataType.cs

@@ -0,0 +1,156 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace TelpoPush.WanJiaAn.Worker.Models.Enum
+{
+    /// <summary>
+    /// 数据类型，标识发布到kafka的消息的数据类型
+    /// </summary>
+    public enum MqDataType : int
+    {
+        /// <summary>
+        /// 报警消息
+        /// </summary>
+        AlarmInfo = 0,
+
+        /// <summary>
+        /// 温度数据信息
+        /// </summary>
+        TemperatureInfo = 1,
+
+        /// <summary>
+        /// 步数信息
+        /// </summary>
+        StepInfo = 2,
+
+        /// <summary>
+        /// 电量信息
+        /// </summary>
+        BatteryLevelInfo = 3,
+
+        /// <summary>
+        /// 设备配置信息
+        /// </summary>
+        DeviceConfigInfo = 4,
+
+        /// <summary>
+        /// 设备通话记录
+        /// </summary>
+        DeviceCallLog = 5,
+
+        /// <summary>
+        /// 设备短信记录
+        /// </summary>
+        DeviceSmsLog = 6,
+
+        /// <summary>
+        /// 位置信息
+        /// </summary>
+        WanJiaAnInfo = 7,
+
+        /// <summary>
+        /// 支付
+        /// </summary>
+        PayInfo = 8,
+
+        /// <summary>
+        /// 设备状态（offline，online）
+        /// </summary>
+        Status = 9,
+
+        /// <summary>
+        /// 设备激活状态（激活1，未激活0）
+        /// </summary>
+        Active = 10,
+
+        /// <summary>
+        /// 指令回调
+        /// </summary>
+        reply = 11,
+
+        /// <summary>
+        /// 天气查询
+        /// </summary>
+        Weather = 12,
+
+        /// <summary>
+        /// 短信阅读事件
+        /// </summary>
+        ReadMsg = 13,
+        /// <summary>
+        /// 学习能力状态上报事件
+        /// </summary>
+        StudyAINotifyStatusUpload = 14,
+        /// <summary>
+        /// 心率
+        /// </summary>
+        HeartRateInfo = 15,
+        /// <summary>
+        /// 血氧
+        /// </summary>
+        Spo2Info = 16,
+        /// <summary>
+        /// 周期性报体温数据。
+        /// </summary>
+        Temperature1Info = 17,
+        /// <summary>
+        /// 周期心率。
+        /// </summary>
+        HeartRate1Info = 18,
+        /// <summary>
+        /// 周期性血氧
+        /// </summary>
+        Spo21Info = 19,
+        /// <summary>
+        /// 溺水状态
+        /// </summary>
+        DrownReportInfo = 20,
+        /// <summary>
+        /// 手表佩戴状态
+        /// </summary>
+        WearStatusInfo = 21,
+        /// <summary>
+        /// 血压
+        /// </summary>
+        BloodPressInfo = 22,
+        /// <summary>
+        /// 周期性血压
+        /// </summary>
+        BloodPress1Info = 23,
+        /// <summary>
+        /// 心理监测
+        /// </summary>
+        PsychInfo = 24,
+
+        /// <summary>
+        /// AI呼叫回调结果
+        /// </summary>
+        AiCallResult = 25,
+
+        /// <summary>
+        /// 越界上报(围栏进出告警)
+        /// </summary>
+        CrossBorder = 26,
+
+        /// <summary>
+        /// 运动数据上报
+        /// </summary>
+        SportResult = 27,
+        /// <summary>
+        /// 运动数据上报
+        /// </summary>
+        BloodSugar = 28,
+
+        /// <summary>
+        /// 中考实时心率数据上报
+        /// </summary>
+        ZkRealHeartRate = 29,
+        /// <summary>
+        /// 绑定业务
+        /// </summary>
+        BindDevice = 100
+    }
+}

TelpoPush.WanJiaAn.Worker/Models/Enum/MqHeader.cs

@@ -0,0 +1,26 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace TelpoPush.WanJiaAn.Worker.Models.Enum
+{
+    public static class MqHeader
+    {
+        /// <summary>
+        /// DataType
+        /// </summary>
+        public const string DataType = "DataType";
+
+        /// <summary>
+        /// OperType
+        /// </summary>
+        public const string OperType = "OperType";
+
+        /// <summary>
+        /// AlarmType
+        /// </summary>
+        public const string AlarmTypes = "AlarmType";
+    }
+}

TelpoPush.WanJiaAn.Worker/Models/MqTemplates/BaseModel.cs

@@ -0,0 +1,22 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using TelpoPush.WanJiaAn.Worker.Models.PushTemplates;
+
+namespace TelpoPush.WanJiaAn.Worker.Models.MqTemplates
+{
+    public class BaseModel
+    {
+        public string msg_id { get; set; }
+        public string device_id { get; set; }
+        public string type { get; set; }
+        public string device_type { get; set; }
+        public string event_time { get; set; }
+        public int app_id { get; set; }
+        public string user_id { get; set; }
+        public object data { get; set; }
+    }
+
+}

TelpoPush.WanJiaAn.Worker/Models/MqTemplates/MqAlarmEventTemplate.cs

@@ -0,0 +1,26 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace TelpoPush.WanJiaAn.Worker.Models.MqTemplates
+{
+    public class MqAlarmEventTemplate
+    {
+        public AlarmEvent alarm_event { get; set; }
+    }
+
+    public class AlarmEvent
+    {
+        public string event_id { get; set; } //事件唯一ID,用于区分不同事件。每次均上报
+        public string event_start { get; set; }//事件开始时间,秒级时间戳。每次均上报
+        public string event_end { get; set; }  //事件结束时间,秒级时间戳。每次均上报,event_type=2时值为””(空串)
+        public int event_type { get; set; } //2:哭声检测;52:遮脸提醒
+        public string image { get; set; } //事件图片URL地址。event_type=3时上报
+        public int report_type { get; set; } //1:事件开始，2:事件结束，3:资源上报(图片或者图片+视频)
+        public string url { get; set; }//视频首片URL地址。event_type=3时上报
+        public string video_start { get; set; } //视频开始时间, 秒级时间戳。event_type=3时上报
+        public string video_end { get; set; }//视频结束时间, 秒级时间戳。event_type=3时上报
+    }
+}

TelpoPush.WanJiaAn.Worker/Models/PushTemplates/PushFencePositionTemplate.cs

@@ -0,0 +1,35 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace TelpoPush.WanJiaAn.Worker.Models.PushTemplates
+{
+    public class PushFenceWanJiaAnTemplate
+    {
+
+        public string messageId { get; set; }
+        public string topic { get; set; }
+        public string time { get; set; }
+        public data data { get; set; }
+    }
+
+    public class data
+    {
+        public string DeviceId { get; set; }
+        public int Radius { get; set; }
+        public string imei { get; set; }
+        public string wifiInfo { get; set; }
+        public int locationType { get; set; }
+        public string address { get; set; }
+        public decimal baiduLatitude { get; set; }
+        public decimal baiduLongitude { get; set; }
+        public decimal gaodeLatitude { get; set; }
+        public decimal gaodeLongitude { get; set; }
+        public decimal originalLatitude { get; set; }
+        public decimal originalLongitude { get; set; }
+        public string LastUpdate { get; set; }
+        public string UtcDate { get; set; }
+    }
+}

TelpoPush.WanJiaAn.Worker/Models/PushTemplates/PushPositionTemplate.cs

@@ -0,0 +1,73 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace TelpoPush.WanJiaAn.Worker.Models.PushTemplates
+{
+    public class PushWanJiaAnTemplate
+    {
+
+        /// <summary>
+        /// 服务跟踪Id
+        /// </summary>
+        public string MessageId { get; set; }
+
+        /// <summary>
+        /// 海拔
+        /// </summary>
+        public double Altitude { get; set; }
+
+        /// <summary>
+        /// 定位类型
+        /// </summary>
+        public int LocationType { get; set; }
+        /// <summary>
+        /// 设备IMEI
+        /// </summary>
+        public string Imei { get; set; }
+
+        /// <summary>
+        /// 时间
+        /// yyyy-MM-dd HH:mm:ss
+        /// </summary>
+        public string Time { get; set; }
+
+        /// <summary>
+        /// 原始纬度
+        /// </summary>
+        public decimal OriginalLatitude { get; set; }
+
+        /// <summary>
+        /// 原始经度
+        /// </summary>
+        public decimal OriginalLongitude { get; set; }
+
+        /// <summary>
+        /// 百度地图纬度
+        /// </summary>
+        public decimal BaiduLatitude { get; set; }
+
+        /// <summary>
+        /// 百度地图经度
+        /// </summary>
+        public decimal BaiduLongitude { get; set; }
+
+        /// <summary>
+        /// 高德地图纬度
+        /// </summary>
+        public decimal GaodeLatitude { get; set; }
+
+        /// <summary>
+        /// 高德地图经度
+        /// </summary>
+        public decimal GaodeLongitude { get; set; }
+
+        /// <summary>
+        /// 地址
+        /// </summary>
+        public string Address { get; set; }
+        public string wifiMacs { get; set; }
+    }
+}

TelpoPush.WanJiaAn.Worker/Program.cs

@@ -0,0 +1,103 @@
+
+using Newtonsoft.Json;
+using Newtonsoft.Json.Serialization;
+using Serilog;
+using TelpoDataService.Util.Clients;
+using TelpoPush.WanJiaAn.Worker;
+using TelpoPush.WanJiaAn.Worker.Common;
+using TelpoPush.WanJiaAn.Worker.Handlers;
+using TelpoPush.WanJiaAn.Worker.Models.Config;
+using TelpoPush.WanJiaAn.Worker.Service.Cache;
+using TelpoPush.WanJiaAn.Worker.Service.Mq;
+
+
+#region 日志
+//using Serilog.Events;
+//using Serilog;
+//using Microsoft.AspNetCore.Builder;
+
+//Log.Logger = new LoggerConfiguration()
+//#if DEBUG
+//    .MinimumLevel.Debug()
+//#else
+// .MinimumLevel.Information()
+//#endif
+//    .MinimumLevel.Override("Microsoft", LogEventLevel.Information)
+//    .MinimumLevel.Override("Microsoft.EntityFrameworkCore", LogEventLevel.Warning)
+//    .Enrich.FromLogContext()
+//    //.Filter.ByExcluding(c => c.Properties.Any(p => p.Value.ToString().Contains("Microsoft")))//过滤
+//    .WriteTo.Async(c => c.File("/var/telpo_pushthird_ssl2/logs/infos/info.log",
+//        restrictedToMinimumLevel: LogEventLevel.Information,
+//        rollingInterval: RollingInterval.Day,//滚动策略(天)
+//                                             //fileSizeLimitBytes: 20971520, //设置单个文件大小为3M 默认1G
+//        rollOnFileSizeLimit: true, //超过文件大小后创建新的
+//        retainedFileCountLimit: 7,//默认31，意思就是只保留最近的31个日志文件"
+//        outputTemplate: "{Timestamp:yyyy-MM-dd HH:mm:ss.fff}[{Level:u3}] [Thread-{ThreadId}] [{SourceContext:l}] {Message:lj}{NewLine}{Exception}"
+//        )
+//    )
+//      .WriteTo.Async(c => c.File("/var/telpo_pushthird_ssl2/logs/errors/errors.log",
+//        restrictedToMinimumLevel: LogEventLevel.Error,
+//        rollingInterval: RollingInterval.Day,
+//        rollOnFileSizeLimit: true,
+//        retainedFileCountLimit: 7,
+//        outputTemplate: "{Timestamp:yyyy-MM-dd HH:mm:ss.fff }[{Level:u3}] [Thread-{ThreadId}] [{SourceContext:l}] {Message:lj}{NewLine}{Exception}"
+//        )
+//    )
+//    .WriteTo.Async(c => c.Console())
+//    .CreateLogger();
+
+//选择配置文件appsetting.json
+var configuration = new ConfigurationBuilder()
+    .SetBasePath(Directory.GetCurrentDirectory())
+    .AddJsonFile("appsettings.json", optional: true, reloadOnChange: true)
+    .Build();
+
+Log.Logger = new LoggerConfiguration()
+    .ReadFrom.Configuration(configuration)
+    .Enrich.WithThreadId()
+    .CreateLogger();
+#endregion
+try
+{
+    Log.Information("Starting up");
+    var builder = Host.CreateApplicationBuilder(args);
+
+    var config = builder.Configuration;
+    builder.Services.Configure<WanJiaAnConfig>(config.GetSection("WanJiaAnConfig"));
+    builder.Services.Configure<ServiceConfig>(config.GetSection("ServiceConfig"));
+    builder.Services.Configure<RedisConfig>(config.GetSection("Redis"));
+
+    JsonSerializerSettings setting = new JsonSerializerSettings();
+    JsonConvert.DefaultSettings = () =>
+    {
+        setting.DateFormatString = "yyyy-MM-dd HH:mm:ss";
+        setting.ContractResolver = new CamelCasePropertyNamesContractResolver();
+        return setting;
+    };
+    builder.Services.AddTelpoDataServices(opt =>
+    {
+        opt.TelpoDataUrl = config["ServiceConfig:TelpoDataUrl"];
+    });
+
+    builder.Services.AddSerilog();
+    builder.Services.AddHttpClient();
+    builder.Services.AddTransient<HttpHelperAsync>();
+    builder.Services.AddSingleton<SqlMapper>();
+    builder.Services.AddSingleton<RedisUtil>();
+    builder.Services.AddSingleton<IKafkaService, KafkaService>();
+    builder.Services.AddSingleton<KafkaSubscribe>();
+    builder.Services.AddSingleton<MessageProducer>();
+    builder.Services.AddSingleton<MqProcessMessage>();
+    builder.Services.AddSingleton<WanJiaAnProcess>();
+    builder.Services.AddHostedService<Worker>();
+    var host = builder.Build();
+    host.Run();
+}
+catch (Exception ex)
+{
+    Log.Fatal(ex, "Application start-up failed");
+}
+finally
+{
+    Log.CloseAndFlush();
+}

TelpoPush.WanJiaAn.Worker/Properties/launchSettings.json

@@ -0,0 +1,12 @@
+{
+  "$schema": "http://json.schemastore.org/launchsettings.json",
+  "profiles": {
+    "TelpoPush.WanJiaAn.Worker": {
+      "commandName": "Project",
+      "dotnetRunMessages": true,
+      "environmentVariables": {
+        "DOTNET_ENVIRONMENT": "Development"
+      }
+    }
+  }
+}

TelpoPush.WanJiaAn.Worker/Service/Cache/MemoryCacheUtil.cs

@@ -0,0 +1,82 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Caching.Memory;
+
+namespace TelpoPush.WanJiaAn.Worker.Service.Cache
+{
+    internal class MemoryCacheUtil
+    {
+            static MemoryCache cache = new MemoryCache(new MemoryCacheOptions());
+            /// <summary>
+            /// 创建缓存项的文件
+            /// </summary>
+            /// <param name="key">缓存Key</param>
+            /// <param name="obj">object对象</param>
+            public static void Set(string key, object value)
+            {
+                if (key != null)
+                {
+                    cache.Set(key, value);
+                }
+            }
+            /// <summary>
+            /// 创建缓存项过期
+            /// </summary>
+            /// <param name="key">缓存Key</param>
+            /// <param name="obj">object对象</param>
+            /// <param name="expires">过期时间(秒)</param>
+            public static void Set(string key, object value, int expires)
+            {
+                if (key != null)
+                {
+                    cache.Set(key, value, new MemoryCacheEntryOptions()
+                        //设置缓存时间，如果被访问重置缓存时间。设置相对过期时间x秒
+                        .SetSlidingExpiration(TimeSpan.FromSeconds(expires)));
+                }
+            }
+
+            /// <summary>
+            /// 获取缓存对象
+            /// </summary>
+            /// <param name="key">缓存Key</param>
+            /// <returns>object对象</returns>
+            public static object Get(string key)
+            {
+                object val = null;
+                if (key != null && cache.TryGetValue(key, out val))
+                {
+
+                    return val;
+                }
+                else
+                {
+                    return default(object);
+                }
+            }
+
+            /// <summary>
+            /// 获取缓存对象
+            /// </summary>
+            /// <typeparam name="T">T对象</typeparam>
+            /// <param name="key">缓存Key</param>
+            /// <returns></returns>
+            public static T Get<T>(string key)
+            {
+                object obj = Get(key);
+                return obj == null ? default(T) : (T)obj;
+            }
+
+
+            /// <summary>
+            /// 移除缓存项的文件
+            /// </summary>
+            /// <param name="key">缓存Key</param>
+            public static void Remove(string key)
+            {
+                cache.Remove(key);
+            }
+        }
+    }

TelpoPush.WanJiaAn.Worker/Service/Cache/RedisUtil.cs

@@ -0,0 +1,208 @@
+using Microsoft.Extensions.Options;
+using TelpoPush.WanJiaAn.Worker.Models.CacheTemplates;
+using TelpoPush.WanJiaAn.Worker.Models.Config;
+
+namespace TelpoPush.WanJiaAn.Worker.Service.Cache
+{
+    public class RedisUtil
+    {
+        private const string CACHE_HASH_KEY_TELPO_GPSDEVICE = "TELPO#GPSDEVICE_HASH";
+        private const string CACHE_HASH_KEY_TELPO_GPSDEVICE_PERSON = "TELPO#GPSDEVICE_PERSON_HASH";
+        private const string CACHE_HASH_KEY_TELPO_MANUFACTOR_CONFIG = "TELPO#MANUFACTOR_CONFG_HASH";
+        private const string CACHE_HASH_KEY_TELPO_GPSDEVICE_PUSHSITTTIGS = "TELPO#GPSDEVICE_PUSH_SITTINGS_HASH";
+
+        private readonly ILogger<RedisUtil> _logger;
+        private readonly ServiceConfig _configService;
+        private readonly SqlMapper _sqlMapper;
+        public RedisUtil(ILogger<RedisUtil> logger,IOptions<RedisConfig> optConfigRedis, IOptions<ServiceConfig> configService, SqlMapper sqlMapper)
+        {
+            _configService = configService.Value;
+            _logger = logger;
+            optConfigRedis.Value.Prefix = "";
+            optConfigRedis.Value.DefaultDatabase = 7;
+            var csredis = new CSRedis.CSRedisClient(optConfigRedis.Value.ToString());
+            RedisHelper.Initialization(csredis);
+            _sqlMapper = sqlMapper;
+        }
+
+        public async Task<DeviceInfoModel> GetGpsDevice(string imei)
+        {
+            if (string.IsNullOrWhiteSpace(imei)) return null;
+            string keyCache = $"{imei}_GpsDevice";
+            try
+            {
+                var objCache = MemoryCacheUtil.Get<DeviceInfoModel>(keyCache);
+                if (objCache == null)
+                {
+                    var obj = await RedisHelper.HGetAsync<DeviceInfoModel>(CACHE_HASH_KEY_TELPO_GPSDEVICE, imei);
+                    if (obj == null)
+                    {
+                        var deviceInfo = _sqlMapper.DeviceInfo(imei);
+                        if (deviceInfo != null)
+                        {
+                            RedisHelper.HSetAsync(CACHE_HASH_KEY_TELPO_GPSDEVICE, imei, deviceInfo);
+                            MemoryCacheUtil.Set(keyCache, deviceInfo, _configService.CacheDurationSeconds10);
+                            return deviceInfo;
+                        }
+                    }
+                    else
+                    {
+                        MemoryCacheUtil.Set(keyCache, obj, _configService.CacheDurationSeconds10);
+                        return obj;
+                    }
+                }
+                return objCache;
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError($"GetGpsDevice,key={imei},缓存异常，重新获取数据,{ex.Message}|{ex.Source}|{ex.StackTrace}");
+                var deviceInfo = _sqlMapper.DeviceInfo(imei);
+                if (deviceInfo != null)
+                {
+                    RedisHelper.HSetAsync(CACHE_HASH_KEY_TELPO_GPSDEVICE, imei, deviceInfo);
+                    MemoryCacheUtil.Set(keyCache, deviceInfo, _configService.CacheDurationSeconds10);
+                    return deviceInfo;
+                }
+            }
+            return null;
+        }
+
+        public async Task<PersonInfoModel> SetPersonInfoHash(string imei)
+        {
+            if (string.IsNullOrWhiteSpace(imei)) return null;
+            string keyCache = $"{imei}_PersonInfoHash";
+            var personInfo = _sqlMapper.PersonInfo(imei);
+            PersonInfoModel model = new PersonInfoModel()
+            {
+                person = personInfo,
+                time = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")
+            };
+            if (personInfo != null)
+            {
+                await RedisHelper.HSetAsync(CACHE_HASH_KEY_TELPO_GPSDEVICE_PERSON, imei, model);
+                MemoryCacheUtil.Set(keyCache, model, 60);//1分钟
+            }
+            else
+            {
+                await RedisHelper.HDelAsync(CACHE_HASH_KEY_TELPO_GPSDEVICE_PERSON, imei);
+                MemoryCacheUtil.Remove(keyCache);
+            }
+            return model;
+        }
+
+        public async Task<string> GetHealthyDeviceKey(string imei)
+        {
+            if (string.IsNullOrWhiteSpace(imei)) return null;
+            string HealthyDeviceKey = $"Telpol:HealthyDeviceKey:{imei}";
+            string keyCache = $"{imei}_HealthyDeviceKey";
+            try
+            {
+                var objCache = MemoryCacheUtil.Get<string>(keyCache);
+                if (objCache == null)
+                {
+                    string deviceKey = await RedisHelper.HGetAsync<string>(HealthyDeviceKey, "data");
+                    if (!string.IsNullOrEmpty(deviceKey))
+                    {
+                        MemoryCacheUtil.Set(keyCache, deviceKey, _configService.CacheDurationSeconds10);
+                        return deviceKey;
+                    }
+                }
+                return objCache;
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError($"GetHealthyDeviceKey,key={imei},缓存异常，重新获取数据,{ex.Message}|{ex.Source}|{ex.StackTrace}");
+                string deviceKey = await RedisHelper.HGetAsync<string>(HealthyDeviceKey, "data");
+                if (!string.IsNullOrEmpty(deviceKey))
+                {
+                    MemoryCacheUtil.Set(keyCache, deviceKey, _configService.CacheDurationSeconds10);
+                    return deviceKey;
+                }
+                else
+                    return "";
+            }
+        }
+
+        public async Task<SettingInfosItem> GetManufactorPushSettingHash(string imei, string manufactorId, int dataType)
+        {
+            if (string.IsNullOrWhiteSpace(manufactorId)) return null;
+            string keyCache = $"{manufactorId}_{dataType}_ManufactorPushSettingHash";
+            SettingInfosItem settingInfos = null;
+            try
+            {
+                var objCache = MemoryCacheUtil.Get<SettingInfosItem>(keyCache);
+                if (objCache == null)
+                {
+                    var obj = await RedisHelper.HGetAsync<ManufactorPushSettingInfoModel>(CACHE_HASH_KEY_TELPO_GPSDEVICE_PUSHSITTTIGS, manufactorId);
+                    if (obj != null)
+                    {
+                        if (obj.pushSettings.Any())
+                        {
+                            DateTime dt = DateTime.Now;
+                            var settingsItem = obj.pushSettings.FirstOrDefault<PushSettingsItem>(x => x.dataType == dataType);
+                            if (settingsItem != null)
+                            {
+                                if (settingsItem.settingInfos.Any())
+                                {
+                                    foreach (var item in settingsItem.settingInfos)
+                                    {
+                                        DateTime startTime = DateTime.Parse($"{dt.Year}-{dt.Month}-{dt.Day} {item.pushStartTime}");
+                                        DateTime endTime = DateTime.Parse($"{dt.Year}-{dt.Month}-{dt.Day} {item.pushEndTime}");
+
+                                        if (item.pushFlag && (dt > startTime && dt < endTime))
+                                        {
+                                            settingInfos = item;
+                                            break;
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                        if (settingInfos != null)
+                            MemoryCacheUtil.Set(keyCache, settingInfos, _configService.CacheDurationSeconds10);
+                        else
+                            MemoryCacheUtil.Remove(keyCache);
+                    }
+                }
+                else
+                    settingInfos = objCache;
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError($"GetManufactorPushSettingHash(imei={imei},dataType={dataType}),key={manufactorId},缓存异常，重新获取数据,{ex.Message}|{ex.Source}|{ex.StackTrace}");
+                var obj = await RedisHelper.HGetAsync<ManufactorPushSettingInfoModel>(CACHE_HASH_KEY_TELPO_GPSDEVICE_PUSHSITTTIGS, manufactorId);
+                if (obj != null)
+                {
+                    if (obj.pushSettings.Any())
+                    {
+                        DateTime dt = DateTime.Now;
+                        var settingsItem = obj.pushSettings.FirstOrDefault<PushSettingsItem>(x => x.dataType == dataType);
+                        if (settingsItem != null)
+                        {
+                            if (settingsItem.settingInfos.Any())
+                            {
+                                foreach (var item in settingsItem.settingInfos)
+                                {
+                                    DateTime startTime = DateTime.Parse($"{dt.Year}-{dt.Month}-{dt.Day} {item.pushStartTime}");
+                                    DateTime endTime = DateTime.Parse($"{dt.Year}-{dt.Month}-{dt.Day} {item.pushEndTime}");
+
+                                    if (item.pushFlag && (dt > startTime && dt < endTime))
+                                    {
+                                        settingInfos = item;
+                                        break;
+                                    }
+                                }
+                            }
+                        }
+                    }
+                    if (settingInfos != null)
+                        MemoryCacheUtil.Set(keyCache, settingInfos, _configService.CacheDurationSeconds10);
+                    else
+                        MemoryCacheUtil.Remove(keyCache);
+                }
+            }
+            return settingInfos;
+        }
+
+    }
+}

TelpoPush.WanJiaAn.Worker/Service/Cache/SqlMapper.cs

@@ -0,0 +1,39 @@
+using Dapper;
+using MySql.Data.MySqlClient;
+using System.Data;
+using TelpoPush.WanJiaAn.Worker.Models.CacheTemplates;
+
+namespace TelpoPush.WanJiaAn.Worker.Service.Cache
+{
+    public class SqlMapper
+    {
+        private readonly IConfiguration _config;
+        private static string gps_conn = "";
+        private static string telcommon_conn = "";
+        private static string healthy_conn = "";
+        public SqlMapper(IConfiguration config)
+        {
+            _config = config;
+            gps_conn = _config["ConnectionStrings:DB_Connection_String"].ToString();
+            telcommon_conn = _config["ConnectionStrings:Telpo_common_ConnString"].ToString();
+            healthy_conn = _config["ConnectionStrings:Telpo_Healthy_ConnString"].ToString();
+        }
+
+        public DeviceInfoModel DeviceInfo(string imei)
+        {
+            using (IDbConnection connection = new MySqlConnection(gps_conn))
+            {
+                var sql = @"SELECT d.device_id deviceId, p.nick_name deviceName,p.api_uid apiUid,d.serialno imei, d.org_uid orgId, d.active_status activeStatus,active_time activeTime FROM gps_device as d LEFT JOIN `gps_person` as p on p.device_id=d.device_id WHERE d.serialno=@imei";
+                return connection.QueryFirstOrDefault<DeviceInfoModel>(sql, new { imei });
+            }
+        }
+        public PersonModel PersonInfo(string imei)
+        {
+            using (IDbConnection connection = new MySqlConnection(gps_conn))
+            {
+                var sql = @"SELECT person_id personId, serialno, person_name personName, device_id deviceId, nick_name nickName, gender,  height, weight, born_date bornDate, school, grade, class_name className, image_path imagePath,image_path_small imagePathSmall, age, create_time createTime,  remarks, ishypertension, emotion,profession,regularity,chronic_disease chronicDisease,api_uid apiUid,api_remark apiRemark FROM gps_person WHERE serialno=@imei";
+                return connection.QueryFirstOrDefault<PersonModel>(sql, new { imei });
+            }
+        }
+    }
+}

TelpoPush.WanJiaAn.Worker/Service/Mq/IKafkaService.cs

@@ -0,0 +1,9 @@
+using Confluent.Kafka;
+
+namespace TelpoPush.WanJiaAn.Worker.Service.Mq
+{
+    public interface IKafkaService
+    {
+        Task SubscribeAsync(Action<string, string, Headers> messageFunc, CancellationToken cancellationToken);
+    }
+}

TelpoPush.WanJiaAn.Worker/Service/Mq/KafkaHeader.cs

@@ -0,0 +1,15 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace TelpoPush.WanJiaAn.Worker.Service.Mq
+{
+    public class KafkaHeader
+    {
+        public static readonly string DataType = "DataType";
+        public static readonly string AlarmType = "AlarmType";
+        public static readonly string OperType = "OperType";
+    }
+}

TelpoPush.WanJiaAn.Worker/Service/Mq/KafkaService.cs

@@ -0,0 +1,159 @@
+using Confluent.Kafka;
+using Microsoft.Extensions.Options;
+using Newtonsoft.Json;
+using TelpoPush.WanJiaAn.Worker.Models.Config;
+
+namespace TelpoPush.WanJiaAn.Worker.Service.Mq
+{
+    public class KafkaService : IKafkaService
+    {
+        private readonly ConsumerConfig _consumerConfig;
+        private readonly IHostEnvironment env;
+        private readonly ILogger<KafkaService> logger;
+        private readonly ServiceConfig _configService;
+        public KafkaService(ILogger<KafkaService> _logger, IHostEnvironment _env, IOptions<ServiceConfig> optConfigService)
+        {
+            _configService = optConfigService.Value;
+            env = _env;
+            logger = _logger;
+
+            //_consumerConfig = new ConsumerConfig
+            //{
+            //    BootstrapServers = _configService.KafkaBootstrapServers,
+            //    GroupId = _configService.KafkaGroupId,
+            //    EnableAutoCommit = false, // 禁止AutoCommit
+            //    Acks = Acks.Leader, // 假设只需要Leader响应即可
+            //    AutoOffsetReset = AutoOffsetReset.Earliest,// 从最早的开始消费起
+            //    CancellationDelayMaxMs = 1//set CancellationDelayMaxMs
+            //};
+
+            _consumerConfig = new ConsumerConfig
+            {
+                BootstrapServers = _configService.KafkaBootstrapServers,
+                SecurityProtocol = SecurityProtocol.Plaintext,
+                SaslMechanism = SaslMechanism.Plain,
+                GroupId = _configService.KafkaGroupId,
+                SaslUsername = _configService.KafkaUserName,
+                SaslPassword = _configService.KafkaPassword,
+                EnableAutoCommit = false, // 禁止AutoCommit
+                AutoOffsetReset = AutoOffsetReset.Earliest, // 从最早的开始消费起
+                CancellationDelayMaxMs = 1
+            };
+
+
+        }
+        public async Task SubscribeAsync(Action<string, string, Headers> messageFunc, CancellationToken cancellationToken)
+        {
+            List<string> topics = _configService.KafkaTopics;
+            using (var consumer = new ConsumerBuilder<Ignore, string>(_consumerConfig)
+                                 .SetErrorHandler((_, e) =>
+                                 {
+                                     logger.LogError($"Error: {e.Reason}");
+                                 })
+                                 .SetStatisticsHandler((_, json) =>
+                                 {
+                                     logger.LogInformation($" - {DateTime.Now:yyyy-MM-dd HH:mm:ss} > 消息监听中..");
+                                 })
+                                 .SetPartitionsAssignedHandler((c, partitions) =>
+                                 {
+                                     string partitionsStr = string.Join(", ", partitions);
+                                     logger.LogInformation($" - 分配的 kafka 分区: {partitionsStr}");
+                                 })
+                                 .SetPartitionsRevokedHandler((c, partitions) =>
+                                 {
+                                     string partitionsStr = string.Join(", ", partitions);
+                                     logger.LogInformation($" - 回收了 kafka 分区: {partitionsStr}");
+                                 })
+                                 .Build())
+            {
+
+                consumer.Subscribe(topics);
+                try
+                {
+                    while (true)
+                    {
+                        try
+                        {
+                            var consumeResult = consumer.Consume(cancellationToken);
+                            string topic = consumeResult.Topic;
+                            string messageResult = consumeResult.Message.Value;
+                            Headers headers = consumeResult.Message.Headers;
+                            bool isPartitionEOF = consumeResult.IsPartitionEOF;
+                            var partition = consumeResult.Partition;
+
+                            int DataType = -1, AlarmType = -1, OperType = -1;
+                            foreach (var item in headers)
+                            {
+                                if (item.Key == KafkaHeader.DataType)
+                                    DataType = BitConverter.ToInt32(item.GetValueBytes(), 0);
+                                else if (item.Key == KafkaHeader.AlarmType)
+                                    AlarmType = BitConverter.ToInt32(item.GetValueBytes(), 0);
+                                else if (item.Key == KafkaHeader.OperType)
+                                    OperType = BitConverter.ToInt32(item.GetValueBytes(), 0);
+                            }
+                            var Headers = new { DataType, AlarmType, OperType };
+                            logger.LogInformation($"Consumed  topic '{topic}' , message '{messageResult}' ,  headers '{JsonConvert.SerializeObject(Headers)}', at '{consumeResult?.TopicPartitionOffset}'.");
+                            if (isPartitionEOF)
+                            {
+                                logger.LogInformation($" - {DateTime.Now:yyyy-MM-dd HH:mm:ss} 已经到底了：{topic}, partition {partition}, offset {consumeResult?.Offset}.");
+                                continue;
+                            }
+                            if (!string.IsNullOrEmpty(messageResult))
+                            {
+                                messageFunc(topic, messageResult, headers);
+                                try
+                                {
+                                    consumer.Commit(consumeResult);
+                                }
+                                catch (KafkaException e)
+                                {
+                                    logger.LogError($" - {e.Message}.");
+                                }
+                            }
+                            #region 注释
+                            //string messageResult = null;
+                            //Headers headers = null;
+                            //try
+                            //{
+                            //    messageResult = consumeResult.Message.Value;
+                            //    headers = consumeResult.Message.Headers;
+                            //}
+                            //catch (Exception ex)
+                            //{
+                            //    var errorMessage = $" - {DateTime.Now:yyyy-MM-dd HH:mm:ss}【Exception 消息反序列化失败，Value：{consumeResult.Message.Value}】 ：{ex.StackTrace?.ToString()}";
+                            //    Console.WriteLine(errorMessage);
+                            //    logger.LogError(errorMessage);
+                            //    messageResult = null;
+                            //}
+                            //if (!string.IsNullOrEmpty(messageResult)/* && consumeResult.Offset % commitPeriod == 0*/)
+                            //{
+                            //    string topic = consumeResult.Topic;
+                            //    messageFunc(topic, messageResult, headers);
+
+                            //    //try
+                            //    //{
+                            //    //    consumer.Commit(consumeResult);
+                            //    //}
+                            //    //catch (KafkaException e)
+                            //    //{
+                            //    //    Console.WriteLine(e.Message);
+                            //    //}
+                            //}
+                            #endregion
+                        }
+                        catch (ConsumeException e)
+                        {
+                            logger.LogError($"Consume error: {e.Error.Reason}");
+                        }
+                    }
+                }
+                catch (OperationCanceledException)
+                {
+                    logger.LogError("Closing consumer.");
+                    consumer.Close();
+                }
+            }
+            await Task.CompletedTask;
+        }
+    }
+}

TelpoPush.WanJiaAn.Worker/Service/Mq/MessageProducer.cs

@@ -0,0 +1,77 @@
+using Confluent.Kafka;
+using Microsoft.Extensions.Options;
+using Newtonsoft.Json;
+using TelpoPush.WanJiaAn.Worker.Models.Config;
+
+namespace TelpoPush.WanJiaAn.Worker.Service.Mq
+{
+    /// <summary>
+    /// 消息生产者
+    /// </summary>
+    public class MessageProducer
+    {
+        private readonly ILogger<MessageProducer> _logger;
+        private readonly ServiceConfig _configService;
+        private readonly IProducer<Null, string> _producer;
+
+        public MessageProducer(ILogger<MessageProducer> logger, IOptions<ServiceConfig> optConfigService)
+        {
+            _logger = logger;
+            _configService = optConfigService.Value;
+
+            var config = new ProducerConfig
+            {
+                BootstrapServers = _configService.KafkaBootstrapServers,
+                EnableIdempotence = true,
+                Acks = Acks.All,
+                //LingerMs=5000,
+                //BatchNumMessages =1000,
+                //BatchSize=32768,
+                //CompressionType= CompressionType.Lz4,
+                MessageSendMaxRetries = 3
+            };
+            _producer = new ProducerBuilder<Null, string>(config).Build();
+        }
+
+        public Headers CreateHeader(Dictionary<string, int> pair = null)
+        {
+            if (pair == null)
+            {
+                return null;
+            }
+            else
+            {
+                Headers headers = new Headers();
+
+                foreach (var item in pair)
+                {
+                    headers.Add(item.Key, BitConverter.GetBytes(item.Value));
+                }
+                return headers;
+            }
+        }
+
+        public async Task ProduceAsync(List<TopicModel> topic, object message)
+        {
+            try
+            {
+                foreach (var item in topic)
+                {
+                    //  producer = new ProducerBuilder<Null, string>(config).Build();
+                    await _producer.ProduceAsync(item.Topic, new Message<Null, string>
+                    {
+                        Headers = item.Headers,
+                        Value = JsonConvert.SerializeObject(message)
+                    });
+                }
+            }
+            catch (ProduceException<Null, string> e)
+            {
+                _logger.LogError($"推送到kafka失败，topic: {topic}，\n message:{JsonConvert.SerializeObject(message)}: \n{e.Error.Reason}");
+            }
+        }
+
+
+
+    }
+}

TelpoPush.WanJiaAn.Worker/Service/Mq/MqProcessMessage.cs

@@ -0,0 +1,70 @@
+using Newtonsoft.Json;
+using TelpoPush.WanJiaAn.Worker.Models.Enum;
+using TelpoPush.WanJiaAn.Worker.Models.MqTemplates;
+
+namespace TelpoPush.WanJiaAn.Worker.Service.Mq
+{
+    public class MqProcessMessage
+    {
+        private readonly ILogger<MqProcessMessage> _logger;
+        private readonly MessageProducer _messageProducer;
+
+        public MqProcessMessage(ILogger<MqProcessMessage> logger, MessageProducer producer)
+        {
+            _logger = logger;
+            _messageProducer = producer;
+        }
+
+        public async Task ProcessProperty(string imei, BaseModel model, HeadersDto headers)
+        {
+            List<TopicModel> ls = new List<TopicModel>();
+
+            ls.Add(new TopicModel()
+            {
+                Topic = "topic.push.property",
+                Headers = _messageProducer.CreateHeader(new Dictionary<string, int>
+                {
+                    {MqHeader.DataType,headers.DataType.Value }
+                })
+            });
+            await _messageProducer.ProduceAsync(ls, model);
+            _logger.LogInformation($"【成功】Third推送(topic.property):IMEI<{imei}>,pushData:{JsonConvert.SerializeObject(model)}");
+        }
+
+        public async Task ProcessDataPushServer(string imei, object model, Dictionary<string, int> headers, string tag)
+        {
+            List<TopicModel> ls = new List<TopicModel>();
+            ls.Add(new TopicModel()
+            {
+                Topic = "topic.push",
+                Headers = _messageProducer.CreateHeader(headers)
+            });
+            await _messageProducer.ProduceAsync(ls, model);
+            _logger.LogInformation($"【{tag}-成功】Third推送(topic.push):IMEI<{imei}>,Header<{JsonConvert.SerializeObject(headers)}>,pushData:{JsonConvert.SerializeObject(model)}");
+        }
+
+        public async Task ProcessFenceServer(string imei, object model, Dictionary<string, int> headers, string tag)
+        {
+            List<TopicModel> ls = new List<TopicModel>();
+            ls.Add(new TopicModel()
+            {
+                Topic = "topic.push.fence",
+                Headers = _messageProducer.CreateHeader(headers)
+            });
+            await _messageProducer.ProduceAsync(ls, model);
+            _logger.LogInformation($"【{tag}-成功】Third推送(topic.push.fence):IMEI<{imei}>,Header<{JsonConvert.SerializeObject(headers)}>,pushData:{JsonConvert.SerializeObject(model)}");
+        }
+
+        public async Task ProcessThirdhServer(string imei, object model, Dictionary<string, int> headers, string tag)
+        {
+            List<TopicModel> ls = new List<TopicModel>();
+            ls.Add(new TopicModel()
+            {
+                Topic = "topic.push.third",
+                Headers = _messageProducer.CreateHeader(headers)
+            });
+            await _messageProducer.ProduceAsync(ls, model);
+            _logger.LogInformation($"【{tag}-成功】Third推送(topic.push.third):IMEI<{imei}>,Header<{JsonConvert.SerializeObject(headers)}>,pushData:{JsonConvert.SerializeObject(model)}");
+        }
+    }
+}

TelpoPush.WanJiaAn.Worker/Service/Mq/TopicModel.cs

@@ -0,0 +1,15 @@
+using Confluent.Kafka;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace TelpoPush.WanJiaAn.Worker.Service.Mq
+{
+    public class TopicModel
+    {
+        public string Topic { get; set; }
+        public Headers Headers { get; set; }
+    }
+}

TelpoPush.WanJiaAn.Worker/TelpoPush.WanJiaAn.Worker.csproj

@@ -0,0 +1,29 @@
+<Project Sdk="Microsoft.NET.Sdk.Worker">
+
+  <PropertyGroup>
+    <TargetFramework>net8.0</TargetFramework>
+    <Nullable>enable</Nullable>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <UserSecretsId>dotnet-TelpoPush.WanJiaAn.Worker-c50cea5b-2ab3-4824-b44f-a1b2edae61a3</UserSecretsId>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Confluent.Kafka" Version="2.4.0" />
+    <PackageReference Include="Microsoft.Extensions.Hosting" Version="8.0.0" />
+    <PackageReference Include="Microsoft.VisualStudio.Azure.Containers.Tools.Targets" Version="1.19.6" />
+    <PackageReference Include="Serilog.AspNetCore" Version="8.0.1" />
+    <PackageReference Include="Serilog.Enrichers.Thread" Version="3.1.0" />
+    <PackageReference Include="Serilog.Sinks.Async" Version="1.5.0" />
+    <PackageReference Include="Serilog.Sinks.Exceptionless" Version="4.0.0" />
+    <PackageReference Include="CSRedisCore" Version="3.8.803" />
+    <PackageReference Include="Dapper" Version="2.1.35" />
+    <PackageReference Include="Microsoft.Extensions.Caching.Memory" Version="8.0.0" />
+    <PackageReference Include="Microsoft.Extensions.Logging" Version="8.0.0" />
+    <PackageReference Include="MySql.Data" Version="8.4.0" />
+    <PackageReference Include="TelpoDataService.Util" Version="1.6.9.28-beta1" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <Folder Include="Service\Biz\" />
+  </ItemGroup>
+</Project>

TelpoPush.WanJiaAn.Worker/Worker.cs

@@ -0,0 +1,25 @@
+using TelpoPush.WanJiaAn.Worker.Handlers;
+
+namespace TelpoPush.WanJiaAn.Worker
+{
+    public class Worker : BackgroundService
+    {
+        private readonly ILogger<Worker> _logger;
+        KafkaSubscribe _kafkaSubscribe;
+
+        public Worker(ILogger<Worker> logger, KafkaSubscribe kafkaSubscribe)
+        {
+            _logger = logger;
+            _kafkaSubscribe = kafkaSubscribe;
+        }
+
+        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+        {
+            while (!stoppingToken.IsCancellationRequested)
+            {
+                await _kafkaSubscribe.SubscribeAsync();
+                await Task.Delay(1000, stoppingToken);
+            }
+        }
+    }
+}

TelpoPush.WanJiaAn.Worker/appsettings.Development.json

@@ -0,0 +1,40 @@
+{
+  "ConnectionStrings": {
+    "DB_Connection_String": "server=139.224.254.18;port=3305;database=gps_card;uid=root;pwd=telpo#1234;CharSet=utf8;MinimumPoolSize=10;MaximumPoolSize=1000;SslMode=none",
+    "Telpo_common_ConnString": "server=139.224.254.18;port=3305;database=telpo_common;uid=root;pwd=telpo#1234;CharSet=utf8;MinimumPoolSize=10;MaximumPoolSize=1000;SslMode=none",
+    "Telpo_Healthy_ConnString": "server=139.224.254.18;port=3305;database=telpo_healthy;uid=root;pwd=telpo#1234;CharSet=utf8;MinimumPoolSize=10;MaximumPoolSize=1000;SslMode=none"
+
+    //"DB_Connection_String": "server=rm-uf6j529mu0v6g0btpco.mysql.rds.aliyuncs.com;port=3305;database=gps_card;uid=root;pwd=telpo#1234;CharSet=utf8;MinimumPoolSize=10;MaximumPoolSize=1000;SslMode=none",
+    //"Telpo_common_ConnString": "server=rm-uf6j529mu0v6g0btpco.mysql.rds.aliyuncs.com;port=3305;database=telpo_common;uid=linwl;pwd=linwl#1234;CharSet=utf8;MinimumPoolSize=10;MaximumPoolSize=1000;SslMode=none"
+  },
+
+  "WanJiaAnConfig": {
+    "CoreServiceApiUrl": "http://localhost:8094/api/v1",
+    "GpsWebApiUrl": "http://id.ssjlai.com/webapi",
+    //-------融智联
+    "RzlManufactorId": "2cae99d6-0475-42fe-9fa7-ca27e03077de",
+    "RzlPushTemperatureUrl": "http://121.40.140.74:8071/api/CollectData/TemperatureTranspond",
+    "RzlPushTranspondUrl": "http://121.40.140.74:8071/api/CollectData/ThirdPushTranspond",
+    "RzlVoiceCallback": "http://121.40.140.74/GspCardStand/api/Voice/VoiceCallback"
+  },
+  "ServiceConfig": {
+    "TelpoDataUrl": "http://id.ssjlai.com/data",
+    "KafkaBootstrapServers": "47.116.67.214:9092", // "172.19.42.53:9092",
+    "KafkaTopics": [ "topic.push.WanJiaAn" ],
+    "KafkaGroupId": "telpo.WanJiaAn",
+    "CacheDurationSeconds": 1200, //20分钟
+    "CacheDurationSeconds10": 600 //10分钟
+  },
+  "Redis": {
+    //"Server": "139.224.254.18:8090",
+    "Server": "47.116.142.20:8090",
+    "Password": "telpo#1234",
+    "DefaultDatabase": 7,
+    "Poolsize": 150,
+    "Preheat": 50,
+    "Prefix": "_Third_",
+    "ConnectTimeout": 5000,
+    "IdleTimeout": 20000
+  },
+  "Environment": "dev"
+}

TelpoPush.WanJiaAn.Worker/appsettings.json

@@ -0,0 +1,70 @@
+{
+  "Logging": {
+    "LogLevel": {
+      "Default": "Information",
+      "Microsoft": "Warning",
+      "Microsoft.Hosting.Lifetime": "Information",
+      "System.Net.Http.HttpClient": "Warning"
+    }
+  },
+  "Serilog": {
+    "Using": [ "Serilog.Sinks.File", "Serilog.Sinks.Async", "Serilog.Sinks.Console", "Serilog.Sinks.Exceptionless" ],
+    "MinimumLevel": {
+      "Default": "Verbose",
+      "Override": {
+        "Microsoft": "Warning",
+        "Microsoft.Hosting.Lifetime": "Information",
+        "System.Net.Http.HttpClient": "Warning"
+      }
+    },
+    "WriteTo:Information": {
+      "Name": "Async",
+      "Args": {
+        "Configure": [
+          {
+            "Name": "File",
+            "Args": {
+              "RestrictedToMinimumLevel": "Information",
+              "RollingInterval": "Day",
+              "RollOnFileSizeLimit": "true",
+              "OutputTemplate": "{Timestamp:yyyy-MM-dd HH:mm:ss }[{Level:u3}] [Thread-{ThreadId}] [{SourceContext:l}] {Message:lj}{NewLine}{Exception}",
+              "Path": "/var/telpo_pushwanjiaan/logs/infos/info.log",
+              "RetainedFileCountLimit": 7 // "--设置日志文件个数最大值，默认31，意思就是只保留最近的31个日志文件", "等于null时永远保留文件": null
+              //              "FileSizeLimitBytes": 20971520, //设置单个文件大小为3M 默认1G
+              //              "RollOnFileSizeLimit": true //超过文件大小后创建新的
+
+            }
+          }
+        ]
+      }
+    },
+    "WriteTo:Error": {
+      "Name": "Async",
+      "Args": {
+        "Configure": [
+          {
+            "Name": "File",
+            "Args": {
+              "RestrictedToMinimumLevel": "Error",
+              "RollingInterval": "Day",
+              "RollOnFileSizeLimit": "true",
+              "OutputTemplate": "{Timestamp:yyyy-MM-dd HH:mm:ss }[{Level:u3}] [Thread-{ThreadId}][{SourceContext:l}] {Message:lj}{NewLine}{Exception}",
+              "Path": "/var/telpo_pushwanjiaan/logs/errors/error.log",
+              "RetainedFileCountLimit": 7 // "--设置日志文件个数最大值，默认31，意思就是只保留最近的31个日志文件", "等于null时永远保留文件": null
+              //              "FileSizeLimitBytes": 20971520, //设置单个文件大小为3M 默认1G
+              //              "RollOnFileSizeLimit": true //超过文件大小后创建新的
+            }
+          }
+        ]
+      }
+    },
+    "WriteTo:Console": {
+      "Name": "Console",
+      "Args": {
+        "restrictedToMinimumLevel": "Verbose",
+        "outputTemplate": "{Timestamp:yyyy-MM-dd HH:mm:ss }[{Level:u3}] [Thread-{ThreadId}] [{SourceContext:l}] {Message:lj}{NewLine}{Exception}",
+        "theme": "Serilog.Sinks.SystemConsole.Themes.AnsiConsoleTheme::Code, Serilog.Sinks.Console"
+      }
+    }
+  }
+}

TelpoPush.WanJiaAn.Worker/appsettings.production.json

@@ -0,0 +1,36 @@
+{
+  "ConnectionStrings": {
+    "DB_Connection_String": "server=rm-uf6j529mu0v6g0btp.mysql.rds.aliyuncs.com;port=3305;database=gps_card;uid=root;pwd=telpo#1234;CharSet=utf8;MinimumPoolSize=10;MaximumPoolSize=1000;SslMode=none",
+    "Telpo_common_ConnString": "server=rm-uf6j529mu0v6g0btp.mysql.rds.aliyuncs.com;port=3305;database=telpo_common;uid=linwl;pwd=linwl#1234;CharSet=utf8;MinimumPoolSize=10;MaximumPoolSize=1000;SslMode=none",
+    "Telpo_Healthy_ConnString": "server=rm-uf6j529mu0v6g0btp.mysql.rds.aliyuncs.com;port=3305;database=telpo_healthy;uid=root;pwd=telpo#1234;CharSet=utf8;MinimumPoolSize=10;MaximumPoolSize=1000;SslMode=none"
+  },
+
+  "WanJiaAnConfig": {
+    "CoreServiceApiUrl": "http://ai.ssjlai.com/gateway/core/api/v1",
+    "GpsWebApiUrl": "https://ai.ssjlai.com/webapi",
+    //-------融智联
+    "RzlManufactorId": "2cae99d6-0475-42fe-9fa7-ca27e03077de",
+    "RzlPushTemperatureUrl": "http://121.40.140.74:8071/api/CollectData/TemperatureTranspond",
+    "RzlPushTranspondUrl": "http://121.40.140.74:8071/api/CollectData/ThirdPushTranspond",
+    "RzlVoiceCallback": "http://121.40.140.74/GspCardStand/api/Voice/VoiceCallback"
+  },
+  "ServiceConfig": {
+    "TelpoDataUrl": "https://ai.ssjlai.com/data",
+    "KafkaBootstrapServers": "172.19.42.40:9092,172.19.42.41:9092,172.19.42.48:9092",
+    "KafkaTopics": [ "topic.push.WanJiaAn" ],
+    "KafkaGroupId": "telpo.WanJiaAn",
+    "CacheDurationSeconds": 1200, //20分钟
+    "CacheDurationSeconds10": 600 //10分钟
+  },
+  "Redis": {
+    "Server": "172.19.42.40:8090",
+    "Password": "telpo#1234",
+    "DefaultDatabase": 7,
+    "Poolsize": 150,
+    "Preheat": 50,
+    "Prefix": "_Third_",
+    "ConnectTimeout": 5000,
+    "IdleTimeout": 20000
+  },
+  "Environment": "production"
+}

TelpoPush.WanJiaAn.Worker/appsettings.test.json

@@ -0,0 +1,42 @@
+{
+  "ConnectionStrings": {
+    "DB_Connection_String": "server=139.224.254.18;port=3305;database=gps_card;uid=root;pwd=telpo#1234;CharSet=utf8;MinimumPoolSize=10;MaximumPoolSize=1000;SslMode=none",
+    "Telpo_common_ConnString": "server=139.224.254.18;port=3305;database=telpo_common;uid=root;pwd=telpo#1234;CharSet=utf8;MinimumPoolSize=10;MaximumPoolSize=1000;SslMode=none",
+    "Telpo_Healthy_ConnString": "server=139.224.254.18;port=3305;database=telpo_healthy;uid=root;pwd=telpo#1234;CharSet=utf8;MinimumPoolSize=10;MaximumPoolSize=1000;SslMode=none"
+
+    //"DB_Connection_String": "server=rm-uf6j529mu0v6g0btpco.mysql.rds.aliyuncs.com;port=3305;database=gps_card;uid=root;pwd=telpo#1234;CharSet=utf8;MinimumPoolSize=10;MaximumPoolSize=1000;SslMode=none",
+    //"Telpo_common_ConnString": "server=rm-uf6j529mu0v6g0btpco.mysql.rds.aliyuncs.com;port=3305;database=telpo_common;uid=linwl;pwd=linwl#1234;CharSet=utf8;MinimumPoolSize=10;MaximumPoolSize=1000;SslMode=none"
+  },
+
+  "WanJiaAnConfig": {
+    "CoreServiceApiUrl": "http://localhost:8094/api/v1",
+    "GpsWebApiUrl": "http://id.ssjlai.com/webapi",
+    //-------融智联
+    "RzlManufactorId": "2cae99d6-0475-42fe-9fa7-ca27e03077de",
+    "RzlPushTemperatureUrl": "http://121.40.140.74:8071/api/CollectData/TemperatureTranspond",
+    "RzlPushTranspondUrl": "http://121.40.140.74:8071/api/CollectData/ThirdPushTranspond",
+    "RzlVoiceCallback": "http://121.40.140.74/GspCardStand/api/Voice/VoiceCallback"
+  },
+  "ServiceConfig": {
+    "TelpoDataUrl": "http://id.ssjlai.com/data",
+    "KafkaBootstrapServers": "139.196.106.140:9092", // "172.19.42.53:9092",
+    "KafkaTopics": [ "topic.wanjiaan.push.telpo" ],
+    "KafkaGroupId": "telpo.wanjiaan",
+    "KafkaUserName": "wanjiaan",
+    "KafkaPassword": "wanjiaan@telpo#123",
+    "CacheDurationSeconds": 1200, //20分钟
+    "CacheDurationSeconds10": 600 //10分钟
+  },
+
+  "Redis": {
+    "Server": "172.19.42.44:8090",
+    "Password": "telpo#1234",
+    "DefaultDatabase": 7,
+    "Poolsize": 150,
+    "Preheat": 50,
+    "Prefix": "_Third_",
+    "ConnectTimeout": 5000,
+    "IdleTimeout": 20000
+  },
+  "Environment": "dev"
+}

TelpoPushWanJiaAn.sln

@@ -0,0 +1,25 @@
+
+Microsoft Visual Studio Solution File, Format Version 12.00
+# Visual Studio Version 17
+VisualStudioVersion = 17.9.34902.65
+MinimumVisualStudioVersion = 10.0.40219.1
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TelpoPush.WanJiaAn.Worker", "TelpoPush.WanJiaAn.Worker\TelpoPush.WanJiaAn.Worker.csproj", "{5A474148-7D68-4B7A-8F93-C32D3F8064FD}"
+EndProject
+Global
+	GlobalSection(SolutionConfigurationPlatforms) = preSolution
+		Debug|Any CPU = Debug|Any CPU
+		Release|Any CPU = Release|Any CPU
+	EndGlobalSection
+	GlobalSection(ProjectConfigurationPlatforms) = postSolution
+		{5A474148-7D68-4B7A-8F93-C32D3F8064FD}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{5A474148-7D68-4B7A-8F93-C32D3F8064FD}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{5A474148-7D68-4B7A-8F93-C32D3F8064FD}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{5A474148-7D68-4B7A-8F93-C32D3F8064FD}.Release|Any CPU.Build.0 = Release|Any CPU
+	EndGlobalSection
+	GlobalSection(SolutionProperties) = preSolution
+		HideSolutionNode = FALSE
+	EndGlobalSection
+	GlobalSection(ExtensibilityGlobals) = postSolution
+		SolutionGuid = {E49EC01C-7F86-406A-AAAD-57FF5685D214}
+	EndGlobalSection
+EndGlobal

nuget.config

@@ -0,0 +1,7 @@
+<?xml version="1.0" encoding="utf-8"?> 
+<configuration> 
+	<packageSources> 
+		<add key="nuget.org" value="https://api.nuget.org/v3/index.json" protocolVersion="3" /> 
+		<add key="company" value="http://139.224.254.18:3344" /> 
+	</packageSources> 
+</configuration>

setup_production.sh

@@ -0,0 +1,18 @@
+#!/usr/bin/env bash
+image_version=$version
+# 删除镜像
+docker rmi -f $(
+  docker images | grep registry.cn-shanghai.aliyuncs.com/gps_card/telpo_push_wanjiaan | awk '{print $3}'
+)
+# 构建telpo/mrp:$image_version镜像
+docker build -f ./TelpoPush.WanJiaAn.Worker/Dockerfile . -t telpo/telpo_push_wanjiaan:$image_version
+#TODO:推送镜像到阿里仓库
+echo '=================开始推送镜像======================='
+docker login --username=rzl_wangjx@1111649216405698 --password=telpo.123 registry.cn-shanghai.aliyuncs.com
+docker tag telpo/telpo_push_wanjiaan:$image_version registry.cn-shanghai.aliyuncs.com/gps_card/telpo_push_wanjiaan:$image_version
+docker push registry.cn-shanghai.aliyuncs.com/gps_card/telpo_push_wanjiaan:$image_version
+echo '=================推送镜像完成======================='
+#删除产生的None镜像
+docker rmi -f $(docker images | grep none | awk '{print $3}')
+# 查看镜像列表
+docker images

setup_test.sh

@@ -0,0 +1,17 @@
+#!/usr/bin/env bash
+image_version=$version
+# 删除镜像
+docker rmi -f $(
+  docker images | grep 139.224.254.18:5000/telpo_push_wanjiaan | awk '{print $3}'
+)
+# 构建telpo/mrp:$image_version镜像
+docker build -f ./TelpoPush.Position.Worker/Dockerfile . -t telpo/telpo_push_wanjiaan:$image_version
+#TODO:推送镜像到私有仓库
+echo '=================开始推送镜像======================='
+docker tag telpo/telpo_push_wanjiaan:$image_version 139.224.254.18:5000/telpo_push_wanjiaan:$image_version
+docker push 139.224.254.18:5000/telpo_push_wanjiaan:$image_version
+echo '=================推送镜像完成======================='
+#删除产生的None镜像
+docker rmi -f $(docker images | grep none | awk '{print $3}')
+# 查看镜像列表
+docker images

telpo_push_wanjiaan_run.sh

@@ -0,0 +1,27 @@
+#!/bin/bash
+environment=$1
+version=$2
+echo "环境变量为${environment},版本为$version!"
+if [[ ${environment} == 'production' ]]; then
+	echo "开始远程构建容器"
+	docker stop telpo_push_wanjiaan || true
+	docker rm telpo_push_wanjiaan || true
+	docker rmi -f $(docker images | grep registry.cn-shanghai.aliyuncs.com/gps_card/telpo_push_wanjiaan | awk '{print $3}')
+	docker login --username=rzl_wangjx@1111649216405698 --password=telpo.123 registry.cn-shanghai.aliyuncs.com
+	docker pull registry.cn-shanghai.aliyuncs.com/gps_card/telpo_push_wanjiaan:$version
+	docker run -d --network=host -e environment=production -v /home/data/telpo_push_wanjiaan/log:/var/telpo_pushwanjiaan/logs --restart=always --name telpo_push_wanjiaan registry.cn-shanghai.aliyuncs.com/gps_card/telpo_push_wanjiaan:$version;
+	#删除产生的None镜像
+	docker rmi -f $(docker images | grep none | awk '{print $3}')
+	docker ps -a
+
+elif [[ ${environment} == 'test' ]]; then
+	echo "开始在测试环境远程构建容器"
+	docker stop telpo_push_wanjiaan || true
+	docker rm telpo_push_wanjiaan || true
+	docker rmi -f $(docker images | grep 139.224.254.18:5000/telpo_push_wanjiaan | awk '{print $3}')
+	docker pull 139.224.254.18:5000/telpo_push_wanjiaan:$version
+	docker run -d --network=host -e environment=${environment} -v /home/data/telpo_push_wanjiaan/log:/var/telpo_pushwanjiaan/logs --restart=always --name telpo_push_wanjiaan 139.224.254.18:5000/telpo_push_wanjiaan:$version;
+	#删除产生的None镜像
+	docker rmi -f $(docker images | grep none | awk '{print $3}')
+	docker ps -a
+fi