cmppdbaccess.java

中国移动CMPP3.0短消息通信程序

package com.pansonlu.cmpp;
/**
 * <p>Title:CMPPDBAccess</p>
 * <p>Description: 湖南移动短信网关通讯程序</p>
 * <p>数据库访问类</p>
 * <p>Copyright: Copyright (c) 2008</p>
 * <p>Company: Sunrise tech ltd.</p>
 * @author pansonlu
 * @version 1.0
 */

import java.util.*;
import java.sql.*;

import com.pansonlu.common.database.*;
import com.pansonlu.common.util.*;

public class CMPPDBAccess {

  private static CMPPDBAccess DBAccess;

  public boolean isAvail = false;

  /** 上行数据缓冲,当CMPPMOReceiveThread接受到上行数据后,将数据保存在该缓冲中,由
   *  CMPPMODataSaveThread线程将数据再保存到数据库中 **/
  public Vector vctMOData = new Vector(1,1);

  /** 下行数据缓冲,当CMPPDBAccess从数据库中取出数据后,保存在该缓冲中,由
   *  单连接时的CMPPMOSendThread线程或双连接时的CMPPMTSendThread线程将数据发送到ISMG **/
  public Vector vctMTData = new Vector(1,1);

  public String inertSql  = "";

  private CMPPMODataSaveThread moSaveThread;

  public CMPPDBAccess() {
    if(CMPParameter.DBType==1) inertSql = this.strInsertOracle;
    else inertSql = this.strInsertSqlServer;
  }


  /**取得CMPPMOSocketProcess的唯一实例 **/
  public static CMPPDBAccess getInstance(){
    if(DBAccess == null){
      DBAccess = new CMPPDBAccess();
    }
    if(!DBAccess.isAvail){
      DBAccess.connect();
    }
    else{

    }
    return DBAccess;
  }




  /**
   * 连接到数据库
   * @return true(成功)or false（失败）
   */
  public boolean connect(){

    Connection conn = null;
    ResultSet   rs  = null;
    try {
//      Class.forName(DBDriver);
//      conn = DriverManager.getConnection(Connect_URL, DBUser, DBPass);
      conn = ConnectionPool.getConnection();
      //从连接池取得连接并测试连接是否成功
      String strSql = "";
      if(CMPParameter.DBType ==1)
        strSql = "SELECT SYSDATE FROM DUAL";
      else if(CMPParameter.DBType ==2)
        strSql = "select getdate()";
      else if(CMPParameter.DBType ==3)
        strSql = "select getdate()";

      rs = conn.createStatement().executeQuery(strSql);
      rs.next();
      rs.getString(1);
      Debug.outInfo("[CMPPMaster]"+PublicFunction.getFormatTime()+" 通讯程序连接数据库成功!");
      rs.close();
      this.isAvail = true;

      moSaveThread = new CMPPMODataSaveThread();
      moSaveThread.start();
      Debug.outInfo("[CMPPMaster]"+PublicFunction.getFormatTime()+" 缓冲数据线程启动...");
    }
//    catch (ClassNotFoundException cnfe) {
//      disconnect();
//      System.out.println("Not Found Database Connect Driver " + cnfe);
//    }
    catch (Exception ex) {
      System.out.println("[CMPPMaster]连接数据库失败DBAccess.connect() 异常：" + ex);
      this.isAvail = false;
    }
    finally{
      discloseconn(conn,null,null);  //关闭connection(放回连接池)
    }

    return this.isAvail;
  }



    String strInsertOracle
        = "insert into cmpp_from_ismg ( from_ismg_id, deliver_msg_id, deliver_destnation_id," +
          "deliver_service_id, deliver_msg_fmt, deliver_src_terminal_id, deliver_registered_delivery,"+
          "deliver_msg_length, deliver_msg_content values(seq_sms.nextval,";
    String strInsertSqlServer
          = "insert into cmpp_from_ismg (deliver_msg_id, deliver_destnation_id," +
            "deliver_service_id, deliver_msg_fmt, deliver_src_terminal_id, deliver_registered_delivery,"+
            "deliver_msg_length, deliver_msg_content values(";
  /**
   * 将vctMOData中的MO数据保存到数据库中
   * @param vctMOData
   * @throws java.lang.Exception
   */

  public void saveMODataFromVector(Vector vctMOData) throws Exception{

    Connection conn = null;
    Statement  stmt = null;

    try{

      conn = ConnectionPool.getConnection();
      stmt = conn.createStatement();
      conn.setAutoCommit(false);

      //将状态报告,Deliver消息插入到数据库的SQL语句
      for(int i=0;i<vctMOData.size();i++){

        CMPP submitMsg = (CMPP)vctMOData.elementAt(i);
        if(submitMsg.Registered_Delivery == 1){
          //当为状态报告时
          String strSql = "insert into cmpp_report (Msg_Id,Stat,Submit_time,Done_time,Dest_terminal_Id,SMSC_sequence) ";
                 strSql+= "values(?,?,?,?,?,?)";
          PreparedStatement pstmt = conn.prepareStatement(strSql);
          pstmt.setLong(1,submitMsg.Sequence_Id);
          pstmt.setString(2,submitMsg.Report_Stat);
          pstmt.setString(3,submitMsg.Report_Submit_time);
          pstmt.setString(4,submitMsg.Report_Done_time);
          pstmt.setString(5,submitMsg.Report_Dest_terminal_Id);
          pstmt.setInt(6,submitMsg.Report_SMSC_sequence);
          pstmt.execute();
          pstmt.close();
        }
        else{
          //当为Deliver消息时
          StringBuffer sbSql = new StringBuffer("");
          sbSql.append("insert into cmpp_from_ismg ( from_ismg_id, deliver_msg_id, deliver_destnation_id,");
          sbSql.append("deliver_service_id, deliver_msg_fmt, deliver_src_terminal_id, deliver_registered_delivery,");
          sbSql.append("deliver_msg_length, deliver_msg_content");
          sbSql.append(") values (seq_sms.nextval,");

          sbSql.append("" +submitMsg.Msg_Id+",");
          sbSql.append("'"+submitMsg.Dest_terminal_Id+"',");
          sbSql.append("'"+submitMsg.Service_Id+"',");
          sbSql.append("" +submitMsg.Msg_Fmt+",");
          sbSql.append("'"+submitMsg.Src_terminal_Id+"',");
          sbSql.append("'"+submitMsg.Registered_Delivery+"',");
          sbSql.append("'"+submitMsg.Msg_Length+"',");
          sbSql.append("'"+submitMsg.Msg_Content+"'");
          sbSql.append(")");
          stmt.addBatch(sbSql.toString());
        }
      }

      stmt.executeBatch();
      conn.commit();
      conn.setAutoCommit(true);
    }
    catch(Exception ex){
      throw ex;
    }
    finally{
      discloseconn(conn,stmt,null);
    }

  }




  /**
   * 将待发送数据放入缓冲
   * @return
   */
  public boolean fectchMTDataToBuffer(){

    String strSql = "";
    if(CMPParameter.DBType ==1)
    strSql = "select * from (select rownum as my_rownum,a.* from( "+
             "SELECT * FROM cmpp_to_ismg WHERE submit_status = 1"+
             ") a where rownum<= 50 )where my_rownum> 0";
    else
    strSql = "select * from cmpp_to_ismg where submit_status = 1";


    ResultSet rs    = null;
    Statement stmt  = null;
    Connection conn = null;

    try {
        //System.out.println("[CMPPMaster]读取待发送信息到缓冲..." + strSql);
        conn = ConnectionPool.getConnection();
        stmt = conn.createStatement();
        rs   = stmt.executeQuery(strSql);
    }
    catch (Exception ex) {
        this.isAvail = false;
        Debug.outWarn("[CMPPMaster]Fetch data from cmpp_to_ismg table failed: " + ex) ;
        return false;
    }

    byte[] msg;
    try {
         int num        = 0;   //取出待发送数据的计数器
         String content = "";
         while( rs.next()) {
            if(num++ > 1000) break;  //每次最多取出1000条记录

            String msgcontent      = rs.getString ("MSG_CONTENT")==null?"":rs.getString ("MSG_CONTENT").trim();

            //如果发送字符长，则截成多段
            int length     = 70;  //标准为70个字符
            int iMsgLenth = msgcontent.length();

//            int iLength = (0 + length > iMsgLenth) ? iMsgLenth - 0 : length;
//            int iBeginIndex = 0;
//            int iEndIndex   = iEndIndex   =
//                (iBeginIndex + iLength > iMsgLenth) ? (iMsgLenth) : (iBeginIndex + iLength);
//
//            for (;iBeginIndex  < iEndIndex;
//                 iBeginIndex = iEndIndex,
//                 iEndIndex   = (iBeginIndex + iLength > iMsgLenth) ?
//                 (iMsgLenth) : (iBeginIndex + iLength)
//                )
//            {
             CMPP mtMsg   = new CMPP(CMPP.CMPP_SUBMIT);
             mtMsg.Msg_Id = rs.getLong("to_ismg_id");
             mtMsg.Service_Id = rs.getString("service_id");
             mtMsg.Pk_total   = (byte)rs.getInt("pk_total");
             mtMsg.Pk_number  = (byte)rs.getInt("pk_number");
             mtMsg.Msg_Fmt    = (byte)rs.getInt("msg_fmt");
             mtMsg.Registered_Delivery = (byte)rs.getInt("registered_delivery");
             mtMsg.Msg_src    = rs.getString("icp_id");
             mtMsg.FeeType    = rs.getString("fee_type");
             mtMsg.FeeCode    = rs.getString("fee_code");
             mtMsg.Fee_terminal_Id  = rs.getString("fee_termainal_id");
             mtMsg.Src_terminal_Id  = rs.getString("src_terminal_id");
             mtMsg.Dest_terminal_Id = rs.getString("dest_terminal_id");

             mtMsg.Msg_Content = msgcontent;
             mtMsg.Msg_Length  = (byte)mtMsg.Msg_Content.getBytes().length;

             Debug.outInfo("\n[CMPPMaster]"+PublicFunction.getFormatTime()+ " 发送数据\n" +
                                "Src_Number:"+mtMsg.Src_terminal_Id+"\nDest_Number:"+mtMsg.Dest_terminal_Id + "\nFee_Type:"+mtMsg.FeeType+"\nFee_Code:"+mtMsg.FeeCode+"\nMsg_Content:"
                                + msgcontent+"\nMsg_Id:"+mtMsg.Msg_Id+"\nLink_Id:"+mtMsg.LinkID+"");
             synchronized(vctMTData){
               vctMTData.addElement(mtMsg);
             }

//           } // end for (;iBeginIndex  < iEndIndex;.....
        }
    }
    catch(Exception e){
       e.printStackTrace();
       return false;
    }

    //关闭连接
    try{
        rs.close() ;
        stmt.close() ;
    }
    catch(SQLException e){
      e.printStackTrace();
      this.isAvail = false;
      return false;
    }

    //System.out.println("开始只标志为");

    //置发送标记SEND_STATUS为-2
    CMPP mtMsg;
    int sndSize    = vctMTData.size() ;

    for(int i = 0; i < sndSize; i++){
        mtMsg = (CMPP)vctMTData.elementAt (i);
        strSql = "update cmpp_to_ismg set submit_status = -2 where to_ismg_id =  " + mtMsg.Msg_Id ;
        try {
            stmt = conn.createStatement();
            stmt.execute(strSql);
            stmt.close() ;
        }
        catch (SQLException ex) {
            System.out.println("[CMPPMaster]DBAccess.Set Send_Status error :" +ex.toString());
            vctMTData.removeElementAt (i);
            i--;
            sndSize--;
            this.isAvail = false;
            return false;
        }
    }

    discloseconn(conn,stmt,null);  //关闭connection(放回连接池)
    return true;
  }




  /**释放连接池的连接closeconn();  //关闭connection(放回连接池)**/
  private void discloseconn(Connection conn,Statement stmt,ResultSet rst){
    try{
      if(rst != null){
        rst.close();
      }
       if(conn != null){
        conn.close();
      }
      if(stmt != null){
        stmt.close();
      }
   }
    catch (Exception sqle){
      System.out.println("Database disconnecting failed"+sqle);
    }
  }



  /**
   * <p>Title: CMPPMODataSaveThread保存上行数据的线程</p>
   * <p>Description: 湖南移动短信网关通讯程序</p>
   * <p>Copyright: Copyright (c) 2008</p>
   * <p>Company: Sunrise tech ltd.</p>
   * @author pansonlu
   * @version 1.0
   */
  public class CMPPMODataSaveThread extends Thread{

    public boolean isAvail = false;

    public CMPPMODataSaveThread(){
      isAvail = true;
    }

    public void run(){
      //转储缓冲中的MO数据
      Vector vctWillSaveMO = new Vector(1,1);

      try{
        while(isAvail && CMPPDBAccess.getInstance().isAvail){

          if(vctMOData.size()>0){
            synchronized(vctMOData){

              for(int i=0;i<vctMOData.size();i++){
                vctWillSaveMO.addElement(vctMOData.elementAt(i));
              }
              //清空缓冲中的数据
              vctMOData.removeAllElements();
            }

            //调用CMPPDBAccess的存储方法存储上行数据
            CMPPDBAccess.getInstance().saveMODataFromVector(vctWillSaveMO);
            vctWillSaveMO.removeAllElements();

          }
          else{
            PublicFunction.sleep(20);
          }
        }
      }
      catch(Exception ex){
        //打印异常,并关闭连接
        Debug.outInfo("[CMPPMaster]保存MO数据失败:"+ex.toString());
        //将未写入数据库的MO数据放回原MO缓冲
        if(vctWillSaveMO.size()>0){
          for(int i =0;i<vctWillSaveMO.size();i++){
            vctMOData.addElement(vctWillSaveMO.elementAt(i));
            vctWillSaveMO.removeElementAt(i);
          }
        }
        DBAccess.getInstance().disclose(null);
      }
    }
  }



  /**
   * 发送成功后，删除待发送表中的数据
   * @param mtData
   * @return
   */
  public void dealProccessMsg(CMPP cmppMsg){
    Statement stmt = null;
    PreparedStatement pstmt = null;
    String sqlStatement =null;
    Connection conn     = null;

    try {

      if(cmppMsg.Result ==0){
        String strSql ="delete from cmpp_to_ismg where to_ismg_id = "+cmppMsg.Sequence_Id;
        Debug.outDebug(strSql);
        conn = ConnectionPool.getConnection();
        stmt = conn.createStatement();
        stmt.execute(strSql);
        stmt.close();
      }
      else{
        System.out.println("[CMPPMaster]发送短信失败，错误状态码:" +cmppMsg.Result+"") ;
        String strSql ="update cmpp_to_ismg set submit_status = "+cmppMsg.Result+" where to_ismg_id = "+cmppMsg.Sequence_Id;
        conn = ConnectionPool.getConnection();
        stmt = conn.createStatement();
        stmt.execute(strSql);
        stmt.close();
      }

    }
    catch(Exception e){
      System.out.println("[CMPPMaster]delete table XCZQ_SMS_OUT has error:msgid=" + cmppMsg.Result +","+e) ;
      disclose(conn);
    }
    finally{
      discloseconn(conn,null,null);  //关闭connection(放回连接池)
    }
  }




  /**
   * 断开数据库连接
   */
  private void disclose(Connection conn) {
    try {
      if(this.isAvail){
        if(conn != null){
          conn.close();
        }
        this.isAvail = false;
        Debug.outInfo("通讯程序关闭数据库连接");
      }
    }
    catch (Exception sqle){
      System.out.println("Database disconnecting failed");
    }
  }


}