多线程的使用2

逃离我推掉我的手 2022-08-31 04:54 51阅读 0赞

@Path("/sms")
    public class SmsResource { 
    
        @InjectParam
        private SmsService msgSendService;
        /** * 短信发送请求保存接口 * * @param param 入参 @require=true @schema={@link SmsReq} * @return 返回结果 @schema={@link com.lufax.wmc.ext.common.bean.LUResponse} @codeName=res_code * @returnCase success @code=0 @schema={@link com.lufax.wmc.ext.common.bean.LUResponse } */
        @POST
        @Path("/send")
        @Produces(value = MediaType.APPLICATION_JSON)
        public String sendSms(@FormParam("param") String param) { 
            Logger.info(this, "received sms request here: the param is:" + param);
            SmsReq smsReq = JSON.parseObject(param, SmsReq.class);
            Validator.isValid(smsReq);
            BaseMsgContext context = MsgContextFactory.getContextBuilder(MsgType.SMS).build(smsReq);
            msgSendService.doSend(context);
            return new LUResponse().toJsonStr();
        }
    }

public interface SmsService { 
        void doSend(BaseMsgContext Context);
        List<SmsDetailCount> countSmsDetailByBatchId(SmsDetailCountReq req);
        PaginationGsonV2<GetSmsDetail> getSmsDetailByBatchIdPagination(GetSmsDetailReq req);
        void fixIncorrectStatus();
    }

@Service
    public class SmsServiceImpl implements SmsService { 
    
        @Autowired
        @Qualifier("smsSendProcessingTaskExecutor")
        private Executor smsSendProcessingTaskExecutor;
        @Autowired
        private BeSmsRequestDetailsDAO beSmsRequestDetailsDAO;
        @Autowired
        private SmsRequestDAO smsRequestDAO;
        @Autowired
        private SmsRequestDetailsDAO smsRequestDetailsDAO;
        @Autowired
        private ExtProperties properties;
    
    
       @Override
        public void doSend(BaseMsgContext context){ 
            Preconditions.isTrue(context.getMsgType() != null, ResultCode.BAD_PARAMETER);
            ParamStrategyContext paramStrategyContext = new ParamStrategyContext(context.getMsgType());
    
            paramStrategyContext.doParamProcess(context);
            // 异步发送业务短信
            MsgRealtimeSenderRunnable smsRealtimeSenderRunnable = new MsgRealtimeSenderRunnable(context);
            smsSendProcessingTaskExecutor.execute(smsRealtimeSenderRunnable);
        }
    
        @Override
        public List<SmsDetailCount> countSmsDetailByBatchId(SmsDetailCountReq req){ 
            return this.beSmsRequestDetailsDAO.countSmsDetailByBatchId(req.getBatchIdList(), req.getStartTime(), req.getEndTime());
        }
        @Override
        public PaginationGsonV2<GetSmsDetail> getSmsDetailByBatchIdPagination(GetSmsDetailReq req) { 
            // step 1: 计算totalCount
            long totalCount = beSmsRequestDetailsDAO.getSmsDetailByBatchIdCount(req.getBatchId(), req.getStatus());
            PaginationGsonV2<GetSmsDetail> paginationGson = new PaginationGsonV2<>(ResultCode.OK, req.getPageSize(), totalCount, req.getPageNo());
            if (totalCount == 0) { 
                return paginationGson;
            }
            //获取分页数据
            int startRow = (req.getPageNo() - 1) * req.getPageSize();
            //int endRow = req.getPageNo() * req.getPageSize();
            // 分页数据改动--mysql
            int pageSize = req.getPageSize();
            List<GetSmsDetail> getSmsDetailList = beSmsRequestDetailsDAO.getSmsDetailByBatchIdPagination(req.getBatchId(), req.getStatus(), startRow, pageSize);
            paginationGson.setData(getSmsDetailList);
            return paginationGson;
        }
    
    
        @Override
        public void fixIncorrectStatus() { 
            int expiredDays = properties.getDefaultExpiredDays();
            Date expiredDay = new DateTime().minusDays(expiredDays).toDate();
            int processingTimeoutMinutes = properties.getDefaultProcessingTimeoutMinutes();
            int batchLimit = properties.getDefaultBatchLimit();
    
    
            // 先修复SmsRequestDetails
            int fixCount = 0;
            for (int i = 0; i < 20; i++) { 
                DateTime expiredTime = new DateTime().minusMinutes(processingTimeoutMinutes);
                SmsRequestDetailsStatus toStatus = SmsRequestDetailsStatus.INITIALIZATION;
    
                long countIncorrectStatusRecords = smsRequestDetailsDAO.countIncorrectStatusRecords(expiredDay, expiredTime.toDate());
                Logger.info(this, "Found SmsRequestDetail incorrect status count: " + countIncorrectStatusRecords);
                if (countIncorrectStatusRecords <= 0) { 
                    break;
                }
    
                Logger.info(this, String.format("smsRequestDetailsDao.fixIncorrectStatus() with params { expiredTime[%s], toStatus[%s], batchLimit[%s] } starting...", expiredTime, toStatus, batchLimit));
                fixCount = smsRequestDetailsDAO.fixIncorrectStatus(expiredDay, expiredTime.toDate(), toStatus.id(), batchLimit);
                Logger.info(this, String.format("smsRequestDetailsDao.fixIncorrectStatus() with params { expiredTime[%s], toStatus[%s], batchLimit[%s] } done. fixCount: [%s]", expiredTime, toStatus, batchLimit, fixCount));
    
                if (fixCount < batchLimit) break;
            }
    
            // 再修复SmsRequest
            DateTime expiredTime = new DateTime().minusMinutes(processingTimeoutMinutes);
            SmsRequestStatus toStatus = SmsRequestStatus.INITIALIZATION;
    
            long countIncorrectStatusRecords = smsRequestDAO.countIncorrectStatusRecords(expiredDay, expiredTime.toDate());
            Logger.info(this, "Found SmsRequest incorrect status count: " + countIncorrectStatusRecords);
            if (countIncorrectStatusRecords <= 0) { 
                return;
            }
    
            Logger.info(this, String.format("smsRequestDao.fixIncorrectStatus() with params { expiredTime[%s], toStatus[%s], batchLimit[%s] } starting...", expiredTime, toStatus, batchLimit));
            fixCount = smsRequestDAO.fixIncorrectStatus(expiredDay, expiredTime.toDate(), toStatus.id(), batchLimit);
            Logger.info(this, String.format("smsRequestDao.fixIncorrectStatus() with params { expiredTime[%s], toStatus[%s], batchLimit[%s] } done. fixCount: [%s]", expiredTime, toStatus, batchLimit, fixCount));
        }
    
    }

**
     * 异步发送
     */
    public class MsgRealtimeSenderRunnable implements Runnable { 
    
        private BaseMsgContext msgContext;
    
        public MsgRealtimeSenderRunnable(BaseMsgContext msgContext) { 
            this.msgContext = msgContext;
        }
    
        public void run() { 
            Long requestId = msgContext.getRequestId();
            try { 
                Logger.info(this, String.format("EmailRealtimeSenderRunnable.run, emailRequestId [%s] ", requestId));
                MsgProcessorFactory.getProcessor(msgContext.getMsgType()).doSend(msgContext);
            } catch (RequestNotExistedException e) { 
                Logger.warn(this, String.format("EmailRealtimeSenderRunnable.run, emailRequestId [%s] does not exist!", requestId), e);
            } catch (RepeatRequestException e) { 
                Logger.warn(this, String.format("EmailRealtimeSenderRunnable.run, emailRequestId [%s] repeated!", requestId), e);
            } catch (Exception e) { 
                Logger.error(this, String.format("EmailRealtimeSenderRunnable.run, emailRequestId [%s] error: [%s]!", requestId, e.getMessage()), e);
            }
        }
    
    }

@Service
    public class MsgProcessorFactory implements InitializingBean { 
        @Autowired
        @Qualifier("smsProcessor")
        private AbstractProcessor smsProcessor;
    
        @Autowired
        @Qualifier("imsProcessor")
        private AbstractProcessor imsProcessor;
    
        @Autowired
        @Qualifier("emailProcessor")
        private AbstractProcessor emailProcessor;
    
        @Autowired
        @Qualifier("getuiProcessor")
        private AbstractProcessor getuiProcessor;
    
    
        public  static final Map<MsgType, AbstractProcessor> builderMap = new HashMap<>();
        public static AbstractProcessor getProcessor(MsgType type) { 
            if (type == null) { 
                return null;
            }
            return builderMap.get(type);
        }
    
        @Override
        public void afterPropertiesSet() throws Exception { 
            builderMap.put(MsgType.SMS, smsProcessor);
            builderMap.put(MsgType.IMS, imsProcessor);
            builderMap.put(MsgType.EMAIL, emailProcessor);
            builderMap.put(MsgType.GETUI, getuiProcessor);
        }
    
    }

@Component
    public abstract class AbstractProcessor { 
    
        @Autowired
        private ExtProperties properties;
        @Autowired
        private MsgProcessorHandlerChainConfig msgProcessorHandlerChainConfig;
        protected AbstractProcessor getProcessor(){ 
            return this;
        }
    
        protected void doBefore(BaseMsgContext msgContext){ 
            return;
        }
        protected void doAfter(BaseMsgContext msgContext){ 
            return;
        }
        public void doSend(BaseMsgContext context){ 
            this.doBefore(context);
            for (IMsgSendHandler handler : this.getMsgProcessorHandlerChainConfig().getHandlerList()) { 
                handler.handle(context);
            }
            this.doAfter(context);
        }
    
    
        public MsgProcessorHandlerChainConfig getMsgProcessorHandlerChainConfig() { 
            return msgProcessorHandlerChainConfig;
        }
    
    }

public interface IMsgSendHandler { 
    
       void handle(BaseMsgContext context);
    }

@Component
    public class SenderFactory implements InitializingBean { 
        @Autowired
        @Qualifier("smsSender")
        private ISender smsSender;
    
        @Autowired
        @Qualifier("emailSender")
        private ISender emailSender;
    
        @Autowired
        @Qualifier("imsSender")
        private ISender imsSender;
    
        @Autowired
        @Qualifier("getuiSender")
        private ISender getuiSender;
    
        public  static final Map<MsgType, ISender> senderMap = new HashMap<>();
        public static ISender getStrategy(MsgType type) { 
            if (type == null) { 
                return null;
            }
            return senderMap.get(type);
        }
    
        @Override
        public void afterPropertiesSet(){ 
            senderMap.put(MsgType.SMS, smsSender);
            senderMap.put(MsgType.EMAIL, emailSender);
            senderMap.put(MsgType.IMS, imsSender);
            senderMap.put(MsgType.GETUI, getuiSender);
        }
    }

/** * 疲劳度处理接口 */
    public interface ISender { 
        void handle(BaseMsgContext context);
    }

@Service
    public class SmsSender implements ISender{ 
        @Autowired
        private SmsGwRemoteService smsGwRemoteService;
        @Autowired
        private SmsChannelRouter smsRouter;
        @Autowired
        private SmsResultHandler smsResultHandler;
    
    
        @Override
        public void handle(BaseMsgContext msgContext) { 
            SmsSenderContext context = new SmsSenderContext((SmsContext)msgContext);
            List<SmsRequestDetail> smsRequestDetails = context.getSmsRequestDetails();
            for (SmsRequestDetail detail : smsRequestDetails){ 
                if (SmsRequestDetailsStatus.PROCESSING.id() != detail.getStatus()
                        && SmsRequestDetailsStatus.PROCESSING_RETRY.id() != detail.getStatus()){ 
                    return;
                }
                //1.获取参数
                String locale = detail.getLocale();
                String mobileNo = detail.getMobileNo();
                String  mobileCountryCode = detail.getMobileCountryCode();
                String smsContent = detail.getSmsContent();
                SmsType smsType = SmsType.getById(detail.getSmsTemplate().getSmsType());
                String templateChannelList = detail.getSmsTemplate().getChannelList();
                Logger.debug(this, String.format("start to send sms message by channels, smsType=[%s],mobileNos=%s", smsType.getName(), HelperUtils.markMobileNos(mobileNo)));
                //2.模板通道配置优先级高
                String channelListStr = StringUtils.isNotEmpty(templateChannelList) ? templateChannelList : smsRouter.getChannelListStr(locale, mobileCountryCode, smsType);
                //3.发送（尝试不同的通道）
                LUResponse<SmsCallResult> response = callRemoteService(channelListStr, mobileCountryCode, mobileNo, smsContent);
                //4.处理发送结果
                smsResultHandler.handleSmsCallResponse(context.getSmsRequest(), response, detail.getId());
            }
            // 5.任务结束后会翻smsRequest请求的状态为处理完毕
        }
    
    
        private LUResponse<SmsCallResult> callRemoteService(String channelListStr, String mobileCountryCode, String mobileNo, String smsContent) { 
    
            String channelName = smsRouter.getChannelName(channelListStr, mobileNo);
            String mobileNoSend = mobileNo;
            //todo 此处若需要支持多个手机号的群发，mobileNo则是mobileNoList，以逗号分开，现在是单个手机号，暂不支持群发
            mobileNoSend = HelperUtils.isCnMobile(mobileCountryCode) ? mobileNo : mobileCountryCode.concat(mobileNo);
            LUResponse<SmsCallResult> response = smsGwRemoteService.sendSmsBySmsGw(mobileNoSend, smsContent, channelName);
            //失败后重试
            if (response.isSuccess() && response.getData() != null) { 
                List<ChannelNode> channelNodeList = smsRouter.buildChannelNodes(channelListStr);
                for (ChannelNode failChannelNode : channelNodeList) { 
                    String retryChannel = failChannelNode.getName();
                    if (!channelName.equals(retryChannel)) { 
                        Logger.debug(this, "fail try channels is = " + retryChannel);
                        response = smsGwRemoteService.sendSmsBySmsGw(mobileNoSend, smsContent, retryChannel);
                        if (response.isSuccess() && response.getData() != null) { 
                            continue;
                        } else { 
                            return response;
                        }
                    }
                }
            }
            return response;
        }
    }