test

parser.go

@@ -166,37 +166,34 @@ func (p *Parser) GetQueue() *Queue {
 // ADParserServer 主入口循环
 func ADParserServer(adp IParser) {
 
+	adresponse := adp.GetLogDB().ADParserSelect(adp.GetSpiderID()) // select from db
+	adrChan := make(chan logdb.ADResonse, 100)
+
 	wg := new(sync.WaitGroup)
 
-	for i := 0; i < 10; i++ {
+	for i := 0; i <= 10; i++ {
 		wg.Add(1)
-		go func(n int) {
-			defer wg.Done()
-			for _ = range data {
+		go parserAndSendMQ(adp, adrChan, wg)
 	}
-		}(i)
-	}
-
-	for i := 0; i < 10000; i++ {
-		data <- i
-	}
-	close(data)
-	wg.Wait()
-
-	adresponse := adp.GetLogDB().ADParserSelect(adp.GetSpiderID()) // select from db
-	adresponseList := make(chan logdb.ADResonse, len(adresponse))
 
 	for _, adr := range adresponse {
-		parserAndSendMQ(&adr, adp)
+		adrChan <- adr
 	}
+
+	close(adrChan)
+	wg.Wait()
 }
 
-func parserAndSendMQ(adr *logdb.ADResonse, adp IParser) {
+func parserAndSendMQ(adp IParser, adrChan chan logdb.ADResonse, wg *sync.WaitGroup) {
+	defer wg.Done()
+
+	for adr := range adrChan {
+
 		pjson, err := adp.ToDoParser(adr.Response)
 		if err != nil {
 			log.Println("uid:", adr.UID, "err:", err)
 			adp.GetLogDB().ADError(adr.UID, err.Error())
-		return
+			continue
 		}
 		// send pjson to mq
 		// update UID status finish
@@ -205,10 +202,10 @@ func parserAndSendMQ(adr *logdb.ADResonse, adp IParser) {
 		if err != nil {
 			log.Println("uid:", adr.UID, "err:", err)
 			adp.GetLogDB().ADError(adr.UID, err.Error())
-		return
+			continue
 		}
-
 		adp.GetLogDB().ADParserSuccess(adr.UID, pjson)
+	}
 }
 
 // NewADParser 创建一个ADParser的类, 包含很多传到终端的所有结构

require.sh

@@ -1,2 +1,3 @@
 https_proxy=474420502.top:7070
 go get -u github.com/streadway/amqp
+go get -u -insecure  474420502.top/test/logdb