mmm-go / partnermg · 提交

转到一个项目

作者 唐旭辉
提交 e746e21013fd69dae99e8982ab3239a9cd38e0ee 1 个父辈 0a0ace38

调试

内嵌 并排对比
正在显示 4 个修改的文件 包含 125 行增加 和 56 行删除
... ... @@ -6,6 +6,7 @@ import (
"os/signal"
"sync"
"syscall"
"time"
"github.com/astaxie/beego"
"github.com/astaxie/beego/logs"
... ... @@ -13,6 +14,7 @@ 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() {
... ... @@ -37,6 +39,12 @@ 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)
}()
for {
select {
case <-sigs:
... ...
package handles
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"
)
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))
var (
msgData DataFromMessage
err error
)
err = json.Unmarshal(message.Value, &msgData)
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
}
return nil
}
//SyncBestshopOrder 同步
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)
}
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
}
... ...
package produce
// import (
// "fmt"
// "strconv"
// "time"
import (
"fmt"
"strconv"
"time"
// "gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/port/consumer/configs"
"gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/port/consumer/configs"
// "github.com/Shopify/sarama"
// "github.com/astaxie/beego/logs"
// )
"github.com/Shopify/sarama"
"github.com/astaxie/beego/logs"
)
// var (
// producer sarama.SyncProducer
// )
var (
producer sarama.SyncProducer
)
// func init() {
func init() {
// logs.Info("init kafka producer, it may take a few seconds to init the connection\n")
// var err error
// mqConfig := sarama.NewConfig()
// mqConfig.Producer.Return.Successes = true
// mqConfig.Version = sarama.V0_10_2_1
// if err = mqConfig.Validate(); err != nil {
// msg := fmt.Sprintf("Kafka producer config invalidate. config: %v. err: %v", configs.Cfg, err)
// logs.Info(msg)
// panic(msg)
// }
logs.Info("init kafka producer, it may take a few seconds to init the connection\n")
var err error
mqConfig := sarama.NewConfig()
mqConfig.Producer.Return.Successes = true
mqConfig.Version = sarama.V0_10_2_1
if err = mqConfig.Validate(); err != nil {
msg := fmt.Sprintf("Kafka producer config invalidate. config: %v. err: %v", configs.Cfg, err)
logs.Info(msg)
panic(msg)
}
// producer, err = sarama.NewSyncProducer(configs.Cfg.Servers, mqConfig)
// if err != nil {
// msg := fmt.Sprintf("Kafak producer create fail. err: %v", err)
// logs.Info(msg)
// panic(msg)
// }
producer, err = sarama.NewSyncProducer(configs.Cfg.Servers, mqConfig)
if err != nil {
msg := fmt.Sprintf("Kafak producer create fail. err: %v", err)
logs.Info(msg)
panic(msg)
}
// }
}
// func produce(topic string, key string, content string) error {
// msg := &sarama.ProducerMessage{
// Topic: topic,
// Key: sarama.StringEncoder(key),
// Value: sarama.StringEncoder(content),
// Timestamp: time.Now(),
// }
func produce(topic string, key string, content string) error {
msg := &sarama.ProducerMessage{
Topic: topic,
Key: sarama.StringEncoder(key),
Value: sarama.StringEncoder(content),
Timestamp: time.Now(),
}
// _, _, err := producer.SendMessage(msg)
// if err != nil {
// msg := fmt.Sprintf("Send Error topic: %v. key: %v. content: %v", topic, key, content)
// logs.Info(msg)
// return err
// }
// logs.Info("Send OK topic:%s key:%s value:%s\n", topic, key, content)
// return nil
// }
_, _, err := producer.SendMessage(msg)
if err != nil {
msg := fmt.Sprintf("Send Error topic: %v. key: %v. content: %v", topic, key, content)
logs.Info(msg)
return err
}
logs.Info("Send OK topic:%s key:%s value:%s\n", topic, key, content)
return nil
}
// func Producer() error {
// key := strconv.FormatInt(time.Now().UTC().UnixNano(), 10)
// value := "this is a new kafka message!"
// err := produce("topic_test", key, value)
// if err != nil {
// logs.Info("producer err:%s \n", err)
// return err
// }
// return nil
// }
func Producer() error {
key := strconv.FormatInt(time.Now().UTC().UnixNano(), 10)
value := "this is a new kafka message123456!"
err := produce("topic_test", key, value)
if err != nil {
logs.Info("producer err:%s \n", err)
return err
}
return nil
}
... ...
... ... @@ -3,6 +3,7 @@ package consumer
import (
"github.com/Shopify/sarama"
"github.com/astaxie/beego/logs"
"gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/port/consumer/handles"
)
//TopicHandle 处理kafka中得消息
... ... @@ -18,5 +19,5 @@ var TopicHandleRouters = map[string]TopicHandle{
message.Timestamp, message.Topic, message.Offset, string(message.Value))
return nil
},
"xiangmi-orders": SyncBestshopOrder,
"xiangmi-orders": handles.DataFromXiangMi,
}
... ...