moprocessor.java

国内很牛的软件公司花费两年半开发的用EJB3开发的代码,采用STRUTS和EJB3,目前系统进行第二版.所以拿出来共享

package com.ufmobile.platform.mstreet.processor;

import java.math.BigDecimal;
import java.sql.CallableStatement;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.Hashtable;
import java.util.List;
import java.util.ResourceBundle;

import javax.naming.NamingException;
import javax.persistence.EntityManager;
import javax.persistence.Query;
import javax.sql.DataSource;

import oracle.jdbc.driver.OracleTypes;

import com.ufmobile.common.param.PublicParam;
import com.ufmobile.mstreet.sms.command.MOCommandFactroy;
import com.ufmobile.mstreet.sms.command.StreetMOCommand;
import com.ufmobile.platform.InitListener;
import com.ufmobile.platform.Exception.BusinessException;
import com.ufmobile.platform.config.PlatFormConfig;
import com.ufmobile.platform.config.PlatFormSecrecy;
import com.ufmobile.platform.ejb.BaseFunction;
import com.ufmobile.platform.ejb.EJBBoClient;
import com.ufmobile.platform.entity.PlatformConfigEntity;
import com.ufmobile.platform.log.RunTimeLogger;
import com.ufmobile.platform.mstreet.sms.ISmsMOCommandInterface;
import com.ufmobile.platform.mstreet.sms.MOCommandInfo;
import com.ufmobile.platform.mstreet.sms.MOInfo;

public class MOProcessor{	
	static int max_thread_count = 10;
	static String SMS_SEARCH_TEMPLATE = "{0} {1}";
	static final String TABLE_NAME = "TB_SMS_MODELIVER";
	static final String COLUMN_PK_NAME    = "ID";
	static final String COLUMN_DISPATCHEDFLAG = "DISPATCHEDFLAG";
	static final String COLUMN_DISPATCHEDTIME = "DISPATCHEDTIME";
	static final String MO_MAX_THREAD_COUNT = "MO_MAX_THREAD_COUNT";
	static final String KEY_SMS_SEARCH_TEMPLATE = "SMS_SEARCH_TEMPLATE";
	static final int DISPATCHED_FLAG_OK = 1;
	static final int DISPATCHED_FLAG_ERROR = -1;
	static final int UNDISPATCHED_FLAG = 0;
	private static String KEY_MSG_NOTFOUND_PORT = "MSG_NOTFOUND_PORT";
	private static String MSG_NOTFOUND_PORT = "特服号码不存在或者已经过期";
	private static String KEY_MSG_NO_MESSAGE = "MSG_NO_MESSAGE";
	private static String MSG_NO_MESSAGE = "请输入内容进行搜索";	
	private static String KEY_MSG_ERROR_MESSAGE = "MSG_ERROR_MESSAGE";
	private static String MSG_ERROR_MESSAGE = "请输入更准确的关键字进行搜索";
	private static String KEY_MSG_NOTFOUND_BOOTHID = "MSG_NOTFOUND_BOOTHID";
	private static String MSG_NOTFOUND_BOOTHID = "无效的序号，请回复序号访问短信门户，或N看下一条";
	private static String KEY_MSG_NOTFOUND_SERIALNUM = "MSG_NOTFOUND_SERIALNUM";
	private static String MSG_NOTFOUND_SERIALNUM = "请回复序号访问短信门户，或N看下一条";
	private static String KEY_MSG_NO_RESULT = "MSG_NO_RESULT";
	private static String MSG_NO_RESULT = "找不到和您的查询[{0}]相符的商铺，请改用较常见的字词查询";
	private static String KEY_MSG_KEY_W_WAPPUSH = "MSG_KEY_W_WAPPUSH";
	private static String MSG_KEY_W_WAPPUSH = "点击链接访问移动商街";
	private static String KEY_MSG_KEY_MSG_TOO_LENGTH = "MSG_KEY_MSG_TOO_LENGTH";
	private static String MSG_KEY_MSG_TOO_LENGTH = "您搜索的内容过长，请使用简洁的关键词";
	
	private ProcessorPool pool;
	private long minID;
	private int localId;
//	private int count = 0;
//	synchronized public void addCount(int increment){
//		count += increment;
//	}
//	synchronized protected void minusCount(){
//		if(count > 0){
//			count--;
//		}		
//	}
//	public int getCount(){
//		return count;
//	}
	
	MOProcessor(int localId){
		this.localId = localId;
		try{
			ResourceBundle bundle = ProcessorFactory.getResource(); 
			String s = bundle.getString(MO_MAX_THREAD_COUNT);		
			if(s != null && s.compareTo("") != 0){
				max_thread_count = Integer.parseInt(s);
			}
			s = bundle.getString(KEY_SMS_SEARCH_TEMPLATE);
			if(s != null && s.compareTo("") != 0){
				SMS_SEARCH_TEMPLATE = s;
			}
			s = bundle.getString(KEY_MSG_NOTFOUND_PORT);
			if(s != null && s.compareTo("") != 0){
				MSG_NOTFOUND_PORT = s;
			}
			s = bundle.getString(KEY_MSG_NO_MESSAGE);
			if(s != null && s.compareTo("") != 0){
				MSG_NO_MESSAGE = s;
			}
			s = bundle.getString(KEY_MSG_ERROR_MESSAGE);
			if(s != null && s.compareTo("") != 0){
				MSG_ERROR_MESSAGE = s;
			}
			s = bundle.getString(KEY_MSG_NOTFOUND_BOOTHID);
			if(s != null && s.compareTo("") != 0){
				MSG_NOTFOUND_BOOTHID = s;
			}
			s = bundle.getString(KEY_MSG_NOTFOUND_SERIALNUM);
			if(s != null && s.compareTo("") != 0){
				MSG_NOTFOUND_SERIALNUM = s;
			}
			s = bundle.getString(KEY_MSG_NO_RESULT);
			if(s != null && s.compareTo("") != 0){
				MSG_NO_RESULT = s;
			}
			s = bundle.getString(KEY_MSG_KEY_W_WAPPUSH);			
			if(s != null && s.compareTo("") != 0){
				MSG_KEY_W_WAPPUSH = s;
			}
			s = bundle.getString(KEY_MSG_KEY_MSG_TOO_LENGTH);
			if(s != null && s.compareTo("") != 0){
				MSG_KEY_MSG_TOO_LENGTH = s;
			}
		}
		catch(Exception e){
			RunTimeLogger.error(this, "get MO_MAX_THREAD_COUNT from config.properties ", e);
		}
	}	
	
	void run() {
		if(pool == null){
			pool = new ProcessorPool(localId, this, max_thread_count);			
			pool.start();
		}
//		else{
//			pool.activeByHand();
//		}
	}
	
	static boolean process(int localId, MOProcessor moProcessor, long id){
		boolean ret = true;
		EntityManager manager = ProcessorFactory.getEntityManager(localId);
		Query query = manager.createQuery(" from MOInfo a where a.ID = ?1");
		query.setParameter(1, new Long(id).intValue());
		List<MOInfo> list = query.getResultList();
		MOInfo info = null;
		if(list != null && list.size() > 0){
			info = MOInfo.cloneMo(list.get(0));
			if(info.getMSGCONTENT() != null){
				info.setMSGCONTENT(info.getMSGCONTENT().trim());
			}
		}
		if(info != null ){
			if(info.getDISPATCHEDFLAG() == UNDISPATCHED_FLAG){
				int result = 0;
				try{
					result = process(localId, info);					
				}
				catch(Exception e){
					result = 0;					
					RunTimeLogger.error(MOProcessor.class, "处理MO", e);
				}
				manager.getTransaction().begin();
				query = manager.createNativeQuery(" update TB_SMS_MODELIVER  set DISPATCHEDFLAG = :FLAG, DISPATCHEDTIME = :TS where ID = :ID");
				query.setParameter("ID", id);
				if(result == 1){
					query.setParameter("FLAG", DISPATCHED_FLAG_OK);
				}
				else{
					query.setParameter("FLAG", DISPATCHED_FLAG_ERROR);
				}
				query.setParameter("TS", new Date());
				query.executeUpdate();
				manager.getTransaction().commit();
//				finally{
//					moProcessor.minusCount();
//				}	
			}
		}		
		else{
			ret = false;
		}
		return ret;
	}
	
	synchronized long getMinID(int localId) {
		this.minID = 0;
		Query query = ProcessorFactory.getEntityManager(localId).createNativeQuery(" select min("+ COLUMN_PK_NAME +") from " + TABLE_NAME + 
				" where "+ COLUMN_DISPATCHEDFLAG +" = ?1");
		query.setParameter(1, UNDISPATCHED_FLAG);
		Object o = query.getSingleResult();
		if(o != null){
			this.minID = ((BigDecimal)o).longValue();
		}
		return this.minID;
	}
	private static int process(int localId, MOInfo info){		
		int ret = 0;
		DataSource ds = null;
		Connection conn = null;
		try{
			ds = BaseFunction.getDataSource2(localId);
			conn = ds.getConnection();			
			String port = info.getPORT();
			if(port != null && port.compareTo("") != 0){
				StringBuffer boothIdBuff = new StringBuffer();
				int portType = getPortType(conn, port, boothIdBuff);
				switch(portType){
					case -1:	//搜索
						ret = search(localId, conn, info);
						break;
					case 1:		//商铺专用						
						if(boothIdBuff.length() == 0){
							ret = search(localId, conn, info);
						}
						else{
							ret = shopProcess(localId, conn, info, boothIdBuff);
						}
						break;
					case 4:
						ret = OperatorProcess(localId, conn, info, boothIdBuff);
						break;
					case 2:		//共享专用
						ret = userShareProcess(localId, conn, info);
						break;
					case 3:    //群发
						ret = multiSend(localId, conn, info);
						break;
					default:
						//非法的特服号，发送MOMT
						createNotPortMOMT(localId, conn, MSG_NOTFOUND_PORT, info);
						ret = 1;
						break;
				}
			}
			else{
				if(info.getMSGCONTENT() == null || info.getMSGCONTENT().trim().compareTo("") == 0){
					createNotPortMOMT(localId, conn, MSG_NO_MESSAGE, info);
					ret = 1;
				}
				else{
					ret = search(localId, conn, info);
				}
				
			}
		}
		catch(SQLException sqlE){
			RunTimeLogger.error(MOProcessor.class, "处理MO", sqlE);
		} catch (NamingException ne) {
			RunTimeLogger.error(MOProcessor.class, "处理MO", ne);
		} catch (BusinessException be) {
		}
		catch(RuntimeException re){
			RunTimeLogger.error(MOProcessor.class, "处理MO", re);			
		}	
		finally{			
			if(conn != null){
				try {
					conn.close();
				} catch (SQLException e) {
					RunTimeLogger.error(MOProcessor.class, "conn.close()", e);
				}
				conn = null;
			}
		}		
		return ret;
	}
	
	private static int getPortType(Connection conn, String port, StringBuffer boothId) throws SQLException{
		//取特服号类型信息(PortType)
		//涉及结构(业务系统-商街库)TB_PORT:port; porttype(搜索专用-1，商铺专用1，共享专用2);
	    //deadline; deleteflag(1删除);localstate（1启用）
		int ret = 0;
		PreparedStatement ps = conn.prepareStatement("select porttype, BOOTHID, deadline from TB_PORT" +
				" where port = ? and deleteflag <> ? and localstate=?");