优化

collect/collect.go

@@ -41,6 +41,7 @@ type Collect struct {
 	wg           WaitGroupWrap
 	chanPool     chan bool
 	chanItem     chan *Item
+	f            *os.File
 }
 
 func NewCollect(filePath string, needUnzip bool) *Collect {
@@ -69,8 +70,8 @@ func (c *Collect) GetItems() error {
 	if err != nil {
 		return err
 	}
-	defer f.Close()
-	rd := bufio.NewReader(f)
+	c.f = f
+	rd := bufio.NewReader(c.f)
 	c.wg.Wrap(func() {
 
 		for {
@@ -90,22 +91,21 @@ func (c *Collect) GetItems() error {
 		}
 
 	})
-
-	for {
-		select {
-		case item := <-c.chanItem:
-			c.chanPool <- true //缓冲区满之后阻塞，后面的readChan将等待
-			c.wg.Wrap(func() {
-				c.handleItem(item)
-				<-c.chanPool
-			})
-		case <-time.After(30 * time.Second):
-			log.Printf("timed out") //等待30后退出
-			break
-
+	c.wg.Wrap(func() {
+		for {
+			select {
+			case item := <-c.chanItem:
+				c.chanPool <- true //缓冲区满之后阻塞，后面的readChan将等待
+				c.wg.Wrap(func() {
+					c.handleItem(item)
+					<-c.chanPool
+				})
+			case <-time.After(30 * time.Second):
+				log.Printf("timed out") //等待30后退出
+				return
+			}
 		}
-	}
-
+	})
 	return nil
 
 }
@@ -152,5 +152,10 @@ func (c *Collect) handleItem(item *Item) {
 }
 
 func (c *Collect) Exit() {
+
 	c.wg.Wait()
+
+	if c.f != nil {
+		c.f.Close()
+	}
 }

collect/collect_test.go

@@ -22,7 +22,7 @@ func TestOnItem(t *testing.T) {
 	})
 
 	err := c.GetItems()
+	c.Exit()
 	fmt.Println(err)
 
-	c.Exit()
 }