support multiple topics in one kafka

@@ -24,7 +24,8 @@ Processors:
       Brokers:
         - "172.16.186.16:19092"
         - "172.16.186.17:19092"
-      Topic: k8slog
+      Topics:
+        - k8slog
       Group: pro
       NumProducers: 16
   Filters:

@@ -24,7 +24,8 @@ Processors:
       Brokers:
         - "172.16.186.16:19092"
         - "172.16.186.17:19092"
-      Topic: k8slog
+      Topics:
+        - k8slog
       Group: pro
       NumProducers: 16
   Filters:

@@ -3,7 +3,7 @@ package config
 import (
 	"time"
 
-	"github.com/tal-tech/go-queue/kq"
+	"github.com/tal-tech/go-zero/core/service"
 )
 
 type (
@@ -31,9 +31,22 @@ type (
 		Target     string      `json:",optional"`
 	}
 
+	KafkaConf struct {
+		service.ServiceConf
+		Brokers      []string
+		Group        string
+		Topics       []string
+		Offset       string `json:",options=first|last,default=last"`
+		NumConns     int    `json:",default=1"`
+		NumProducers int    `json:",default=8"`
+		NumConsumers int    `json:",default=8"`
+		MinBytes     int    `json:",default=10240"`    // 10K
+		MaxBytes     int    `json:",default=10485760"` // 10M
+	}
+
 	Processor struct {
 		Input struct {
-			Kafka kq.KqConf
+			Kafka KafkaConf
 		}
 		Filters []Filter `json:",optional"`
 		Output  struct {

@@ -5,7 +5,8 @@ Processors:
       Brokers:
         - ":172.16.186.16:19092"
         - "172.16.186.17:19092"
-      Topic: k8slog
+      Topics:
+        - k8slog
       Group: pro
       NumProducers: 16
   Filters:

@@ -18,6 +18,27 @@ import (
 
 var configFile = flag.String("f", "etc/config.yaml", "Specify the config file")
 
+func toKqConf(c config.KafkaConf) []kq.KqConf {
+	var ret []kq.KqConf
+
+	for _, topic := range c.Topics {
+		ret = append(ret, kq.KqConf{
+			ServiceConf:  c.ServiceConf,
+			Brokers:      c.Brokers,
+			Group:        c.Group,
+			Topic:        topic,
+			Offset:       c.Offset,
+			NumConns:     c.NumConns,
+			NumProducers: c.NumProducers,
+			NumConsumers: c.NumConsumers,
+			MinBytes:     c.MinBytes,
+			MaxBytes:     c.MaxBytes,
+		})
+	}
+
+	return ret
+}
+
 func main() {
 	flag.Parse()
 
@@ -50,7 +71,9 @@ func main() {
 		handle := handler.NewHandler(writer, indexer)
 		handle.AddFilters(filters...)
 		handle.AddFilters(filter.AddUriFieldFilter("url", "uri"))
-		group.Add(kq.MustNewQueue(processor.Input.Kafka, handle))
+		for _, k := range toKqConf(processor.Input.Kafka) {
+			group.Add(kq.MustNewQueue(k, handle))
+		}
 	}
 
 	group.Start()