zoukankan      html  css  js  c++  java
  • 多线程+异步 遇到的异常:关键业务未执行前就结束了主线程

    背景:

    1:api提前生成一批数据

    2:hangfire服务中采用 异步(Task)+多线程(Parallel)方式 一个发送第三方消息的服务(每一分钟执行一次)


    问题描述:

    一条消息 发送多次 4、5-N次


    代码:

    hangfire

    /// <summary>
        /// 群发活动服务
        /// </summary>
        public class EnterpriseGroupSendJob : IRecurringJob
        {
    
            private readonly ILogger<EnterpriseGroupSendJob> _logger;
            private readonly string ConnectionString = ConfigurationManager.GetValue("ConnectionString");
            private readonly IEnterpriseGroupSendService _groupService;
            private static object TaskLock = new object();
            public EnterpriseGroupSendJob(ILogger<EnterpriseGroupSendJob> logger, IEnterpriseGroupSendService groupService)
            {
                _logger = logger;
                _groupService = groupService;
            }
    
            public void Execute(PerformContext context)
            {
                //_logger.LogInformation($"【EnterpriseGroupSendJob】");
                lock (TaskLock)
                {
                    //1:获取群发消息主表
                    //1.1:如果处于待发送 状态 并且时间 小于等于当前时间的
                    //2:获取 根据商户  获取这部分的 群发活动 子表数据
    
                    //3:根据商户 开启线程 请求 wehub接口 (文本、图片、视频)
                    //3.1:更新 群发活动主表表 状态 (加锁 TaskLock 不会执行两次),
    
                    //4:在task里写回调更新 字表的 发送状态
    
                    //5:将群发消息推送到crm逻辑:
                    //5.1 表 enterprise_message 与易赚关联表  enterprise_send_msg 消息体记录表 enterprise_groupsend_detail 业务表 三表通过guid关联
                    //5.2 生成数据时将三表数据统一生成,发送逻辑不变  更新 send_msg表的 update状态 
                    //5.3 回调里需要根据uuid customerid找到 send_msg 表里的 requestmsg 同步给crm 
                    var sqlTitle = "企业微信-群发活动服务" + DateTime.Now;
                    _logger.LogInformation($"【{sqlTitle}开始】");
                    var baseDate = DateTime.Now;
                    var groupSendMainList = _groupService.GetGroupSendMainIdList(baseDate);
                    if (groupSendMainList.Count() == 0)
                    {
                        //_logger.LogInformation($"【{sqlTitle} 未查询到群发活动】");
                        return;
                    }
    
                    var groupSendDetailList = _groupService.GetGroupSendDetailList(groupSendMainList);
                    if (groupSendDetailList.Count() == 0)
                    {
                        //_logger.LogInformation($"【{sqlTitle} 未查询到群发活动子表数据】");
                        return;
                    }
                    groupSendDetailList = _groupService.GetEligibleGroup(groupSendDetailList, sqlTitle, baseDate);
                    if (groupSendDetailList.Count() == 0)
                    {
                        //_logger.LogInformation($"【{sqlTitle} 没有符合企业微信有效期内条件的群发活动子表数据】");
                        return;
                    }
                    ParallelOptions parallelOptions = new ParallelOptions() { MaxDegreeOfParallelism = Convert.ToInt32(ConfigurationManager.GetValue("GroupSendCondition:TaskCount")) };
    
                    Parallel.ForEach(groupSendDetailList.DistinctEx(s => s.CompId).Select(s => s.CompId).ToList(), parallelOptions,groupSendCompId =>
                        {
                            _groupService.GroupSend(groupSendMainList.Where(s => s.CompId == groupSendCompId).ToList(), groupSendDetailList.Where(s => s.CompId == groupSendCompId).ToList(), sqlTitle);
                        });
                    
    
                    _logger.LogInformation($"【{sqlTitle}结束】");
    
                }
            }
    
        }
    View Code

    service

    GetGroupSendMainIdList

    /// <summary>
            /// //1:获取群发消息主表
            /// </summary>
            /// <returns></returns>
            public List<EnterpriseSendMain> GetGroupSendMainIdList(DateTime baseDate)
            {
                StringBuilder sql = new StringBuilder().AppendFormat("select Id,ContentType,Title,Content,CompId,MaterialSource,MaterialTitle,HomePicUrl,MaterialCode,MaterialURL from enterprise_groupsend_main where state={0} and SendTime<='{1}' order by sendTime ", (int)SolicitGroupSendMainStateEnum.待推送, baseDate.ToString("yyyy-MM-dd HH:mm:ss"));
                List<EnterpriseSendMain> result = new List<EnterpriseSendMain>();
                using (var conn = new MySqlConnection(ConnectionString))
                {
                    result = conn.Query<EnterpriseSendMain>(sql.ToString()).ToList();
                }
                return result;
            }
    View Code

    GroupSend

     /// <summary>
            /// 3推送内容
            /// </summary>
            /// <param name="groupSendMainList"></param>
            /// <param name="groupSendDetailList"></param>
            /// <param name="baseDate"></param>
            public async void GroupSend(List<EnterpriseSendMain> groupSendMainList, List<EnterpriseGroupSendDetail> groupSendDetailList, string sqlTitle, int reSendTime = 0)
            {
                //这个方法可能会出现的异常 :  本方法 async 且 await pushVideo  
                //会导致pushvideo之前不更新 enterprise_groupsend_main表的数据 1分钟后 新的服务启动后 重新走一遍流程
                //异步方法中 关键控制节点不异步处理
                foreach (var groupSendMain in groupSendMainList)
                {
                    if (groupSendMain.MaterialSource == 1)
                    {
                        PushVideo(groupSendMain, groupSendDetailList.Where(s => s.MainId == groupSendMain.Id).ToList());
                    }
                    else
                    {
                        switch (groupSendMain.ContentType)
                        {
                            case 1://文本
                                PushVideo<YZTextMessageRequest>(groupSendMain, groupSendDetailList.Where(s => s.MainId == groupSendMain.Id).ToList(), sqlTitle);
                                break;
                            case 2://图片
                                PushVideo<YZImageMessageRequest>(groupSendMain, groupSendDetailList.Where(s => s.MainId == groupSendMain.Id).ToList());
                                break;
                            case 3://视频
                                PushVideo<YZVideoRequest>(groupSendMain, groupSendDetailList.Where(s => s.MainId == groupSendMain.Id).ToList());
                                break;
                            default:
                                break;
                        }
                    }
                    string sql = $"update enterprise_groupsend_main set state={(int)SolicitGroupSendMainStateEnum.已推送} where id={groupSendMain.Id} and state={(int)SolicitGroupSendMainStateEnum.待推送}";
                    //_logger.LogInformation($"【{sqlTitle} 修改群发主表状态sql】:{sql}");
                    using (var conn = new MySqlConnection(ConnectionString))
                    {
                        conn.Execute(sql);
                    }
                    if (reSendTime > 0)
                    {
                        if (groupSendDetailList.Count(s => s.MainId == groupSendMain.Id) > 0)
                        {
                            string sqlUpdateDetailReSendTime = $"update enterprise_groupsend_detail set ResendTimes ={reSendTime} where sendState=9 and id in ({string.Join(",", groupSendDetailList.Select(s => s.Id))});";
                            //_logger.LogInformation($"【{sqlTitle} 修改群发子表ResendTimes sql】:{sqlUpdateDetailReSendTime}");
                            using (var conn = new MySqlConnection(ConnectionString))
                            {
                                conn.Execute(sqlUpdateDetailReSendTime);
                            }
                        }
                    }
                }
            }
    View Code

    PushVideo

            /// <summary>
            /// 3.1发送小程序
            /// </summary>
            /// <param name="carManager"></param>
            /// <param name="h5RequestList"></param>
            private async Task PushVideo(EnterpriseSendMain groupSendMain, List<EnterpriseGroupSendDetail> groupSendDetailList)
            {
                var messageList = GetEnterpriseMessage(groupSendDetailList.Select(s => s.UUID).ToList(), new List<EnterpriseMessage>());
                var sendMessage = GetEnterpriseSendMsg(groupSendDetailList.Select(s => s.UUID).ToList(), new List<EnterpriseSendMsg>(), groupSendMain.CompId);
                using (var conn = new MySqlConnection(ConnectionString))
                {
                    foreach (var groupSendDetail in groupSendDetailList)
                    {
                        var videoRequest = new List<EnterprisePushMessageBase<YZMiniProgramRequest>>();
                        var mssage = messageList.Where(s => s.CustomerId == groupSendDetail.UUID).FirstOrDefault();
                        if (mssage != null)
                        {
                            var instance = new YZMiniProgramRequest()
                            {
                                AccountOrgianlId = WechatMiniAccountOrgianlId,
                                Cover = groupSendMain.HomePicUrl,
                                Icon = WechatMiniIcon,
                                Title = groupSendMain.MaterialTitle,
                                Url = CheckEnterpriseMiniProgramUrl(groupSendMain, groupSendDetail),
                            };
                            videoRequest.Add(new EnterprisePushMessageBase<YZMiniProgramRequest>()
                            {
                                Content = instance,
                                NotifyCustomId = mssage.Id.ToString(),
                            });
                            var result = await _yZOperateService.TYZPush(
                                                                         new YZBaseRequest<EnterprisePushMessageBase<YZMiniProgramRequest>>()
                                                                         {
                                                                             EmployeeId = groupSendDetail.EmployeeId.ToString(),
                                                                             CompId = groupSendDetail.CompId,
                                                                             EnterpriseId = groupSendDetail.EnterpriseId,
                                                                             To = groupSendDetail.ExternalUserId,
                                                                             ToType = 0,
                                                                             MsgList = videoRequest,
                                                                         });
                            var sendmsg = sendMessage.FirstOrDefault(s => s.Customid == groupSendDetail.UUID);
                            if (sendmsg != null)
                            {
                                //更新表EnterpriseSendMsg 的requestmsg
                                YZMiniProgramRequestExtend extendMiniRequest = new YZMiniProgramRequestExtend()
                                {
                                    AccountOrgianlId = instance.AccountOrgianlId,
                                    Cover = instance.Cover,
                                    Icon = instance.Icon,
                                    Title = instance.Title,
                                    Url = instance.Url,
                                    OriginalContentUrl = groupSendMain.MaterialURL
                                };
                                var sql = "update enterprise_send_msg set RequestMsg='" + extendMiniRequest.ToJson() + "'";
                                if (result.Status == 1)
                                {
                                    sql += " , state=2 ";
                                }
                                sql += " where id=" + sendmsg.Id;
                                await conn.ExecuteAsync(sql);
                                //_logger.LogInformation("【群发更新enterprise_send_msg】" + sql);
                            }
                        }
                    }
                }
            }
    
            /// <summary>
            /// 拼接易赚发送小程序url参数
            /// </summary>
            /// <param name="groupSendMain"></param>
            /// <param name="groupSendDetail"></param>
            /// <returns></returns>
            public string CheckEnterpriseMiniProgramUrl(EnterpriseSendMain groupSendMain, EnterpriseGroupSendDetail groupSendDetail)
            {
                var url = "";
                switch (groupSendMain.ContentType)
                {
                    case 3:
                        url = string.Format(EnterPriseConvideo, groupSendMain.MaterialCode, groupSendMain.CompId, groupSendDetail.ExternalUserId);
                        break;
                    case 4:
                        url = string.Format(EnterPriseConarticle, groupSendMain.MaterialCode, groupSendMain.CompId, groupSendDetail.ExternalUserId);
                        break;
                    case 2:
                        url = string.Format(EnterPriseConposter, groupSendMain.MaterialCode, groupSendMain.CompId, groupSendDetail.ExternalUserId);
                        break;
                    default:
                        break;
                }
                return url;
            }
    
            /// <summary>
            /// 3.1推送
            /// </summary>
            /// <param name="carManager"></param>
            /// <param name="h5RequestList"></param>
            private async Task PushVideo<T>(EnterpriseSendMain groupSendMain, List<EnterpriseGroupSendDetail> groupSendDetailList, string sqlTitle = "")
            {
                var instance = Activator.CreateInstance(typeof(T));
    
                if (typeof(T) == typeof(YZTextMessageRequest))
                {
                    var instancemsg = instance.GetType().GetProperty("Text");
                    instancemsg.SetValue(instance, groupSendMain.Content);
                }
                else if (typeof(T) == typeof(YZImageMessageRequest))
                {
                    var instancemsg = instance.GetType().GetProperty("ImageHttpUrl");
                    instancemsg.SetValue(instance, groupSendMain.Content);
                }
                else if (typeof(T) == typeof(YZVideoRequest))
                {
                    var instancemsg = instance.GetType().GetProperty("VideoHttpUrl");
                    instancemsg.SetValue(instance, groupSendMain.Content);
                }
                var messageList = GetEnterpriseMessage(groupSendDetailList.Select(s => s.UUID).ToList(), new List<EnterpriseMessage>());
                var sendMessage = GetEnterpriseSendMsg(groupSendDetailList.Select(s => s.UUID).ToList(), new List<EnterpriseSendMsg>(), groupSendMain.CompId);
                using (var conn = new MySqlConnection(ConnectionString))
                {
                    foreach (var groupSendDetail in groupSendDetailList)
                    {
                        var videoRequest = new List<EnterprisePushMessageBase<T>>();
                        var mssage = messageList.Where(s => s.CustomerId == groupSendDetail.UUID).FirstOrDefault();
                        if (mssage != null)
                        {
                            videoRequest.Add(new EnterprisePushMessageBase<T>()
                            {
                                Content = (T)instance,
                                NotifyCustomId = mssage.Id.ToString(),
                            });
                            _logger.LogInformation(sqlTitle + "----" + groupSendDetail.ExternalUserId);
                            var result = await _yZOperateService.TYZPush(new YZBaseRequest<EnterprisePushMessageBase<T>>()
                            {
                                EmployeeId = groupSendDetail.EmployeeId.ToString(),
                                CompId = groupSendDetail.CompId,
                                EnterpriseId = groupSendDetail.EnterpriseId,
                                To = groupSendDetail.ExternalUserId,
                                ToType = 0,
                                MsgList = videoRequest,
                            });
                            //发送记录
                            var sendmsg = sendMessage.FirstOrDefault(s => s.Customid == groupSendDetail.UUID);
                            if (sendmsg != null)
                            {
                                //更新表EnterpriseSendMsg 的requestmsg
    
                                var sql = "update enterprise_send_msg set RequestMsg='" + instance.ToJson() + "'";
                                if (result.Status == 1)
                                {
                                    sql += " , state=2 ";
                                }
                                sql += " where id=" + sendmsg.Id;
                                await conn.ExecuteAsync(sql);
                                //_logger.LogInformation("【群发更新enterprise_send_msg】" + sql);
                            }
                        }
                    }
                }
    
            }
    View Code

    问题出发条件:

    正常情况下没问题,问题出在了 超过500+的时候


    问题所在:

    1:GroupSend 采用了 async ,

    2:hangfire.job.Execute 不支持 async  在 Parallel.foreach的时候 没有 await 或者 .wait

    3:PushVideo 采用了await

    4:导致 groupsend方法 中关键的一个 sql没有执行 时 就跑完了本次 服务 ,然后 下一次服务进来后 还会再跑一边逻辑。


    解决方案:

    1:groupsend 不采用 async 

    2:pushvideo 不 await 


    核心:

    异步、多线程服务中 关键控制节点 保持主线程中写完。

  • 相关阅读:
    python 修改excel
    python 正则表达式规则收集
    anusplina 4.36版本使用提示 说明
    python 汉字编码问题
    python IOError: [Errno 22] invalid mode ('r') or filename:
    将矩阵数据转换为栅格图 filled.contour()
    将数值矩阵映射为栅格图
    一脚踩进java之基础篇23——常用API(Object、String)
    一脚踩进java之基础篇22——面向对象 (不同修饰符使用细节)
    一脚踩进java之基础篇21——面向对象 (访问修饰符、代码块)
  • 原文地址:https://www.cnblogs.com/zhaokunbokeyuan256/p/12768092.html
Copyright © 2011-2022 走看看