正在显示
4 个修改的文件
包含
27 行增加
和
9 行删除
@@ -79,7 +79,7 @@ spec: | @@ -79,7 +79,7 @@ spec: | ||
79 | - name: BUSINESS_ADMIN_HOST | 79 | - name: BUSINESS_ADMIN_HOST |
80 | value: "http://suplus-business-admin-prd.fjmaimaimai.com" | 80 | value: "http://suplus-business-admin-prd.fjmaimaimai.com" |
81 | - name: KAFKA_HOST | 81 | - name: KAFKA_HOST |
82 | - value: "" | 82 | + value: "192.168.0.250:9092;192.168.0.251:9092;192.168.0.252:9092" |
83 | - name: KAFKA_CONSUMER_ID | 83 | - name: KAFKA_CONSUMER_ID |
84 | value: "partnermg_prd" | 84 | value: "partnermg_prd" |
85 | volumes: | 85 | volumes: |
@@ -76,7 +76,7 @@ spec: | @@ -76,7 +76,7 @@ spec: | ||
76 | - name: BUSINESS_ADMIN_HOST | 76 | - name: BUSINESS_ADMIN_HOST |
77 | value: "http://suplus-business-admin-test.fjmaimaimai.com" | 77 | value: "http://suplus-business-admin-test.fjmaimaimai.com" |
78 | - name: KAFKA_HOST | 78 | - name: KAFKA_HOST |
79 | - value: "106.52.15.41:9092" | 79 | + value: "192.168.0.250:9092;192.168.0.251:9092;192.168.0.252:9092" |
80 | - name: KAFKA_CONSUMER_ID | 80 | - name: KAFKA_CONSUMER_ID |
81 | value: "partnermg_test" | 81 | value: "partnermg_test" |
82 | volumes: | 82 | volumes: |
@@ -82,8 +82,6 @@ func NewRuner() *Runer { | @@ -82,8 +82,6 @@ func NewRuner() *Runer { | ||
82 | groupId: configs.Cfg.ConsumerId, | 82 | groupId: configs.Cfg.ConsumerId, |
83 | topicsHandles: TopicHandleRouters, | 83 | topicsHandles: TopicHandleRouters, |
84 | topics: topics, | 84 | topics: topics, |
85 | - // beforeHandles: BeforeHandles, | ||
86 | - // afterHandles: AfterHandles, | ||
87 | }, | 85 | }, |
88 | } | 86 | } |
89 | logs.Debug("kafka_host=%v; topic=%v;groupid=%s ", r.msgConsumer.kafkaHosts, | 87 | logs.Debug("kafka_host=%v; topic=%v;groupid=%s ", r.msgConsumer.kafkaHosts, |
@@ -94,7 +92,7 @@ func NewRuner() *Runer { | @@ -94,7 +92,7 @@ func NewRuner() *Runer { | ||
94 | func (r *Runer) InitConsumer() error { | 92 | func (r *Runer) InitConsumer() error { |
95 | config := sarama.NewConfig() | 93 | config := sarama.NewConfig() |
96 | //config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin | 94 | //config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin |
97 | - config.Consumer.Offsets.Initial = sarama.OffsetOldest | 95 | + config.Consumer.Offsets.Initial = sarama.OffsetNewest |
98 | config.Version = sarama.V0_10_2_1 | 96 | config.Version = sarama.V0_10_2_1 |
99 | if err := config.Validate(); err != nil { | 97 | if err := config.Validate(); err != nil { |
100 | msg := fmt.Sprintf("Kafka producer config invalidate. config: %v. err: %v", configs.Cfg, err) | 98 | msg := fmt.Sprintf("Kafka producer config invalidate. config: %v. err: %v", configs.Cfg, err) |
1 | package consumer | 1 | package consumer |
2 | 2 | ||
3 | import ( | 3 | import ( |
4 | + "os" | ||
5 | + | ||
4 | "github.com/Shopify/sarama" | 6 | "github.com/Shopify/sarama" |
5 | "github.com/astaxie/beego/logs" | 7 | "github.com/astaxie/beego/logs" |
8 | + "gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/port/consumer/handles" | ||
6 | ) | 9 | ) |
7 | 10 | ||
8 | //TopicHandle 处理kafka中得消息 | 11 | //TopicHandle 处理kafka中得消息 |
9 | type TopicHandle func(*sarama.ConsumerMessage) error | 12 | type TopicHandle func(*sarama.ConsumerMessage) error |
10 | 13 | ||
11 | -var BeforeHandles = []TopicHandle{} | ||
12 | -var AfterHandles = []TopicHandle{} | ||
13 | - | ||
14 | //TopicHandleRouters 根据topic区分消息并进行处理 | 14 | //TopicHandleRouters 根据topic区分消息并进行处理 |
15 | var TopicHandleRouters = map[string]TopicHandle{ | 15 | var TopicHandleRouters = map[string]TopicHandle{ |
16 | "topic_test": func(message *sarama.ConsumerMessage) error { | 16 | "topic_test": func(message *sarama.ConsumerMessage) error { |
@@ -18,5 +18,25 @@ var TopicHandleRouters = map[string]TopicHandle{ | @@ -18,5 +18,25 @@ var TopicHandleRouters = map[string]TopicHandle{ | ||
18 | message.Timestamp, message.Topic, message.Offset, string(message.Value)) | 18 | message.Timestamp, message.Topic, message.Offset, string(message.Value)) |
19 | return nil | 19 | return nil |
20 | }, | 20 | }, |
21 | - //"xiangmi-orders": handles.DataFromXiangMi, | 21 | +} |
22 | + | ||
23 | +func init() { | ||
24 | + var runEnv string | ||
25 | + if os.Getenv("KAFKA_CONSUMER_ID") != "" { | ||
26 | + runEnv = os.Getenv("KAFKA_CONSUMER_ID") | ||
27 | + } | ||
28 | + if runEnv == "partnermg_test" { | ||
29 | + initHandleRoutersTest() | ||
30 | + } | ||
31 | + if runEnv == "partnermg_prd" { | ||
32 | + initHandleRoutersProd() | ||
33 | + } | ||
34 | +} | ||
35 | + | ||
36 | +func initHandleRoutersTest() { | ||
37 | + TopicHandleRouters["xiangmi_project_test"] = handles.DataFromXiangMi | ||
38 | +} | ||
39 | + | ||
40 | +func initHandleRoutersProd() { | ||
41 | + TopicHandleRouters["xiangmi_project"] = handles.DataFromXiangMi | ||
22 | } | 42 | } |
-
请 注册 或 登录 后发表评论