优化消息队列

rpc/server.go

@@ -265,11 +265,35 @@ func (server *Server) selfNodeRpcHandlerGo(processor IRpcProcessor, client *Clie
 		return pCall
 	}
 
+	var iParam interface{}
+
+
 	if processor == nil {
 		_, processor = GetProcessorType(args)
 	}
+
+	if args != nil {
+		inParamValue := reflect.New(reflect.ValueOf(args).Type().Elem())
+		//args
+		//复制输入参数
+		iParam = inParamValue.Interface()
+		bytes,err := processor.Marshal(args)
+		if err == nil {
+			err = processor.Unmarshal(bytes,iParam)
+		}
+
+		if err != nil {
+			pCall.Seq = 0
+			pCall.Err = errors.New("RpcHandler " + handlerName + "."+serviceMethod+" deep copy inParam is error:" + err.Error())
+			pCall.done <- pCall
+			log.SError(pCall.Err.Error())
+
+			return pCall
+		}
+	}
+
 	req := MakeRpcRequest(processor, 0, rpcMethodId, serviceMethod, noReply, nil)
-	req.inParam = args
+	req.inParam = iParam
 	req.localReply = reply
 	if rawArgs != nil {
 		var err error
@@ -335,8 +359,23 @@ func (server *Server) selfNodeRpcHandlerAsyncGo(client *Client, callerRpcHandler
 	}
 
 	_, processor := GetProcessorType(args)
+	inParamValue := reflect.New(reflect.ValueOf(args).Type().Elem())
+	//args
+	//复制输入参数
+	iParam := inParamValue.Interface()
+	bytes,err := processor.Marshal(args)
+	if err == nil {
+		err = processor.Unmarshal(bytes,iParam)
+	}
+
+	if err != nil {
+		errM := errors.New("RpcHandler " + handlerName + "."+serviceMethod+" deep copy inParam is error:" + err.Error())
+		log.SError(errM.Error())
+		return errM
+	}
+
 	req := MakeRpcRequest(processor, 0, 0, serviceMethod, noReply, nil)
-	req.inParam = args
+	req.inParam = iParam
 	req.localReply = reply
 
 	if noReply == false {
@@ -370,7 +409,7 @@ func (server *Server) selfNodeRpcHandlerAsyncGo(client *Client, callerRpcHandler
 		}
 	}
 
-	err := rpcHandler.PushRpcRequest(req)
+	err = rpcHandler.PushRpcRequest(req)
 	if err != nil {
 		ReleaseRpcRequest(req)
 		return err

sysservice/messagequeueservice/CustomerSubscriber.go

@@ -25,6 +25,7 @@ type CustomerSubscriber struct {
 	customerId        string
 
 	isStop int32 //退出标记
+	topicCache []TopicData // 从消息队列中取出来的消息的缓存
 }
 
 const DefaultOneBatchQuantity = 1000
@@ -79,6 +80,7 @@ func (cs *CustomerSubscriber) trySetSubscriberBaseInfo(rpcHandler rpc.IRpcHandle
 		cs.StartIndex = uint64(zeroTime.Unix() << 32)
 	}
 
+	cs.topicCache = make([]TopicData, oneBatchQuantity)
 	return nil
 }
 
@@ -156,7 +158,7 @@ func (cs *CustomerSubscriber) SubscribeRun() {
 
 func (cs *CustomerSubscriber) subscribe() bool {
 	//先从内存中查找
-	topicData, ret := cs.subscriber.queue.FindData(cs.StartIndex, cs.oneBatchQuantity)
+	topicData, ret := cs.subscriber.queue.FindData(cs.StartIndex+1, cs.oneBatchQuantity, cs.topicCache[:0])
 	if ret == true {
 		cs.publishToCustomer(topicData)
 		return true

sysservice/messagequeueservice/MemoryQueue.go

@@ -49,13 +49,22 @@ func (mq *MemoryQueue) findData(startPos int32, startIndex uint64, limit int32)
 	if findStartPos <= mq.tail {
 		findEndPos = mq.tail + 1
 	} else {
-		findEndPos = int32(cap(mq.topicQueue))
+		findEndPos = int32(len(mq.topicQueue))
+	}
+	
+	if findStartPos >= findEndPos {
+		return nil, false
+	}
+
+	// 要取的Seq 比内存中最小的数据的Seq还小，那么需要返回错误
+	if mq.topicQueue[findStartPos].Seq > startIndex {
+		return nil, false
 	}
 
 	//二分查找位置
 	pos := int32(algorithms.BiSearch(mq.topicQueue[findStartPos:findEndPos], startIndex, 1))
 	if pos == -1 {
-		return nil, true
+		return nil, false
 	}
 
 	pos += findStartPos
@@ -69,29 +78,31 @@ func (mq *MemoryQueue) findData(startPos int32, startIndex uint64, limit int32)
 }
 
 // FindData 返回参数[]TopicData 表示查找到的数据，nil表示无数据。bool表示是否不应该在内存中来查
-func (mq *MemoryQueue) FindData(startIndex uint64, limit int32) ([]TopicData, bool) {
+func (mq *MemoryQueue) FindData(startIndex uint64, limit int32, dataQueue []TopicData) ([]TopicData, bool) {
 	mq.locker.RLock()
 	defer mq.locker.RUnlock()
 
 	//队列为空时，应该从数据库查找
 	if mq.head == mq.tail {
 		return nil, false
-	}
-
-	/*
-		//先判断startIndex是否比第一个元素要大
-		headTopic := (mq.head + 1) % int32(len(mq.topicQueue))
-		//此时需要从持久化数据中取
-		if  startIndex+1 > mq.topicQueue[headTopic].Seq {
-			return nil, false
+	} else if mq.head < mq.tail {
+		// 队列没有折叠
+		datas,ret := mq.findData(mq.head + 1, startIndex, limit)
+		if ret {
+			dataQueue = append(dataQueue, datas...)
+		}
+		return dataQueue, ret
+	} else {
+		// 折叠先找后面的部分
+		datas,ret := mq.findData(mq.head+1, startIndex, limit)
+		if ret {
+			dataQueue = append(dataQueue, datas...)
+			return dataQueue, ret
 		}
-	*/
 
-	retData, ret := mq.findData(mq.head+1, startIndex, limit)
-	if mq.head <= mq.tail || ret == true {
-		return retData, true
+		// 后面没找到，从前面开始找
+		datas,ret = mq.findData(0, startIndex, limit)
+		dataQueue = append(dataQueue, datas...)
+		return dataQueue, ret
 	}
-
-	//如果是正常head在后，尾在前，从数组0下标开始找到tail
-	return mq.findData(0, startIndex, limit)
 }

sysservice/messagequeueservice/MessageQueueService.go

@@ -15,7 +15,7 @@ type QueueDataPersist interface {
 	OnExit()
 	OnReceiveTopicData(topic string, topicData []TopicData)                                   //当收到推送过来的数据时
 	OnPushTopicDataToCustomer(topic string, topicData []TopicData)                            //当推送数据到Customer时回调
-	PersistTopicData(topic string, topicData []TopicData, retryCount int) ([]TopicData, bool) //持久化数据,失败则返回false，上层会重复尝试，直到成功，建议在函数中加入次数，超过次数则返回true
+	PersistTopicData(topic string, topicData []TopicData, retryCount int) ([]TopicData, []TopicData, bool) //持久化数据,失败则返回false，上层会重复尝试，直到成功，建议在函数中加入次数，超过次数则返回true
 	FindTopicData(topic string, startIndex uint64, limit int64) []TopicData                   //查找数据,参数bool代表数据库查找是否成功
 	LoadCustomerIndex(topic string, customerId string) (uint64, bool)                         //false时代表获取失败，一般是读取错误，会进行重试。如果不存在时，返回(0,true)
 	GetIndex(topicData *TopicData) uint64                                                     //通过topic数据获取进度索引号

sysservice/messagequeueservice/MongoPersist.go

@@ -123,7 +123,6 @@ func (mp *MongoPersist) OnReceiveTopicData(topic string, topicData []TopicData)
 
 // OnPushTopicDataToCustomer 当推送数据到Customer时回调
 func (mp *MongoPersist) OnPushTopicDataToCustomer(topic string, topicData []TopicData) {
-
 }
 
 // PersistTopicData 持久化数据
@@ -142,20 +141,25 @@ func (mp *MongoPersist) persistTopicData(collectionName string, topicData []Topi
 
 	_, err := s.Collection(mp.dbName, collectionName).InsertMany(ctx, documents)
 	if err != nil {
-		log.SError("PersistTopicData InsertMany fail,collect name is ", collectionName)
+		log.SError("PersistTopicData InsertMany fail,collect name is ", collectionName," error:",err.Error())
 
 		//失败最大重试数量
 		return retryCount >= mp.retryCount
 	}
 
-	//log.SRelease("+++++++++====", time.Now().UnixNano())
 	return true
 }
 
+func (mp *MongoPersist) IsSameDay(timestamp1 int64,timestamp2 int64) bool{
+	t1 := time.Unix(timestamp1, 0)
+	t2 := time.Unix(timestamp2, 0)
+	return t1.Year() == t2.Year() && t1.Month() == t2.Month()&&t1.Day() == t2.Day()
+}
+
 // PersistTopicData 持久化数据
-func (mp *MongoPersist) PersistTopicData(topic string, topicData []TopicData, retryCount int) ([]TopicData, bool) {
+func (mp *MongoPersist) PersistTopicData(topic string, topicData []TopicData, retryCount int) ([]TopicData, []TopicData, bool) {
 	if len(topicData) == 0 {
-		return nil, true
+		return nil, nil,true
 	}
 
 	preDate := topicData[0].Seq >> 32
@@ -163,7 +167,7 @@ func (mp *MongoPersist) PersistTopicData(topic string, topicData []TopicData, re
 	for findPos = 1; findPos < len(topicData); findPos++ {
 		newDate := topicData[findPos].Seq >> 32
 		//说明换天了
-		if preDate != newDate {
+		if mp.IsSameDay(int64(preDate),int64(newDate)) == false {
 			break
 		}
 	}
@@ -172,11 +176,11 @@ func (mp *MongoPersist) PersistTopicData(topic string, topicData []TopicData, re
 	ret := mp.persistTopicData(collectName, topicData[:findPos], retryCount)
 	//如果失败，下次重试
 	if ret == false {
-		return nil, false
+		return nil, nil, false
 	}
 
 	//如果成功
-	return topicData[findPos:len(topicData)], true
+	return topicData[findPos:len(topicData)], topicData[0:findPos], true
 }
 
 // FindTopicData 查找数据
@@ -394,8 +398,7 @@ func (mp *MongoPersist) PersistIndex(topic string, customerId string, index uint
 
 	ctx, cancel := s.GetDefaultContext()
 	defer cancel()
-	ret, err := s.Collection(mp.dbName, CustomerCollectName).UpdateOne(ctx, condition, updata, UpdateOptionsOpts...)
-	fmt.Println(ret)
+	_, err := s.Collection(mp.dbName, CustomerCollectName).UpdateOne(ctx, condition, updata, UpdateOptionsOpts...)
 	if err != nil {
 		log.SError("PersistIndex fail :", err.Error())
 	}

sysservice/messagequeueservice/Subscriber.go

@@ -27,7 +27,7 @@ func (ss *Subscriber) PushTopicDataToQueue(topic string, topics []TopicData) {
 	}
 }
 
-func (ss *Subscriber) PersistTopicData(topic string, topics []TopicData, retryCount int) ([]TopicData, bool) {
+func (ss *Subscriber) PersistTopicData(topic string, topics []TopicData, retryCount int) ([]TopicData, []TopicData, bool) {
 	return ss.dataPersist.PersistTopicData(topic, topics, retryCount)
 }
 

sysservice/messagequeueservice/TopicRoom.go

@@ -113,25 +113,28 @@ func (tr *TopicRoom) topicRoomRun() {
 		}
 
 		//如果落地失败，最大重试maxTryPersistNum次数
-		var ret bool
-		for j := 0; j < maxTryPersistNum; {
+		for retryCount := 0; retryCount < maxTryPersistNum; {
 			//持久化处理
-			stagingBuff, ret = tr.PersistTopicData(tr.topic, stagingBuff, j+1)
-			//如果存档成功，并且有后续批次，则继续存档
-			if ret == true && len(stagingBuff) > 0 {
-				//二次存档不计次数
-				continue
-			}
-
-			//计数增加一次，并且等待100ms，继续重试
-			j += 1
-			if ret == false {
+			stagingBuff, savedBuff, ret := tr.PersistTopicData(tr.topic, stagingBuff, retryCount+1)
+			
+			if ret == true {
+				// 1. 把成功存储的数据放入内存中
+				if len(savedBuff) > 0 {
+					tr.PushTopicDataToQueue(tr.topic, savedBuff)
+				}
+				
+				// 2. 如果存档成功，并且有后续批次，则继续存档
+				if ret == true && len(stagingBuff) > 0 {
+					continue
+				}
+				
+				// 3. 成功了，跳出
+				break
+			} else {
+				//计数增加一次，并且等待100ms，继续重试
+				retryCount++
 				time.Sleep(time.Millisecond * 100)
-				continue
 			}
-
-			tr.PushTopicDataToQueue(tr.topic, stagingBuff)
-			break
 		}
 	}