mmm-go / partnermg · 提交

转到一个项目

作者 唐旭辉
提交 80a728af82b96969459a61775149a89777bb76e9 1 个父辈 a8cf0486

调试

内嵌 并排对比
正在显示 5 个修改的文件 包含 70 行增加 和 70 行删除
... ... @@ -6,7 +6,6 @@ import (
"os/signal"
"sync"
"syscall"
"time"
"github.com/astaxie/beego"
"github.com/astaxie/beego/logs"
... ... @@ -14,7 +13,6 @@ import (
_ "gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/log"
_ "gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/port/beego"
"gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/port/consumer"
"gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/port/consumer/produce"
)
func main() {
... ... @@ -26,7 +24,7 @@ func main() {
beego.Run()
}()
consumerRun := consumer.NewRuner()
logs.Debug("%+v", consumerRun)
if err := consumerRun.InitConsumer(); err != nil {
logs.Error("启动kafka消息消费者失败:%s", err)
}
... ... @@ -40,12 +38,11 @@ func main() {
<-consumerRun.IsReady()
logs.Info("Sarama consumer up and running!...")
}()
go func() {
t := time.NewTimer(7 * time.Second)
<-t.C
err := produce.Producer()
logs.Info("err:%s", err)
}()
// go func() {
// t := time.NewTimer(7 * time.Second)
// <-t.C
// produce.Producer()
// }()
for {
select {
case <-sigs:
... ...
... ... @@ -18,8 +18,8 @@ type MessageConsumer struct {
groupId string
topics []string
topicsHandles map[string]TopicHandle
beforeHandles []TopicHandle
afterHandles []TopicHandle
// beforeHandles []TopicHandle
// afterHandles []TopicHandle
}
//实现对应的接口
... ... @@ -41,13 +41,8 @@ func (c *MessageConsumer) ConsumeClaim(groupSession sarama.ConsumerGroupSession,
err error
)
for message := range groupClaim.Messages() {
logs.Debug("Done Message claimed: timestamp = %v, topic = %s offset = %v value = %v \n",
message.Timestamp, message.Topic, message.Offset, string(message.Value))
for i := range c.beforeHandles {
c.beforeHandles[i](message)
}
groupSession.MarkMessage(message, "")
if topicHandle, err = c.FindTopichandle(groupClaim.Topic()); err != nil {
logs.Error("FindTopichandle err:%s \n", err)
continue
... ... @@ -55,9 +50,7 @@ func (c *MessageConsumer) ConsumeClaim(groupSession sarama.ConsumerGroupSession,
if err = topicHandle(message); err != nil {
logs.Error("Message claimed: kafka消息处理错误 topic =", message.Topic, message.Offset, err)
}
for i := range c.beforeHandles {
c.afterHandles[i](message)
}
groupSession.MarkMessage(message, "")
}
return nil
}
... ... @@ -79,18 +72,20 @@ func NewRuner() *Runer {
for key := range TopicHandleRouters {
topics = append(topics, key)
}
return &Runer{
r := &Runer{
msgConsumer: &MessageConsumer{
ready: make(chan struct{}),
kafkaHosts: configs.Cfg.Servers,
groupId: configs.Cfg.ConsumerId,
topicsHandles: TopicHandleRouters,
topics: topics,
beforeHandles: BeforeHandles,
afterHandles: AfterHandles,
// beforeHandles: BeforeHandles,
// afterHandles: AfterHandles,
},
}
logs.Debug("kafka_host=%v; topic=%v;groupid=%s ", r.msgConsumer.kafkaHosts,
r.msgConsumer.topics, r.msgConsumer.groupId)
return r
}
func (r *Runer) InitConsumer() error {
... ...
package handles
import "encoding/json"
type DataFromMessage struct {
Module string `json:"module"`
Action string `json:"action"`
Data json.RawMessage `json:"data"`
}
... ...
... ... @@ -10,12 +10,6 @@ import (
syncOrderSrv "gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/application/syncOrder/service"
)
type DataFromMessage struct {
Module string `json:"module"`
Action string `json:"action"`
Data json.RawMessage `json:"data"`
}
func DataFromXiangMi(message *sarama.ConsumerMessage) error {
logs.Info("Done Message claimed: timestamp = %v, topic = %s offset = %v value = %v \n",
message.Timestamp, message.Topic, message.Offset, string(message.Value))
... ... @@ -27,33 +21,38 @@ func DataFromXiangMi(message *sarama.ConsumerMessage) error {
if err != nil {
return fmt.Errorf("[Consumer][SyncBestshopOrder] 解析kafka数据失败；%s", err)
}
err = SyncBestshopOrder(msgData.Data)
if err != nil {
e := fmt.Errorf("[Consumer][SyncBestshopOrder] %s", err)
return e
dataAction := msgData.Module + "/" + msgData.Action
switch dataAction {
case "xiangmi.order/ship":
err = syncBestshopOrder(msgData.Data)
if err != nil {
e := fmt.Errorf("[Consumer][SyncBestshopOrder] %s", err)
return e
}
default:
logs.Error("未找到执行动作：Module=%s,Action=%s", msgData.Module, msgData.Action)
}
return nil
}
//SyncBestshopOrder 同步
func SyncBestshopOrder(data []byte) error {
func syncBestshopOrder(data []byte) error {
var (
cmd syncOrderCmd.CreateOrderFromBestshop
err error
)
err = json.Unmarshal(data, &cmd)
if err != nil {
return fmt.Errorf("[Consumer][SyncBestshopOrder] 解析kafka数据失败；%s", err)
return fmt.Errorf("[Consumer][syncBestshopOrder] 解析kafka数据失败；%s", err)
}
if cmd.PartnerId <= 0 {
logs.Info("[Consumer][SyncBestshopOrder] PartnerId<=0 ,不处理消息")
logs.Info("[Consumer][syncBestshopOrder] PartnerId<=0 ,不处理消息")
return nil
}
srv := syncOrderSrv.NewOrderInfoService(nil)
err = srv.SyncOrderFromBestshop(cmd)
if err != nil {
e := fmt.Errorf("[Consumer][SyncBestshopOrder] %s", err)
e := fmt.Errorf("[Consumer][syncBestshopOrder] %s", err)
return e
}
return err
... ...
package consumer
import (
"encoding/json"
"fmt"
// import (
// "encoding/json"
// "fmt"
"github.com/Shopify/sarama"
"github.com/astaxie/beego/logs"
syncOrderCmd "gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/application/syncOrder/command"
syncOrderSrv "gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/application/syncOrder/service"
)
// "github.com/Shopify/sarama"
// "github.com/astaxie/beego/logs"
// syncOrderCmd "gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/application/syncOrder/command"
// syncOrderSrv "gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/application/syncOrder/service"
// )
//SyncBestshopOrder 同步
func SyncBestshopOrder(message *sarama.ConsumerMessage) error {
logs.Info("Done Message claimed: timestamp = %v, topic = %s offset = %v value = %v \n",
message.Timestamp, message.Topic, message.Offset, string(message.Value))
var (
cmd syncOrderCmd.CreateOrderFromBestshop
err error
)
err = json.Unmarshal(message.Value, &cmd)
if err != nil {
return fmt.Errorf("[Consumer][SyncBestshopOrder] 解析kafka数据失败；%s", err)
}
if cmd.PartnerId <= 0 {
logs.Info("[Consumer][SyncBestshopOrder] PartnerId<=0 ,不处理消息")
return nil
}
srv := syncOrderSrv.NewOrderInfoService(nil)
err = srv.SyncOrderFromBestshop(cmd)
if err != nil {
e := fmt.Errorf("[Consumer][SyncBestshopOrder] %s", err)
return e
}
return err
}
// //SyncBestshopOrder 同步
// func SyncBestshopOrder(message *sarama.ConsumerMessage) error {
// logs.Info("Done Message claimed: timestamp = %v, topic = %s offset = %v value = %v \n",
// message.Timestamp, message.Topic, message.Offset, string(message.Value))
// var (
// cmd syncOrderCmd.CreateOrderFromBestshop
// err error
// )
// err = json.Unmarshal(message.Value, &cmd)
// if err != nil {
// return fmt.Errorf("[Consumer][SyncBestshopOrder] 解析kafka数据失败；%s", err)
// }
// if cmd.PartnerId <= 0 {
// logs.Info("[Consumer][SyncBestshopOrder] PartnerId<=0 ,不处理消息")
// return nil
// }
// srv := syncOrderSrv.NewOrderInfoService(nil)
// err = srv.SyncOrderFromBestshop(cmd)
// if err != nil {
// e := fmt.Errorf("[Consumer][SyncBestshopOrder] %s", err)
// return e
// }
// return err
// }
... ...