smspushthread.java

短信

package com.asiainfo.batchsend.sms.sms;

import java.util.ArrayList;
import java.util.List;

import org.apache.log4j.Logger;

import com.asiainfo.batchsend.CtrCenEntry;
import com.asiainfo.batchsend.sms.sms.cmpp.CMPP_Service;
import com.asiainfo.batchsend.sms.sms.cmpp.message.SubmitMsg;
import com.asiainfo.batchsend.sms.sms.cmpp.type.CMPP2;
import com.asiainfo.batchsend.sms.sms.cmpp.type.CMPP3;
import com.asiainfo.batchsend.sms.sms.cmpp.type.CMPP_DR;
import com.asiainfo.batchsend.sms.sms.cmpp.type.CMPP_HW;
import com.asiainfo.batchsend.sms.sms.cmpp.type.CMPP_SX;
import com.asiainfo.batchsend.sms.sms.cmpp.type.CMPP_YX;
import com.asiainfo.batchsend.sms.sms.db.SmsLoadSvc;
import com.asiainfo.batchsend.util.SmsUtil;

/**
 * <p>Title: 短信网关发送线程</p>
 * <p>Description: 负责发送数据</p>
 * <p>Company: Asiainfo Technologies(China),Inc </p>
 * <p>Date: Dec 19, 2007 5:44:19 PM </p>
 * <p>Email to: [e-mail] </p>
 * @author jiangyl
 * @version 1.0  
 */
public class SmsPushThread extends Thread {
	private static final Logger log = Logger.getLogger(SmsPushThread.class);
	private String channelId ;
	private CMPP_Service cmpp = null;
	private SmsChannelConfig config = null; 
	private int verson = 0;
	private boolean safeExit = false;
	    
	public void safeExit() {
		this.safeExit = true;
	}

	public SmsPushThread(String channelId) {
		super("SMS-SmsPushThread");
		this.channelId = channelId;
	    //从配置文件中获取 网关类型
		config = SmsUtil.getSmsConfig(channelId);
		String gatewaytype = config.getGatewaytype().trim();
		if(gatewaytype.equals("CMPP2")){
			this.cmpp = new CMPP2();
			this.verson = 0x20;
		} else if(gatewaytype.equals("CMPP3")){
			this.cmpp = new CMPP3();
			this.verson = 0x30;
		} else if(gatewaytype.equals("DR")){
			this.cmpp = new CMPP_DR();
			this.verson = 0x12;
		} else if(gatewaytype.equals("YX")){
			this.cmpp = new CMPP_YX();
			this.verson = 0x30;
		} else if(gatewaytype.equals("HW")){
			this.cmpp = new CMPP_HW();
			this.verson = 0x30;
		} else if(gatewaytype.equals("SX")){
			this.cmpp = new CMPP_SX();
			this.verson = 0x30;
		}
	}

	public void run() {
		String taskId = null;
		SmsLoadSvc loader = new SmsLoadSvc(this.channelId);
		while ( !safeExit ){
			taskId = CtrCenEntry.getInstance().getTaskIdByChannel(channelId);
			if ( taskId == null ){
				threadSleep(5000);
				continue;
			}
			try{	
				int stat = this.cmpp.initCMPP(config.getISMG_ADDR(), config.getICP_ID(),config.getICP_ID(),	config.getICP_PASS(), "", this.verson, config.getISMG_PORT(), "MT", config.getChannelId());
				log.info("initCMPP status = " + stat );
				if (stat==0) {
					log.info("以" + config.getICP_ID() + "/" + config.getICP_PASS() + "建立到短信网关(" + config.getISMG_ADDR() + ":" + config.getISMG_PORT()	+ ")的连接成功.");
					//jiangyl 修改成为长链接的方式
					/*while  ( !safeExit  && taskId != null ) {
						doStart(taskId,loader);
						taskId = CtrCenEntry.getInstance().getTaskIdByChannel(channelId);
					}*/
					while  ( !safeExit ) {
						doStart(taskId,loader);
						taskId = CtrCenEntry.getInstance().getTaskIdByChannel(channelId);
					}
				}else{
					log.info("以" + config.getICP_ID() + "/" + config.getICP_PASS() + "建立到短信网关(" + config.getISMG_ADDR() + ":" + config.getISMG_PORT()	+ ")的连接不成功,请联系系统管理员.");
				}
			}catch(Exception e){
				log.error("发送数据时发送严重的错误",e);
			}finally{
				try{
					this.cmpp.quit();
				} catch(Exception e){
				}	
			}
		}
	}

	/**
	 * 处理包括数据库操作和短信发送的整个流程方法,一次处理100条数据库记录
	 * 
	 * @throws Exception
	 * 
	 */
	public void doStart(String taskId,SmsLoadSvc loader) throws Exception {
		if ( taskId == null )
			return;
		loader.setTaskId(taskId);
		List messages = loader.loadData();
		List errors = new ArrayList();
		for (int i = 0; i < messages.size(); i++) {
			MessageInfo mbi = (MessageInfo) messages.get(i);
			// 创建消息体并发送短信
			try {				    
				SubmitMsg msg = new SubmitMsg();
				this.new_cmpp_submit(msg, mbi.getPhonenumber(), mbi.getMessage());
				long be = System.currentTimeMillis();
				int stat = this.cmpp.submit(msg);
				log.info("submit status = " + stat +" , time = " + (System.currentTimeMillis() - be) + "ms");
				threadSleep(15);
				if(stat!=0){
					errors.add(mbi);
				}
			} catch (Exception ex) {
				errors.add(mbi);// 当消息发送发现错误的时候
				threadSleep(5*1000);
			}
		}	
		if(errors.size()>0){
			loader.failDone(errors,config.getSenduplimit());
		}
	}

	/**
	 * 创建短信发送消息包内容
	 * 
	 * @param cmppe_submit1
	 * @param msgScrAddr
	 * @param msgContent
	 * @throws Exception
	 */
	private void new_cmpp_submit(SubmitMsg msg, String msisdn,
			String msgContent) throws Exception {
		msg.nFeeUserType = 0;
		msg.nMsgFormat = 15;
		msg.nDestUsrTl =1;
		msg.nMsgLength = msgContent.length();
		msg.nNeedReply = 0;
		msg.sFeeCode = config.getINFO_FEE();
		msg.sFeeMobile = null;
		msg.sDestTerminalId = msisdn;
		msg.sMsgContent = msgContent;
		msg.sMsgSrc = config.getMSG_SCR_ADDR();
		msg.sSrcId = config.getSERVICE_TYPE();
		msg.sFeeType = config.getFEE_TYPE();
		msg.sServiceId = config.getSERVICE_TYPE();
		msg.sValidTime = config.getMSG_VALIDATE();		
		msg.picAndRing = null;
		 
	}

	public void destroy() {
		this.interrupt();
		try {
			this.join(500);
		} catch (Exception e) {
		}
	}
	
	public static void threadSleep(long millis) {
		try {
			Thread.sleep(millis);
		} catch (InterruptedException e) {
		}
	}   

}