add kafka demo

@@ -14,7 +14,7 @@ require (
 	//github.com/opentracing/opentracing-go v1.1.1-0.20190913142402-a7454ce5950e
 	//github.com/openzipkin-contrib/zipkin-go-opentracing v0.4.5
 	//github.com/openzipkin/zipkin-go v0.2.5
-	github.com/tiptok/gocomm v1.0.6
+	github.com/tiptok/gocomm v1.0.8
 )
 
-//replace github.com/tiptok/gocomm v1.0.5 => F:\go\src\learn_project\gocomm
+//replace github.com/tiptok/gocomm v1.0.8 => F:\go\src\learn_project\gocomm

@@ -57,5 +57,8 @@ func main() {
 	//}
 	log.Debug(constant.REDIS_HOST, constant.REDIS_PORT)
 
+	//go message.RunConsumer()
+	//go message.RunPublish()
+
 	beego.Run()
 }

+package message
+
+import (
+	"github.com/tiptok/gocomm/pkg/broker"
+	"github.com/tiptok/gocomm/pkg/broker/local"
+	"github.com/tiptok/gocomm/pkg/broker/models"
+	"github.com/tiptok/gocomm/pkg/log"
+	"gitlab.fjmaimaimai.com/mmm-go/godevp/pkg/domain"
+	"gitlab.fjmaimaimai.com/mmm-go/godevp/pkg/infrastructure/pg"
+	"time"
+)
+
+func RunPublish() {
+	var (
+		err error
+	)
+
+	producer := broker.NewMessageProducer(
+		models.WithKafkaHost(KAFKA_HOSTS),
+		models.WithMessageProduceRepository(local.NewPgMessageProduceRepository(pg.DB, nil)),
+	)
+	for {
+		messages := []interface{}{
+			domain.Users{
+				Id:         20200129,
+				Name:       "user_2020",
+				Phone:      "18860183058",
+				Status:     1,
+				Roles:      []int64{7, 8},
+				CreateTime: time.Now(),
+				UpdateTime: time.Now(),
+			},
+		}
+		err = producer.Publish(TOPIC_BUSSINESS_ADMIN, messages, models.WithMessageProduceRepository(local.NewPgMessageProduceRepository(pg.DB, nil)))
+		if err != nil {
+			log.Error(err)
+		}
+		log.Debug("publish :", messages)
+		time.Sleep(time.Hour * 24)
+	}
+}

+package message
+
+import (
+	"github.com/tiptok/gocomm/common"
+	"github.com/tiptok/gocomm/pkg/broker/kafkax"
+	"github.com/tiptok/gocomm/pkg/broker/local"
+	"github.com/tiptok/gocomm/pkg/broker/models"
+	"github.com/tiptok/gocomm/pkg/log"
+	"gitlab.fjmaimaimai.com/mmm-go/godevp/pkg/domain"
+	"gitlab.fjmaimaimai.com/mmm-go/godevp/pkg/infrastructure/pg"
+)
+
+const (
+	KAFKA_HOSTS           = "106.52.15.41:9092" // "192.168.0.250:9092,192.168.0.251:9092,192.168.0.252:9092" //"106.52.15.41:9092" //
+	GroupId               = "godevp_test"       // "calf_basic_test"                                          //"godevp"
+	TOPIC_BUSSINESS_ADMIN = "mmm_business_test"
+)
+
+func RunConsumer() {
+	saramaConsumer := kafkax.NewSaramaConsumer(KAFKA_HOSTS, GroupId) // models.WithVersion("0.10.2.1")
+	saramaConsumer.WithTopicHandler(TOPIC_BUSSINESS_ADMIN, UserLoginHandler)
+	saramaConsumer.WithMessageReceiver(local.NewPgMessageReceiverRepository(pg.DB, nil)) // 持久化
+
+	err := saramaConsumer.StartConsume()
+	if err != nil {
+		log.Error(err)
+	}
+}
+
+func UserLoginHandler(message interface{}) error {
+	msg, ok := message.(*models.Message)
+	if ok {
+		var user = &domain.Users{}
+		common.JsonUnmarshal(string(msg.Value), user)
+		log.Info("消费消息:", msg.Id, msg.Topic, msg.Value)
+		log.Info("登录用户信息:", user.Id, user.Name)
+	}
+	return nil
+}