作者 庄敏学

kafka topic

1 //go:build !local 1 //go:build !local
  2 +// +build !local
2 3
3 package constant 4 package constant
4 5
@@ -8,14 +9,14 @@ var KAFKA_HOSTS = "127.0.0.1:9092" // 1.116.151.79:9092 @@ -8,14 +9,14 @@ var KAFKA_HOSTS = "127.0.0.1:9092" // 1.116.151.79:9092
8 9
9 var KAFKA_GROUP_ID = "performance_dev" 10 var KAFKA_GROUP_ID = "performance_dev"
10 11
11 -var KAFKA_BUSINESS_ADMIN_TOPIC = "mmm-business-admin-dev" 12 +var KAFKA_BUSINESS_TOPIC = "mmm-business-admin-dev"
12 13
13 func init() { 14 func init() {
14 if os.Getenv("KAFKA_HOSTS") != "" { 15 if os.Getenv("KAFKA_HOSTS") != "" {
15 KAFKA_HOSTS = os.Getenv("KAFKA_HOSTS") 16 KAFKA_HOSTS = os.Getenv("KAFKA_HOSTS")
16 } 17 }
17 - if os.Getenv("KAFKA_BUSINESS_ADMIN_TOPIC") != "" {  
18 - KAFKA_BUSINESS_ADMIN_TOPIC = os.Getenv("KAFKA_BUSINESS_ADMIN_TOPIC") 18 + if os.Getenv("KAFKA_BUSINESS_TOPIC") != "" {
  19 + KAFKA_BUSINESS_TOPIC = os.Getenv("KAFKA_BUSINESS_TOPIC")
19 } 20 }
20 if os.Getenv("KAFKA_GROUP_ID") != "" { 21 if os.Getenv("KAFKA_GROUP_ID") != "" {
21 KAFKA_GROUP_ID = os.Getenv("KAFKA_GROUP_ID") 22 KAFKA_GROUP_ID = os.Getenv("KAFKA_GROUP_ID")
@@ -11,13 +11,14 @@ type User struct { @@ -11,13 +11,14 @@ type User struct {
11 Name string // 用户姓名 11 Name string // 用户姓名
12 Email string // 邮箱 12 Email string // 邮箱
13 Status int // 用户状态(1正常 2禁用) 13 Status int // 用户状态(1正常 2禁用)
14 - DepartmentId []int // 用户归属的部门 14 + DepartmentId []int64 // 用户归属的部门
  15 + PositionId []int64 //用户职位
15 UpdateAt time.Time // 更新时间 16 UpdateAt time.Time // 更新时间
16 DeleteAt *time.Time 17 DeleteAt *time.Time
17 CreateAt time.Time 18 CreateAt time.Time
18 } 19 }
19 20
20 -//1普通员工 2 主管理员 21 +// 1普通员工 2 主管理员
21 const ( 22 const (
22 UserTypeCommon int = 1 23 UserTypeCommon int = 1
23 UserTypeManager int = 2 24 UserTypeManager int = 2
@@ -12,7 +12,7 @@ func Run() { @@ -12,7 +12,7 @@ func Run() {
12 messageHandlerMap := make(map[string]func(message *sarama.ConsumerMessage) error) 12 messageHandlerMap := make(map[string]func(message *sarama.ConsumerMessage) error)
13 messageHandlerMap["demo-v1"] = Demo 13 messageHandlerMap["demo-v1"] = Demo
14 //"指定topic" => 对应的处理方法 14 //"指定topic" => 对应的处理方法
15 - messageHandlerMap["mmm-business-admin-dev"] = handle.SyncDataBusinessAdmin 15 + messageHandlerMap[constant.KAFKA_BUSINESS_TOPIC] = handle.SyncDataBusinessAdmin
16 err := saramaConsumer.StartConsume(constant.KAFKA_HOSTS, constant.SERVICE_NAME, messageHandlerMap, log.Logger) 16 err := saramaConsumer.StartConsume(constant.KAFKA_HOSTS, constant.SERVICE_NAME, messageHandlerMap, log.Logger)
17 log.Logger.Error(err.Error()) 17 log.Logger.Error(err.Error())
18 } 18 }