ing

kafka/kafka.go

@@ -11,34 +11,55 @@ import (
 
 var (
 	kafkaWriters map[string]*kafka.Writer
+	kafkaReaders map[string]*kafka.Reader
 )
 
 func InitPackage(config *occe.Kafka) {
 	if nil == config {
 		return
 	}
-	if nil == config.Producers || 0 == len(config.Producers) {
+
-		return
+	if nil != config.Producers || 0 < len(config.Producers) {
+		kafkaWriters = make(map[string]*kafka.Writer)
+		for n, c := range config.Producers {
+			wc := kafka.WriterConfig{
+				Brokers:           c.Brokers,
+				Topic:             c.Topic,
+				Balancer:          &kafka.LeastBytes{},
+				MaxAttempts:       c.MaxAttempts,
+				QueueCapacity:     c.QueueCapacity,
+				BatchSize:         c.BatchSize,
+				BatchTimeout:      c.BatchTimeout,
+				ReadTimeout:       c.ReadTimeout,
+				WriteTimeout:      c.WriteTimeout,
+				RebalanceInterval: c.RebalanceInterval,
+				RequiredAcks:      c.RequiredAcks,
+				Async:             c.Async,
+			}
+			kafkaWriters[n] = kafka.NewWriter(wc)
+		}
 	}
 
-	kafkaWriters = make(map[string]*kafka.Writer)
+	if nil != config.Consumers || 0 < len(config.Consumers) {
-
+		kafkaReaders = make(map[string]*kafka.Reader)
-	for n, c := range config.Producers {
+		for n, c := range config.Consumers {
-		wc := kafka.WriterConfig{
+			rc := kafka.ReaderConfig{
-			Brokers:           c.Brokers,
+				Brokers:           c.Brokers,
-			Topic:             c.Topic,
+				GroupID:           c.GroupID,
-			Balancer:          &kafka.LeastBytes{},
+				Topic:             c.Topic,
-			MaxAttempts:       c.MaxAttempts,
+				QueueCapacity:     c.QueueCapacity,
-			QueueCapacity:     c.QueueCapacity,
+				MinBytes:          c.MaxBytes,
-			BatchSize:         c.BatchSize,
+				MaxBytes:          c.MaxBytes,
-			BatchTimeout:      c.BatchTimeout,
+				MaxWait:           c.MaxWait,
-			ReadTimeout:       c.ReadTimeout,
+				ReadLagInterval:   c.ReadLagInterval,
-			WriteTimeout:      c.WriteTimeout,
+				HeartbeatInterval: c.HeartbeatInterval,
-			RebalanceInterval: c.RebalanceInterval,
+				CommitInterval:    c.CommitInterval,
-			RequiredAcks:      c.RequiredAcks,
+				SessionTimeout:    c.SessionTimeout,
-			Async:             c.Async,
+				RebalanceTimeout:  c.RebalanceTimeout,
+				RetentionTime:     c.RetentionTime,
+			}
+			kafkaReaders[n] = kafka.NewReader(rc)
 		}
-		kafkaWriters[n] = kafka.NewWriter(wc)
 	}
 }
 
@@ -68,6 +89,20 @@ func DestroyPackage(config *occe.Kafka) {
 	}
 }
 
+func Writer(name string) *kafka.Writer {
+	if nil == kafkaWriters {
+		return nil
+	}
+	return kafkaWriters[name]
+}
+
+func Reader(name string) *kafka.Reader {
+	if nil == kafkaReaders {
+		return nil
+	}
+	return kafkaReaders[name]
+}
+
 func Write(name string, key []byte, value []byte) error {
 	if nil == kafkaWriters {
 		return fmt.Errorf("Kafka client is not valid")