作者 唐旭辉

修复bug

... ... @@ -35,11 +35,6 @@ func main() {
<-consumerRun.IsReady()
logs.Info("Sarama consumer up and running!...")
}()
// go func() {
// t := time.NewTimer(7 * time.Second)
// <-t.C
// produce.Producer()
// }()
for {
select {
case <-sigs:
... ...
... ... @@ -242,7 +242,6 @@ func (s SyncOrderService) copyOrderBestshopToOrderBase(orderBestshop *domain.Ord
for i := range orderBestshop.Goods {
good := domain.NewOrderGood()
orderBestshop.Goods[i].CopyToOrderGood(&good)
good.OrderId = orderbase.Id
good.CompanyId = partnerData.CompanyId
good.Compute()
good.CurrentBonusStatus.WartPayPartnerBonus(&good)
... ... @@ -257,6 +256,9 @@ func (s SyncOrderService) copyOrderBestshopToOrderBase(orderBestshop *domain.Ord
e := fmt.Sprintf("添加order_base数据失败%s", err)
return lib.ThrowError(lib.INTERNAL_SERVER_ERROR, e)
}
for i := range ordergoods {
ordergoods[i].OrderId = orderbase.Id
}
// 添加goods
err = orderGoodRepository.Save(ordergoods)
if err != nil {
... ...
package produce
import (
"fmt"
"strconv"
"time"
"gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/port/consumer/configs"
"github.com/Shopify/sarama"
"github.com/astaxie/beego/logs"
)
// var (
// producer sarama.SyncProducer
// )
func produce(topic string, key string, content string) error {
var (
producer sarama.SyncProducer
)
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)
}
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
}
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
}
package consumer
// 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"
// )
// //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
// }