Web_IoTBase_Sever_Prod/EMQX_HttpAuth/services/KafkaProduce.cs

using System.Text;
using System.Text.Json;
using System.Threading.Channels;
using Common;
using CommonEntity;
using CommonEntity.RCUEntity;
using CommonTools;
using Confluent.Kafka;
using Google.Protobuf;
using MessagePack;
using Microsoft.Extensions.Caching.Memory;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using NLog;
using RestSharp;
using static CSRedis.CSRedisClient;

namespace BLWLogProduce.Services
{
    public class KafkaProduce : BackgroundService
    {
        public IConfiguration Configuration { get; set; }
        public IMemoryCache _Cache { get; set; }

        public KafkaProduce(IConfiguration configuration, IMemoryCache cache)
        {
            this.Configuration = configuration;
            _Cache = cache;
        }

        //private static Channel<string> _messageChannel;
        public static Logger logger = LogManager.GetCurrentClassLogger();
        protected async override Task ExecuteAsync(CancellationToken stoppingToken)
        {
            await Task.Factory.StartNew((state) =>
            {
                try
                {
                    string? ipport = Configuration["Kafka:EndPoint"];
                    string? user = Configuration["Kafka:UserName"];
                    string? pwd = Configuration["Kafka:PassWord"];
                    var config = new ProducerConfig
                    {
                        BootstrapServers = ipport,
                        SecurityProtocol = SecurityProtocol.SaslPlaintext,
                        SaslMechanism = SaslMechanism.Plain,
                        SaslUsername = user,
                        SaslPassword = pwd
                    };
                    var p = new ProducerBuilder<string, byte[]>(config).Build();

                    // JS= #{hostnum=>VS1,endpoint=>JieGuo,data=>RedisData,currenttimestamp=>CurrentTimestamp}
                    var DingYue1 = (RedisKey.NewVersion_RCU_Data, new Action<SubscribeMessageEventArgs>(async (args) =>
                    {
                        string body = args.Body;
                        var USA = JsonConvert.DeserializeObject<RCU_UDPData>(body);
                        var UUA = USA.endpoint;
                        var hostnum = USA.hostnum;

                        var timestamp = USA.currenttimestamp;
                        var data = USA.data;
                        var endpoint_1 = Encoding.ASCII.GetString(UUA);
                        var hostnum_1 = Encoding.ASCII.GetString(hostnum);

                        RCU_UDPData_With_String rs = new RCU_UDPData_With_String();
                        rs.currenttimestamp = timestamp;
                        rs.data = data;
                        rs.hostnum = hostnum_1;
                        rs.endpoint = endpoint_1;

                        byte[] qf = MyMessagePacker.FastSerialize(rs);

                        string TopicKey = KafkaKey.New_RCU_Data_Topic;
                        string DetailKey = KafkaKey.UDPckageAllUDPDataKey;


                        await p.ProduceAsync(TopicKey, new Message<string, byte[]> { Key = DetailKey, Value = qf });

                    }));

                    //var DingYue = ("redis-power", new Action<SubscribeMessageEventArgs>(async (args) =>
                    //{
                    //    string body = args.Body;

                    //    object poo = null;
                    //    byte[] qf = MyMessagePacker.FastSerialize(poo);

                    //    string TopicKey = KafkaKey.BLWLog_RCU_Topic;
                    //    string DetailKey = KafkaKey.UDPPackagePowerMonitor;
                    //    await p.ProduceAsync(TopicKey, new Message<string, byte[]> { Key = DetailKey, Value = qf });


                    //    //宝镜系统使用
                    //    //EnergyConsumption ese = new EnergyConsumption();

                    //    //byte[] data = ese.ToByteArray();
                    //    byte[] data = null;
                    //    string TopicKey1 = "";
                    //    string DetailKey1 = "";

                    //    await p.ProduceAsync(TopicKey1, new Message<string, byte[]> { Key = DetailKey1, Value = data });

                    //}));

                    //CSRedisCacheHelper.redis.Subscribe(DingYue);
                    CSRedisCacheHelper.redis.Subscribe(DingYue1);
                }
                catch (Exception ex)
                {
                    logger.Error(ex.Message);
                    logger.Error(ex.StackTrace);
                }
            }, TaskCreationOptions.LongRunning, stoppingToken);
        }


        /// <summary>
        /// 百度api
        /// </summary>
        /// <returns></returns>
        public static string GetBaiduIp(string ip)
        {
            string location = "";
            try
            {
                string url = $"https://sp0.baidu.com";
                //WebClient client = new WebClient();
                RestSharp.RestClient client1 = new RestSharp.RestClient(url);
                RestSharp.RestRequest request = new RestSharp.RestRequest($"/8aQDcjqpAAV3otqbppnN2DJv/api.php?query={ip}&co=&resource_id=6006&oe=utf8", Method.Get);
                var buffer = client1.DownloadData(request);
                //var buffer = client.DownloadData(url);
                string jsonText = Encoding.UTF8.GetString(buffer);
                JObject jo = JObject.Parse(jsonText);

                Root root = JsonConvert.DeserializeObject<Root>(jo.ToString());
                foreach (var item in root.data)
                {
                    location = item.location;
                }
                return location;
            }
            catch (Exception ex)
            {
                //Console.WriteLine(ex);
                return location;
            }

        }
    }
}