Merge pull request #122 from alimy/pr-search

optimize #118 add backend worker update documents to search engine

config.yaml.sample

@@ -53,6 +53,7 @@ JWT: # 鉴权加密
   Expire: 86400
 TweetSearch: # 推文关键字搜索相关配置
   MaxUpdateQPS: 100            # 最大添加/删除/更新Post的QPS，设置范围[10, 10000], 默认100
+  MinWorker: 10                # 最小后台更新工作者, 设置范围[5, 1000], 默认10
 Zinc: # Zinc搜索配置
   Host: http://zinc:4080
   Index: paopao-data

internal/conf/settting.go

@@ -49,6 +49,7 @@ type AppSettingS struct {
 
 type CacheIndexSettingS struct {
 	MaxUpdateQPS int
+	MinWorker    int
 }
 
 type SimpleCacheIndexSettingS struct {
@@ -82,6 +83,7 @@ type FeaturesSettingS struct {
 
 type TweetSearchS struct {
 	MaxUpdateQPS int
+	MinWorker    int
 }
 
 type ZincSettingS struct {

internal/dao/search.go

@@ -47,8 +47,17 @@ func NewTweetSearchService() core.TweetSearchService {
 	}
 	logrus.Infof("use %s as tweet search serice by version %s", bts.ts.Name(), bts.ts.Version())
 
+	numWorker := conf.TweetSearchSetting.MinWorker
+	if numWorker < 5 {
+		numWorker = 5
+	} else if numWorker > 1000 {
+		numWorker = 1000
+	}
+	logrus.Debugf("use %d backend worker to update documents to search engine", numWorker)
 	// 启动文档更新器
-	go bts.startUpdateDocs()
+	for ; numWorker > 0; numWorker-- {
+		go bts.startUpdateDocs()
+	}
 
 	return bts
 }

internal/dao/search_bridge.go

@@ -42,10 +42,17 @@ func (s *bridgeTweetSearchServant) updateDocs(doc *documents) {
 	case s.updateDocsCh <- doc:
 		logrus.Debugln("addDocuments send documents by chan")
 	default:
-		go func(ch chan<- *documents, item *documents) {
-			s.updateDocsCh <- item
-			logrus.Debugln("addDocuments send documents by goroutine")
-		}(s.updateDocsCh, doc)
+		go func(item *documents) {
+			if len(item.docItems) > 0 {
+				if _, err := s.ts.AddDocuments(item.docItems, item.primaryKey...); err != nil {
+					logrus.Errorf("addDocuments in gorotine occurs error: %v", err)
+				}
+			} else if len(item.identifiers) > 0 {
+				if err := s.ts.DeleteDocuments(item.identifiers); err != nil {
+					logrus.Errorf("deleteDocuments in gorotine occurs error: %s", err)
+				}
+			}
+		}(doc)
 	}
 }
 
@@ -55,8 +62,7 @@ func (s *bridgeTweetSearchServant) startUpdateDocs() {
 			if _, err := s.ts.AddDocuments(doc.docItems, doc.primaryKey...); err != nil {
 				logrus.Errorf("addDocuments occurs error: %v", err)
 			}
-		}
-		if len(doc.identifiers) > 0 {
+		} else if len(doc.identifiers) > 0 {
 			if err := s.ts.DeleteDocuments(doc.identifiers); err != nil {
 				logrus.Errorf("deleteDocuments occurs error: %s", err)
 			}