ZKLT/Crontab/Job/HttpJob.cs

using Newtonsoft.Json.Linq;
using Quartz;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Reflection.PortableExecutable;
using System.Text;
using System.Threading.Tasks;
using ZKLT.Quartz.Model;
using ZKLT.Hadoop.Interface;
using Newtonsoft.Json;
using ZKLT.Hadoop.Model;
using System.Dynamic;
namespace ZKLT.Quartz.Job
{
    /// <summary>
    /// HTTP作业类
    /// </summary>
    public class HttpJob : IJob
    {
        public async Task Execute(IJobExecutionContext context)
        {
            JobDataMap dataMap = context.JobDetail.JobDataMap;
            IHadoopService _HadoopService = (IHadoopService)dataMap.Get("hadoop");
            // 任务配置项
            List<QZ_JobConfig> configs = (List<QZ_JobConfig>)dataMap.Get("Params");
            // 任务Id
            string taskId = dataMap.Get("TaskId").ToString();
            // 是否开启日志
            int isLog = (int)dataMap.Get("IsLog");
            // 请求地址
            string url = configs.Find(_item => _item.Key == "Url").Value;
            // 请求类型
            HttpMethod httpMethod = configs.Find(_item => _item.Key == "Method").Value == "GET" ? HttpMethod.Get : HttpMethod.Post;
            // 返回数据储存表
            string bindDatabase = configs.Find(_item => _item.Key == "BindDatabse").Value;
            // 返回数据集合名
            string resultSet = configs.Find(_item => _item.Key == "ResultSet").Value;
            using (HttpClient client = new HttpClient())
            {
                try
                {
                    var resultColumnConfigs = new List<QZ_JobConfig>();
                    HttpRequestMessage request = new HttpRequestMessage(httpMethod, url);
                    foreach (var _item in configs)
                    {
                        if (_item.Group! == "Headers")
                        {
                            request.Headers.Add(_item.Key, _item.Value);
                        }
                        else if (_item.Group! == "ResultColumns")
                        {
                            resultColumnConfigs.Add(_item);
                        }
                    }
                    //请求提交
                    HttpResponseMessage response = await client.SendAsync(request);
                    response.EnsureSuccessStatusCode();
                    string responseBody = await response.Content.ReadAsStringAsync();
                    var responseObject = JsonConvert.DeserializeObject<JObject>(responseBody);
                    JArray responseList = (JArray)responseObject.GetValue(resultSet);
                    // 如果设定了绑定表 插入表中
                    if (bindDatabase != null)
                    {
                        HDP_Command[] commands = makeCommand(bindDatabase, responseList, resultColumnConfigs);
                        _HadoopService.PatchCommand(commands);
                    }
                    // 日志开启，插入日志
                    if (isLog == HDP_Task.LOGOPEN)
                    {
                        var data = new HDP_TaskLog() 
                        {
                            Id = "TL" + DateTime.Now.ToUniversalTime(),
                            TaskId = taskId,
                            Content = responseBody
                        };
                        var logCommand = new HDP_Command()
                        {
                            TableId = HDP_TaskLog.TABLEID,
                            Data = JToken.FromObject(data)
                        };
                        _HadoopService.Insert(logCommand);
                    }
                }
                catch (HttpRequestException e)
                {
                    Console.WriteLine($"请求错误{e.Message}");
                }
            }
            

            await Task.CompletedTask;
        }

        public HDP_Command[] makeCommand(string bindDatabase,JArray responseList,List<QZ_JobConfig> configList)
        {
            HDP_Command[] commands = new HDP_Command?[responseList.Count];
            foreach (var _item in responseList)
            {
                var command = new HDP_Command();
                command.TableId = bindDatabase;
                command.Type = HDP_CommandType.INSERT;
                var data = new JObject();
                foreach (var _config in configList)
                {
                    data[_config.Value] = _item[_config.Key];
                }
                command.Data = data;
                int _index = responseList.IndexOf(_item);
                commands[_index] = command;
            }
            return commands;
        }
    }
}