使用“阿里大于”和 “企信通” 短信异步发送

         因为项目里有许多发送短信的地方，所以我们就把 发送短信的功能单独拿出来，做成一个service给其他app使用。

　　　　程序的大概流程是这样的 ：

                                       

 首先 我选择生产者模式创建队列。

@Component
public class SmsQueueFactory {

    private static SmsQueueFactory smsQueueFactory =null;

    static final BlockingQueue<SmsMessageDto> messageQueue =new ArrayBlockingQueue<SmsMessageDto>(1024);

    private SmsQueueFactory (){};

    private static Logger log = LoggerFactory.getLogger(SmsQueueFactory.class);

    /**
     * 获取对象
     * @return
     */
    public static  SmsQueueFactory getSmsQueueFactory() {
        if (smsQueueFactory == null) {
            smsQueueFactory = new SmsQueueFactory();
        }
        return smsQueueFactory;
    }

    public BlockingQueue<SmsMessageDto> getMessageQueue(){
        return this.messageQueue;
    }


    public void setMessageDto(SmsMessageDto messageDto) {
        try {
            this.messageQueue.offer(messageDto, 100, TimeUnit.MILLISECONDS);
        }catch (Exception e){
            log.error("setMessageDto出现错误"+e.getMessage());
        }

    }



}

一个接收器来 接收请求，并保存到数据库, 我在这里只是 写了一些伪代码，关键的地方我也会写上去的：

@Service("smsAcceptorImpl")
public class SmsAcceptorImpl {

public void sendSms(SendMessageReq req) throws Exception{
    validation(req); //校验参数
    SmsMessageDto sms  = genSmsObj(req);//封装成 message 对象
    if (sms==null){
        return;
    }
    sms= saveMessage(sms); // 先保存到数据库这样可以拿到 这个消息对象的 ID，方便以后使用。 
    sendToQueue(sms);// 放到队列
}

/**
 * 添加到发送队列
 * @param
 */
private  void  sendToQueue(SmsMessageDto messageDto) {
    try {
        BlockingQueue<SmsMessageDto> queue =SmsQueueFactory.getSmsQueueFactory().getMessageQueue();
        queue.offer(messageDto, 100, TimeUnit.MILLISECONDS);
    }catch (Exception e){
    log.error("加入队列出错"+e.getMessage());
    }
}

}

然后写一个 init() 方法 ，定时到 队列中 看看，有没有可以发送的消息。

@Component
public class SmsQueueProcess {


    @Autowired
    private SmsSupplierSelector smsSupplierSelector;

    private static Logger log = LoggerFactory.getLogger(SmsQueueProcess.class);

    @PostConstruct
    public void init(){
        getMesForQueue();
    }

    //短信量不大 暂时不使用多线程处理
    private void process(SmsMessageDto sms){
        ISupplier supplier = smsSupplierSelector.getSupplier(sms);
        supplier.validation(sms); // 验证的方法就不写了
        supplier.send(sms); // 重点 把 这发送的代码展示一下
    }


    public void getMesForQueue() {
        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(1);
        fixedThreadPool.execute(new getQueueList());
    }

    class getQueueList implements Runnable{
        //todo 获取 队列中的message 对象
        public void getQueueList() throws InterruptedException{
            while(true){
                BlockingQueue<SmsMessageDto> queue = SmsQueueFactory.getSmsQueueFactory().getMessageQueue();
                    SmsMessageDto smsMessage=queue.take();
                    process(smsMessage);
            }
        }

        @Override
        public void run() {
            try{
                getQueueList();
            }catch (Exception e){
                log.error("获取队列 中的对象出错"+e.getMessage());
            }
        }
    }

//  这个类是 运营商选择器，默认是使用 阿里大于，如果发送失败了，就使用其他的(企信通),目前就只选用了两个运营商
@Component
public class SmsSupplierSelector {

    @Autowired
    private ISupplier smsSupplierAliYu ;
    @Autowired
    private ISupplier smsSupplierQxt ;

    private final Map<String ,ISupplier> suppliers = new HashMap<>();


    @PostConstruct
    private void init(){
        suppliers.put("alidayu",smsSupplierAliYu);
        suppliers.put("qxt",smsSupplierQxt);
    }

    public ISupplier getSupplier(SmsMessageDto sms) {

        if(StringUtil.isBlank(sms.getSupplier())){
            sms.setSupplier("alidayu");
            return   smsSupplierAliYu;
        }
        HashMap<String ,ISupplier> cloneSuppliers = new HashMap<>(suppliers);

        cloneSuppliers.remove(sms.getSupplier());

        String[] keys = cloneSuppliers.keySet().toArray(new String[0]);
        Random random = new Random();
        String randomKey = keys[random.nextInt(keys.length)];
        sms.setSupplier(randomKey);
        return cloneSuppliers.get(randomKey);
    }
}

运营商选择器，返回的是运营商对象。两个运营商，每个运营商都有自己的验证和发送的方法，这里展示一下 ，发送的方法。

@Override
public void send(SmsMessageDto sms) {
    //todo 使用企信通的发送
    try {
     String message=sms.getMessage();
     String   params = "DesMobile="+sms.getPhone()+"&Content=" + URLEncoder.encode(message, "GBK")
                +"&t="+ new Date().getTime();
    String logMes = HttpClientUtil.doGet(GET_URL + params);
    sms.setSendCount(sms.getSendCount()==null ?0:(sms.getSendCount()+ 1)); //发送的次数加1
    String code = XmlUtil.getContentFromXml(logMes, "code");
        if (!("01".equals(code) || "03".equals(code))) {
            sms.setStatus(2);   //发送失败
        }else{
            sms.setStatus(3); //发送成功
        }
    } catch (Exception e) {
        log.error("短信发送失败！方法名[send]"+e.getMessage());
    }finally {
        SmsMessage smsMessage=new SmsMessage();
        smsMessage.setId(sms.getId());
        smsMessage.setStatus(sms.getStatus());
        smsMessage.setSendCount(sms.getSendCount());
        smsMessage.setUpdateTime(new Date());
        smsMessage.setSupplier(sms.getSupplier()); 
        smsMessage.setRequestJson(JsonUtil.bean2Json(sms));
        messageService.updateStatus(smsMessage);

    }
}　

 @Override
    public void send(SmsMessageDto sms)  {
        //todo  使用阿里大于发送

        TaobaoClient client = new DefaultTaobaoClient("http://gw.api.taobao.com/router/rest", "app-key", "app-secret");
        AlibabaAliqinFcSmsNumSendRequest req = new AlibabaAliqinFcSmsNumSendRequest();
        req.setExtend(sms.getId()+"sms");
        req.setSmsType("normal");
        Map<String,String > stringStringMap=sms.getSmsParamMap();
        req.setSmsFreeSignName(stringStringMap.get("product"));
        req.setSmsParamString(JsonUtil.bean2Json(stringStringMap));
        req.setRecNum(sms.getPhone() );
        sms.setSendCount(sms.getSendCount()==null ?0:(sms.getSendCount()+ 1));
        req.setSmsTemplateCode(sms.getTemplateCode());
        try {
            client.execute(req);
            sms.setSendCount(sms.getSendCount()==null ?(1):(sms.getSendCount()+ 1));
            sms.setStatus(1);

            SmsMessage smsMessage=new SmsMessage();
            smsMessage.setStatus(sms.getStatus()); // 发送状态
            smsMessage.setId(sms.getId());
            smsMessage.setUpdateTime(new Date());
            smsMessage.setSendCount(sms.getSendCount());
            smsMessage.setSupplier(sms.getSupplier());
            smsMessage.setRequestJson(JsonUtil.bean2Json(sms));
            messageService.updateStatus(smsMessage);

        } catch (ApiException e) {
            log.error("阿里大于发送消息失败,类名[SmsSupplierAliYu]"+e.getMessage());
        }
    }

企信通的发送是实时接收消息的状态，阿里大鱼是异步的，所以就需要有个守护线程，查询数据库，和阿里服务器，修改消息的状态。

 @Scheduled(cron="0/10 * * * * ? ")
    // 获取消息的返回参数
    public void ckMessageStatusTask() throws  Exception {

        try {
            //todo 守护线程 ，定时查询数据库，未发送成功的加入队列，重新发送
            BlockingQueue<SmsMessageDto> queue = SmsQueueFactory.getSmsQueueFactory().getMessageQueue();
            List<SmsMessage> smsMessageList = messageService.selectByStatusAndSendCount();// 要按照时间排序
            List<SmsMessage> listAl =new ArrayList<>();
            for (SmsMessage smsMessage : smsMessageList) {
                if (aLiDaYu.equals(smsMessage.getSupplier())){
                    listAl.add(smsMessage);
                }
            }
            if (null !=listAl && listAl.size()>0){
                getMessageStatusForAli(listAl);
            }
            smsMessageList.removeAll(listAl);
            for (int j = 0; j < smsMessageList.size(); j++) {
                SmsMessage smsMessage=smsMessageList.get(j);
                if (smsMessage.getStatus()==2 || smsMessage.getStatus()==0){ //
                    SmsMessageDto smsMessageDto = JsonUtil.json2Bean(smsMessage.getRequestJson(), SmsMessageDto.class);
                    queue.offer(smsMessageDto, 100, TimeUnit.MILLISECONDS);
                }
            }
        } catch (Exception e) {
            log.error("短信守护线程出错，方法名[ckMessageStatusTask]" + e.getMessage());
        }
    }

    private void  getMessageStatusForAli( List<SmsMessage> smsMessageList){

        //todo 从阿里大鱼 查询消息的状态
        BlockingQueue<SmsMessageDto> queue = SmsQueueFactory.getSmsQueueFactory().getMessageQueue();
　　　　　　// 这里需要在阿里大于 网站上注册一个用户，并创建应用。使用上面的app证书 创建 client
        TaobaoClient client = new DefaultTaobaoClient("http://gw.api.taobao.com/router/rest", "app-key", "app-secret");
        AlibabaAliqinFcSmsNumQueryRequest req = new AlibabaAliqinFcSmsNumQueryRequest();
        SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
        Date date = new Date();
        try {
            for (SmsMessage smsMessage:smsMessageList){
                String extend=smsMessage.getId()+"sms";
                req.setRecNum(smsMessage.getPhone());
                req.setQueryDate(sdf.format(date));
                req.setCurrentPage(1L); // 当前页码
                req.setPageSize(10L);
                AlibabaAliqinFcSmsNumQueryResponse rsp = client.execute(req);
                List<AlibabaAliqinFcSmsNumQueryResponse.FcPartnerSmsDetailDto> smsDetailDtos=rsp.getValues();
                if (null !=smsDetailDtos){
                    for (AlibabaAliqinFcSmsNumQueryResponse.FcPartnerSmsDetailDto fcPartnerSmsDetailDto : smsDetailDtos) { // 只有一个 消息
                        if (extend.equals( fcPartnerSmsDetailDto.getExtend())) {
                            Integer state = Integer.parseInt(Long.toString(fcPartnerSmsDetailDto.getSmsStatus()));
                            if (state==3)
                                smsMessageList.remove(smsMessage);
                            if (state==2){
                                SmsMessageDto smsMessageDto = JsonUtil.json2Bean(smsMessage.getRequestJson(), SmsMessageDto.class);
                                queue.offer(smsMessageDto, 100, TimeUnit.MILLISECONDS);
                            }
                            smsMessage.setStatus(state);
                            messageService.updateMessageStatusById(smsMessage);
                            // 添加日志
                            SmsLog smsLog=new SmsLog();
                            smsLog.setCreateTime(new Date());
                            smsLog.setMessageId(smsMessage.getId());
                            smsLog.setMessageDetails(smsMessage.getMessage());
                            smsLog.setMessageType("(ali)修改记录");
                            smsLog.setMessageStatus(smsMessage.getStatus());
                            smsLog.setMessageSupplier("ali");
                            messageLogService.insertSelective(smsLog);
                        }
                    }
                }
            }
        }catch (Exception e){
            log.error("查询消息状态出错，方法名[getMessageStatusForAli]"+e.getMessage());
        }
    }

基本上这么多，就已经把这个服务的代码都展示出来了。