tb_0030是SEMQ的待确认消息表,存储已接收但等待发送方确认已送达的记录，f003v_0030是确认记录键,可以唯一对应到系统范围内的所有SEMQ实例中的待发送记录.

f003v_0030由节点类型,节点id,SEMQ实例id,SEMQ存储id,记录编号组成.

tb_0030在f003v_0030上建立唯一索引.可以提高定位速度.

 

运行过程中出现了重复insert导致违反约束的情况.---虽然不会造成逻辑错误,但浪费了资源.

检查代码发现,在处理接收到的消息时,检查是否已经接收过此消息,即使判定已经接收过,仍先执行insert，出错后检查是否是违反唯一性约束,是则通过f003v_0030定位该记录,获取记录id,该记录id会出现在发送方处理确认消息后的再确认消息中.再确认的目的是告诉接收方"我知道已经送达了",这时才可以安全地清理tb_0030的记录.

 

消除此错误的代码修改如下:(已修改gyb trunk和das)

(1)semq.h  

SaveAck增加dont_save参数,表示是否保存,如不保存则直接查询待确认记录.

class CSEMQ  : public ISEMQ,public IEventHandler,ACE_Event_Handler {  

public:
   static int SaveAck(SEMQ_ACK_HELPER *ak,short dont_save=0); 
 }
  
(2)semq.cpp   
int CSEMQ::SaveAck(SEMQ_ACK_HELPER *ak,short dont_save) {
 USEDBC(pdbor,ack_queue_dbc_.c_str());

 ITableHandler *th = CBasePlugInModule::db_helper_->NewTableHandler(pdbor,"tb_0030");
 AUTO_POINTER_NODECLARE(ITableHandler,th);

 pdbor->BeginTrans();
 unsigned short num = ak->loc_keys_.size();
 for (unsigned short i=0;i<num;i++) {
  MQ_LOC_KEY *lk = ak->loc_keys_[i];
  char key[128];
  sprintf(key,"%d,%d,%d,%d,%I64d",lk->node_type_,lk->node_id_,lk->mq_id_,lk->mq_db_id_,lk->record_id_);
  unsigned __int64 record_id;
  bool dup_key_error = false;

  if (!dont_save) {
   th->BindField("f003v_0030",key);
   char date[32];
   CDateTime::GetDateTime(date);
   th->BindField("src_type",(char**)&ak->src_.type_,sizeof(ak->src_.type_));
   th->BindField("src_id",(char**)&ak->src_.id_,sizeof(ak->src_.id_));
   th->BindField("dest_type",(char**)&ak->dest_.type_,sizeof(ak->dest_.type_));
   th->BindField("dest_id",(char**)&ak->dest_.id_,sizeof(ak->dest_.id_));
   th->BindField("f005d_0030",date);
   if (th->Insert()) {
    dup_key_error = pdbor->GetDBExt()->IsDuplicateKeyError(pdbor->GetLastErrorCode());
    if (!dup_key_error) {
     return -1;
    }
   }

   if (!dup_key_error) {
    if (pdbor->GetDBExt()->GetLastID(record_id,1,"tb_0030")) {
     return -2;
    }
   }
  }
  if (dont_save||dup_key_error) {
   string sql = LogMsg("select object_id from tb_0030 where f003v_0030='%s'",key);
   _variant_t v;
   if (QueryValue(pdbor,sql.c_str(),&v)) {
    return -3;
   }
   string sVal = ExVariantToString(v);
   record_id = a2Li(sVal.c_str());
  }
  ak->src_record_ids_.push_back(record_id); ///< 保存写入的待确认记录的记录编号
 }
 FAIL_RETURN(pdbor->CommitTrans(),,-1);

 return 0;
}
  
(3)router.cpp  
int CBBoxPlugin::HandleInput(CWrappedMsg<> *pwm,void **pploc,int &action)  {
 int ret = -1;
 try {
  CMsg *msg = pwm->GetMsg();
  bool do_ack = msg->IsAutoAck(); ///< 包含msg->IsAck2
  ret = HandleInput_i(pwm,pploc,action);
  nlogger_->log(LO_STDOUT|LO_FILE,SEVERITY_DEBUG,"************CBBoxPlugin::HandleInput_i()返回%d.\n",ret);
  /// ret: 0-交给插件处理 其它值:错误 (<0)或者转发(>0)
  SEMQ_ACK_HELPER *ah = (SEMQ_ACK_HELPER*)*pploc;
  if (action&&do_ack) {
   if (action==-1)
    semq_.Ack(ah,1);
   else {
    if (semq_.SaveAck(ah,action==3 ? 1:0)==0)
     semq_.Ack(ah,0);
   }
  }
  if (ret) { /// ret==0表示交给插件处理,其它值时释放ah对象
   if (ah) {
    delete ah;
    *pploc = 0;
   }
  }
 } catch(...) {
  nlogger_->log(LO_STDOUT|LO_FILE,SEVERITY_ERROR,"************CBBoxPlugin::HandleInput()异常.\n");
 }

 return ret;
}