多线程的使用1

今天药忘吃喽~ 2022-08-31 00:50 150阅读 0赞

> 第一种引用

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM4MDY4ODEy_size_16_color_FFFFFF_t_70]

package com.lufax.wmc.ext.service.impl;
    
    import com.dianping.cat.Cat;
    import com.google.common.collect.Lists;
    import com.google.gson.Gson;
    import com.google.gson.GsonBuilder;
    import com.lufax.kernel.cache.CacheException;
    import com.lufax.wmc.ext.common.cache.RedisCache;
    import com.lufax.wmc.ext.dao.SmsRequestDetailsDAO;
    import com.lufax.wmc.ext.dao.SmsTemplateDAO;
    import com.lufax.wmc.ext.dto.*;
    import com.lufax.wmc.ext.enums.*;
    import com.lufax.wmc.ext.service.msgSend.MsgTaskRun.JobConstant;
    import com.lufax.wmc.ext.service.msgSend.contextBuilder.context.sms.SmsContext;
    import com.lufax.wmc.ext.service.msgSend.handler.sender.SmsSender;
    import com.lufax.wmc.ext.service.properties.ExtProperties;
    import com.lufax.wmc.ext.utils.*;
    import org.apache.commons.collections.CollectionUtils;
    import org.apache.commons.validator.GenericValidator;
    import org.joda.time.DateTime;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    
    import java.util.*;
    
    /** * @Author dongwenkai960 * @Date 19-6-4 上午10:03 */
    public class SmsRedisJobService { 
    
        @Autowired
        private RedisCache redisCache;
        @Autowired
        private ThreadPoolTaskExecutor handleSendRedisMktSmsThreadPoolTaskExecutor;
        @Autowired
        private ExtProperties properties;
        @Autowired
        private SmsRequestDetailsDAO smsRequestDetailsDao;
        @Autowired
        private SmsTemplateDAO smsTemplateDao;
        @Autowired
        private SmsSender smsSender;
    
        public Long handleSmsRequestDetail() throws CacheException { 
            String key = RedisKeyPrefixEnum.SMS_REQUEST_DETIAL + ":" + "REQUEST_DETAIL_LIST";
            Long listLen = redisCache.listLen(key);
            Logger.debug(this, String.format("handleSmsRequestDetail============reids:REQUEST_DETAIL_LIST length = %s", listLen));
            if (listLen <= 0) { 
                return 0L;
            }
            int maxExtSmsThreadPoolSize = properties.getExtSmsThreadPoolMaxsize();
            int activeExtSmsThreadPoolSize = handleSendRedisMktSmsThreadPoolTaskExecutor.getActiveCount();
            int remainExtSmsThreadPoolSize = maxExtSmsThreadPoolSize - activeExtSmsThreadPoolSize;
            Logger.debug(this, String.format("子线程读取redis队列中数据, 当前可用线程数=%s", remainExtSmsThreadPoolSize));
            if (remainExtSmsThreadPoolSize > 0) { 
                for (int i = 0; i < remainExtSmsThreadPoolSize; i++) { 
                    handleSendRedisMktSmsThreadPoolTaskExecutor.execute(new Runnable() { 
                        @Override
                        public void run() { 
                            runHandleSmsRequestDetail(properties.getExtRedisBlockingTimeoutSeconds());
                        }
                    });
                }
            }
            return listLen;
        }
    
        private void runHandleSmsRequestDetail(int blockingTimeout){ 
            Logger.debug(this, "runHandleSmsRequestDetail ======== start");
    
            try{ 
                String key = RedisKeyPrefixEnum.SMS_REQUEST_DETIAL+":"+"REQUEST_DETAIL_LIST";
                while (!JobConstant.isFinished && redisCache.listLen(key) > 0){ 
                    String redisSmsValue = redisCache.listBlockingRPop(key,blockingTimeout);
                    Logger.debug(this,String.format("runHandleSmsRequestDetail线程读取redis数据 redisSmsValue=%s",redisSmsValue));
                    Gson gson = new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss").create();
                    SmsRequestDetailCacheDTO smsRequestDetailCacheDTO = gson.fromJson(redisSmsValue,SmsRequestDetailCacheDTO.class);
    
                    if(smsRequestDetailCacheDTO == null){ 
                        Logger.debug(this, "runHandleSmsRequestDetail ======== redis队列中无待发送数据");
                        return;
                    }
                    handleSmsRequest(smsRequestDetailCacheDTO);
                }
            }catch (Exception e) { 
                Logger.error(this, String.format("runHandleSmsRequestDetail failed, message=%s", e.getMessage()), e);
            }
            Logger.debug(this, "runHandleSmsRequestDetail ======== end");
        }
    
        private void handleSmsRequest(SmsRequestDetailCacheDTO smsRequestDetailCacheDTO){ 
            List<SmsRequestDetailMessagesCacheDTO> messages = smsRequestDetailCacheDTO.getMessages();
            if (CollectionUtils.isEmpty(messages)) { 
                return;
            }
            StringBuffer sbMobileNos = new StringBuffer();
            List<SmsRequestDetailMessagesCacheDTO> correctMessageList = new ArrayList<SmsRequestDetailMessagesCacheDTO>();
            for (SmsRequestDetailMessagesCacheDTO messagesCacheDTO : messages) { 
                if (sbMobileNos.toString().contains(messagesCacheDTO.getMobileNo())) { 
                    Logger.warn(this, String.format("start to deal with the repeated mobile no, the repeated no is [%s]!", HelperUtils.markMobileNos(messagesCacheDTO.getMobileNo())));
                    String errorMessage = "the mobile number is repeated";
                    SmsRequestDetail smsRequestDetail = new SmsRequestDetail();
                    smsRequestDetail.setId(messagesCacheDTO.getId());
                    smsRequestDetail.setStatus(SmsRequestDetailsStatus.FAILED.id());
                    smsRequestDetail.setSendDate(new Date());
                    smsRequestDetail.setErrorMessage(errorMessage);
                    smsRequestDetailsDao.updateSmsRequestDetails(smsRequestDetail);
                } else { 
                    sbMobileNos = sbMobileNos.append(messagesCacheDTO.getMobileNo()).append(properties.getMobileNoSeparator());
                    correctMessageList.add(messagesCacheDTO);
                }
            }
    
            SmsTemplate smsTemplate = smsTemplateDao.getSmsTemplateById(Long.parseLong(smsRequestDetailCacheDTO.getTemplateId()));
            //1.校验过期日期
            //2.校验模板
            if (smsTemplate == null || SmsTemplateStatus.INACTIVE.id() == smsTemplate.getStatus()) { 
                String errorMessage = "can not find a sms template or the template is inactive";
                for (SmsRequestDetailMessagesCacheDTO correctMessage : correctMessageList) { 
                    SmsRequestDetail smsRequestDetail = new SmsRequestDetail();
                    smsRequestDetail.setId(correctMessage.getId());
                    smsRequestDetail.setStatus(SmsRequestDetailsStatus.FAILED.id());
                    if (smsTemplate == null) { 
                        smsRequestDetail.setFailReason(FilterType.TEMPLATE_NOT_FOUND.getValue());
                    } else { 
                        smsRequestDetail.setFailReason(FilterType.INACTIVE.getValue());
                    }
                    smsRequestDetail.setErrorMessage(errorMessage);
                    smsRequestDetailsDao.updateSmsRequestDetails(smsRequestDetail);
                }
                return;
            }
    
            //设置smsContext
            Map<String, String> smsContentMap = generateSmsContent(smsRequestDetailCacheDTO.getProperties(), smsTemplate);
            if ("FAIL".equals(smsContentMap.get("returnFlag"))) { 
                String errorMessage = String.format("can not find enough properties to replace parameters in template, template_name=%s, smsContent=%s",
                        smsTemplate.getTemplateName(), smsContentMap.get("smsContext"));
                Logger.error(this, errorMessage);
                for (SmsRequestDetailMessagesCacheDTO correctMessage : correctMessageList) { 
                    SmsRequestDetail smsRequestDetail = new SmsRequestDetail();
                    smsRequestDetail.setId(correctMessage.getId());
                    smsRequestDetail.setStatus(SmsRequestDetailsStatus.FAILED.id());
                    smsRequestDetail.setErrorMessage(errorMessage);
                    smsRequestDetailsDao.updateSmsRequestDetails(smsRequestDetail);
                }
                return;
            }
    
            /** * 模板生效时间和过期时间检查: * 模板未生效或者模板过期的，直接设置为发送失败，记录失败的原因和错误消息； */
    
            String errorMessage = null;
            String failReason = null;
            Date currentTime = new Date();
            if (smsTemplate.getEnableAt() != null && currentTime.before(smsTemplate.getEnableAt())) { 
                errorMessage = String.format("Current time[%s] is before the enable time[%s] of template[%s], it's not sent.",
                        DateUtils.formatDateTime(currentTime), DateUtils.formatDateTime(smsTemplate.getEnableAt()), smsTemplate.getTemplateName());
                failReason = FilterType.BEFORE_ENABLE.getValue();
            }
            if (smsTemplate.getExpireAt() != null && currentTime.after(smsTemplate.getExpireAt())) { 
                errorMessage = String.format("Current time[%s] is after the expire time[%s] of template[%s], it's not sent.",
                        DateUtils.formatDateTime(currentTime), DateUtils.formatDateTime(smsTemplate.getExpireAt()), smsTemplate.getTemplateName());
                failReason = FilterType.AFTER_EXPIRE.getValue();
            }
            if (errorMessage != null) { 
                Logger.warn(this, errorMessage);
                for (SmsRequestDetailMessagesCacheDTO correctMessage : correctMessageList) { 
                    SmsRequestDetail smsRequestDetail = new SmsRequestDetail();
                    smsRequestDetail.setId(correctMessage.getId());
                    smsRequestDetail.setStatus(SmsRequestDetailsStatus.FAILED.id());
                    smsRequestDetail.setFailReason(failReason);
                    smsRequestDetail.setErrorMessage(errorMessage);
                    smsRequestDetailsDao.updateSmsRequestDetails(smsRequestDetail);
                }
                return;
            }
    
            // 模板静默其检查
            if (isDuringSleepingTime(smsTemplate, new Date())) { 
                //set the next scheduler date
                Date nextSchedulerDate = getNextSchedulerDate(smsTemplate);
                for (SmsRequestDetailMessagesCacheDTO correctMessage : correctMessageList) { 
                    SmsRequestDetail smsRequestDetail = new SmsRequestDetail();
                    smsRequestDetail.setId(correctMessage.getId());
                    smsRequestDetail.setScheduleDate(nextSchedulerDate);
                    smsRequestDetailsDao.updateSmsRequestDetails(smsRequestDetail);
                }
                return;
            }
    
            SmsType smsType = null;
            if (SmsType.GENERAL.id() == smsTemplate.getSmsType()) { 
                smsType = SmsType.GENERAL;
            } else { 
                smsType = SmsType.MARKETING;
            }
    
            SmsContext smsContext = buildSmsContext(correctMessageList, smsContentMap, smsType, smsTemplate);
            smsSender.handle(smsContext);
        }
    
        private SmsContext buildSmsContext(List<SmsRequestDetailMessagesCacheDTO> correctMessageList, Map<String, String> smsContentMap, SmsType smsType, SmsTemplate smsTemplate){ 
            SmsContext smsContext = new SmsContext();
            SmsRequest smsRequest = new SmsRequest();
            List<SmsRequestDetail> smsRequestDetails = Lists.newArrayList();
            smsContext.setSmsRequest(smsRequest);
            for (SmsRequestDetailMessagesCacheDTO correctMessage : correctMessageList) { 
                SmsRequestDetail smsRequestDetail = new SmsRequestDetail();
                smsRequestDetail.setStatus(SmsRequestDetailsStatus.PROCESSING.id());
                smsRequestDetail.setLocale(correctMessage.getLocale());
                smsRequestDetail.setMobileNo(correctMessage.getMobileNo());
                smsRequestDetail.setMobileCountryCode(correctMessage.getMobileCountryCode());
                smsRequestDetail.setSmsContent(smsContentMap.get("smsContext"));
                smsRequestDetail.setSmsTemplate(smsTemplate);
                smsRequestDetail.setId(correctMessage.getId());
                smsRequestDetails.add(smsRequestDetail);
            }
            return smsContext;
        }
    
        private Map<String, String> generateSmsContent(String properties, SmsTemplate smsTemplate) { 
            Map<String, String> retMap = new HashMap<String, String>();
    
            String smsTemplateContext = smsTemplate.getTemplate();
            Map<String, String> replacementProperties = new HashMap<String, String>();
            if (StringUtils.isNotBlank(properties)) { 
                String[] replacementKeyValues = properties.split("\n");
                for (String s : replacementKeyValues) { 
                    String[] keyValues = { "", ""};
                    if (!s.contains("=")) { 
                        Logger.warn(this, "the properties does not contain the split character =");
                        continue;
                    }
                    keyValues[0] = s.substring(0, s.indexOf("="));
                    keyValues[1] = s.substring(s.indexOf("=") + 1, s.length());
                    String key = "\\[\\$" + keyValues[0].trim() + "\\$\\]";
                    String propertyValue = (keyValues.length == 2) ? keyValues[1].trim() : "";
                    replacementProperties.put(key, propertyValue);
                }
            }
            String smsContext = FileUtils.substitution(replacementProperties, smsTemplateContext).toString();
            //添加cat 打点，记录是否匹配到关键字
            if (org.apache.commons.lang.StringUtils.isNotBlank(smsContext) && smsContext.contains(Constants.MONITOR_KEY_WORD)) { 
                String appName = Cat.getParentDomain();
                Cat.logEvent("EXT.SMS.KEY.WORD.MONITOR", appName + "_" + smsTemplate.getTemplateName());
            }
            retMap.put("smsContext", smsContext);
            // 模板中还有未替换完的参数
            if (GenericValidator.matchRegexp(smsContext, "\\[\\$.*\\$\\]")) { 
                retMap.put("returnFlag", "FAIL");
            } else { 
                retMap.put("returnFlag", "SUCCESS");
            }
            return retMap;
        }
    
        /** * 判断短信模板是否处于静默期中 * * @param smsTemplate 短信模板 * @param now 当前系统时间，方便单测而加的参数 * @return */
        public boolean isDuringSleepingTime(SmsTemplate smsTemplate, Date now) { 
            int start = smsTemplate.getQuietPeriodStart();
            int end = smsTemplate.getQuietPeriodEnd();
            int minutesOfNow = new DateTime(now).getMinuteOfDay();
    
            if (start == end) { 
                // 静默期的开始和结束时间相同时，不做静默期判断
                return false;
            } else if (start > end) { 
                // 跨天: eg. 22~8 start <= now < 0 || 0 <= now < end
                return (start <= minutesOfNow && minutesOfNow < 24 * 60) || (0 <= minutesOfNow && minutesOfNow < end);
            } else { 
                // 不跨天: eg. 0~8 start <= now < end
                return ((start <= minutesOfNow) && (minutesOfNow < end));
            }
        }
    
        private Date getNextSchedulerDate(SmsTemplate smsTemplate) { 
            Date now = new Date();
            Date smsQuietEndTime = null;
            int minutesOfNow = new DateTime(now).getMinuteOfDay();
            if ((0 < minutesOfNow) && (minutesOfNow < smsTemplate.getQuietPeriodEnd())) { 
                smsQuietEndTime = new DateTime(DateUtils.startOfDay(now)).plusMinutes(smsTemplate.getQuietPeriodEnd()).toDate();
            } else { 
                smsQuietEndTime = new DateTime(DateUtils.endOfDay(now)).plusMinutes(smsTemplate.getQuietPeriodEnd()).toDate();
            }
            return smsQuietEndTime;
        }
    }

> 第二种引用

<bean id="asyncLoginEventExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor" destroy-method="shutdown">
            
            <property name="corePoolSize" value="5" />
            
            <property name="keepAliveSeconds" value="30000" />
            
            <property name="maxPoolSize" value="100" />
            
            <property name="queueCapacity" value="5000" />
            
            <property name="rejectedExecutionHandler">
                
                
                
                
                <bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy" />
            </property>
        </bean>

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM4MDY4ODEy_size_16_color_FFFFFF_t_70]: /images/20220829/f11035d031b8461fa3c241f3b224d0bc.png