@@ -26,6 +26,7 @@ const ( | |||||
PVD_JDOwn = "jd_own" | PVD_JDOwn = "jd_own" | ||||
PVD_JDUNIONOWN = "jd_union_own" | PVD_JDUNIONOWN = "jd_union_own" | ||||
PVD_PDDUNIONOWN = "pdd_union_own" | PVD_PDDUNIONOWN = "pdd_union_own" | ||||
PVD_WPHUNIONOWN = "vip_union_own" | |||||
PVD_JDUNIONOWNUPDATE = "jd_union_own_update" | PVD_JDUNIONOWNUPDATE = "jd_union_own_update" | ||||
PVD_SN = "suning" | PVD_SN = "suning" | ||||
PVD_VIP = "vip" | PVD_VIP = "vip" | ||||
@@ -82,6 +82,9 @@ func OrderSaveCreateUpdate(eg *xorm.Engine, pvd string, ordData *[]md.OrderInfo, | |||||
if pvd == md.PVD_TO_KFC { //判断下 | if pvd == md.PVD_TO_KFC { //判断下 | ||||
pvd = md.PVD_KFC | pvd = md.PVD_KFC | ||||
} | } | ||||
if pvd == md.PVD_WPHUNIONOWN { //判断下 | |||||
pvd = md.PVD_VIP | |||||
} | |||||
if pvd == md.PVD_TIKTOKOWNLIVE { //判断下 | if pvd == md.PVD_TIKTOKOWNLIVE { //判断下 | ||||
pvd = md.PVD_TIKTOKLIVE | pvd = md.PVD_TIKTOKLIVE | ||||
} | } | ||||
@@ -22,10 +22,11 @@ func initConsumes() { | |||||
//jobs[consumeMd.ZhiosOrderTrackRefundForTbAllConsumeFunName] = ZhiosOrderTrackRefundForTbAllConsume //淘宝-全部退款 | //jobs[consumeMd.ZhiosOrderTrackRefundForTbAllConsumeFunName] = ZhiosOrderTrackRefundForTbAllConsume //淘宝-全部退款 | ||||
//jobs[consumeMd.ZhiosOrderTrackRefundForTbPartConsumeFunName] = ZhiosOrderTrackRefundForTbPartConsume //淘宝-部分退款 | //jobs[consumeMd.ZhiosOrderTrackRefundForTbPartConsumeFunName] = ZhiosOrderTrackRefundForTbPartConsume //淘宝-部分退款 | ||||
jobs[consumeMd.ZhiosOrderTrackCreateForJdConsumeFunName] = ZhiosOrderTrackCreateForJdConsume //京东-创建订单 | |||||
jobs[consumeMd.ZhiosOrderTrackUpdateForJdConsumeFunName] = ZhiosOrderTrackUpdateForJdConsume //京东-更新订单 | |||||
jobs[consumeMd.ZhiosOrderTrackCreateForPddConsumeFunName] = ZhiosOrderTrackCreateForPddConsume //拼多多-全部订单 | |||||
jobs[consumeMd.ZhiosOrderTrackCreateForTikTokTeamConsumeFunName] = ZhiosOrderTrackCreateForTikTokTeamConsume //抖音团长服务费-全部订单 | |||||
//jobs[consumeMd.ZhiosOrderTrackCreateForJdConsumeFunName] = ZhiosOrderTrackCreateForJdConsume //京东-创建订单 | |||||
//jobs[consumeMd.ZhiosOrderTrackUpdateForJdConsumeFunName] = ZhiosOrderTrackUpdateForJdConsume //京东-更新订单 | |||||
//jobs[consumeMd.ZhiosOrderTrackCreateForPddConsumeFunName] = ZhiosOrderTrackCreateForPddConsume //拼多多-全部订单 | |||||
//jobs[consumeMd.ZhiosOrderTrackCreateForTikTokTeamConsumeFunName] = ZhiosOrderTrackCreateForTikTokTeamConsume //抖音团长服务费-全部订单 | |||||
jobs[consumeMd.ZhiosOrderTrackCreateForWphConsumeFunName] = ZhiosOrderTrackCreateForWphConsume //唯品会-全部订单 | |||||
} | } | ||||
@@ -56,6 +56,15 @@ var RabbitMqQueueKeyList = []*MqQueue{ | |||||
BindKey: "", | BindKey: "", | ||||
ConsumeFunName: "ZhiosOrderTrackCreateForPddConsume", | ConsumeFunName: "ZhiosOrderTrackCreateForPddConsume", | ||||
}, | }, | ||||
{ | |||||
ExchangeName: "zhios.order_track_create.exchange", | |||||
Name: "order_track_create_for_wph", | |||||
Type: DirectQueueType, | |||||
IsPersistent: false, | |||||
RoutKey: "pdd", | |||||
BindKey: "", | |||||
ConsumeFunName: "ZhiosOrderTrackCreateForWphConsume", | |||||
}, | |||||
{ | { | ||||
ExchangeName: "zhios.order_track_create.exchange", | ExchangeName: "zhios.order_track_create.exchange", | ||||
Name: "order_track_create_for_tik_tok_team", | Name: "order_track_create_for_tik_tok_team", | ||||
@@ -90,6 +99,7 @@ const ( | |||||
ZhiosOrderTrackCreateForTbConsumeFunName = "ZhiosOrderTrackCreateForTbConsume" | ZhiosOrderTrackCreateForTbConsumeFunName = "ZhiosOrderTrackCreateForTbConsume" | ||||
ZhiosOrderTrackCreateForJdConsumeFunName = "ZhiosOrderTrackCreateForJdConsume" | ZhiosOrderTrackCreateForJdConsumeFunName = "ZhiosOrderTrackCreateForJdConsume" | ||||
ZhiosOrderTrackCreateForPddConsumeFunName = "ZhiosOrderTrackCreateForPddConsume" | ZhiosOrderTrackCreateForPddConsumeFunName = "ZhiosOrderTrackCreateForPddConsume" | ||||
ZhiosOrderTrackCreateForWphConsumeFunName = "ZhiosOrderTrackCreateForWphConsume" | |||||
ZhiosOrderTrackCreateForTikTokTeamConsumeFunName = "ZhiosOrderTrackCreateForTikTokTeamConsume" | ZhiosOrderTrackCreateForTikTokTeamConsumeFunName = "ZhiosOrderTrackCreateForTikTokTeamConsume" | ||||
ZhiosOrderTrackUpdateForJdConsumeFunName = "ZhiosOrderTrackUpdateForJdConsume" | ZhiosOrderTrackUpdateForJdConsumeFunName = "ZhiosOrderTrackUpdateForJdConsume" | ||||
ZhiosOrderTrackRefundForTbAllConsumeFunName = "ZhiosOrderTrackRefundForTbAllConsume" | ZhiosOrderTrackRefundForTbAllConsumeFunName = "ZhiosOrderTrackRefundForTbAllConsume" | ||||
@@ -0,0 +1,90 @@ | |||||
package consume | |||||
import ( | |||||
"applet/app/cfg" | |||||
"applet/app/db" | |||||
md2 "applet/app/md" | |||||
svc2 "applet/app/svc" | |||||
"applet/app/utils" | |||||
"applet/app/utils/logx" | |||||
"applet/consume/md" | |||||
"code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit" | |||||
"code.fnuoos.com/go_rely_warehouse/zyos_go_order_relate_rule.git/rule/one_circles" | |||||
"encoding/json" | |||||
"errors" | |||||
"fmt" | |||||
"github.com/streadway/amqp" | |||||
"time" | |||||
) | |||||
func ZhiosOrderTrackCreateForWphConsume(queue md.MqQueue) { | |||||
fmt.Println(">>>>>>>>>>>>ZhiosOrderTrackCreateForWphConsume>>>>>>>>>>>>") | |||||
ch, err := rabbit.Cfg.Pool.GetChannel() | |||||
if err != nil { | |||||
logx.Error(err) | |||||
return | |||||
} | |||||
defer ch.Release() | |||||
//1、将自己绑定到交换机上 | |||||
ch.Bind(queue.Name, queue.ExchangeName, queue.RoutKey) | |||||
//2、取出数据进行消费 | |||||
ch.Qos(1) | |||||
delivery := ch.Consume(queue.Name, false) | |||||
one_circles.Init(cfg.RedisAddr) | |||||
var res amqp.Delivery | |||||
var ok bool | |||||
for { | |||||
res, ok = <-delivery | |||||
if ok == true { | |||||
err = handleZhiosOrderTrackCreateForWphConsume(res.Body) | |||||
fmt.Println("err ::: ", err) | |||||
if err != nil { | |||||
fmt.Println("ZhiosOrderTrackCreateForWphConsume_ERR:::::", err.Error()) | |||||
_ = res.Reject(true) //TODO::拒绝 Ack | |||||
//_ = res.Reject(false) | |||||
var msg interface{} | |||||
json.Unmarshal(res.Body, &msg) | |||||
if err.Error() == "Connection timed out" { | |||||
//TODO::重新推回队列末尾,避免造成队列堵塞 | |||||
ch.Publish(queue.ExchangeName, msg, queue.RoutKey) | |||||
} else { | |||||
//TODO::推入新的队列中备份 | |||||
utils.FilePutContents("ZhiosOrderTrackCreateForWphConsume_ERR", utils.SerializeStr(err.Error())) | |||||
ch.Publish("zhios.order_track_create_exception.exchange", map[string]interface{}{ | |||||
"msg": msg, | |||||
"err": err.Error(), | |||||
}, "pdd") | |||||
} | |||||
} else { | |||||
err = res.Ack(true) | |||||
} | |||||
} else { | |||||
panic(errors.New("error getting message")) | |||||
} | |||||
} | |||||
fmt.Println("get msg done") | |||||
} | |||||
func handleZhiosOrderTrackCreateForWphConsume(msgData []byte) error { | |||||
//1、解析mq中queue的数据结构体 | |||||
var msg md2.ZhiosOrderTrackCreateRoutKeyForTb | |||||
err := json.Unmarshal(msgData, &msg) | |||||
if err != nil { | |||||
return err | |||||
} | |||||
time.Sleep(time.Microsecond * 100) // 等待100毫秒 | |||||
fmt.Println("order_track_create_for_Pdd_message:::::::::::>>>>>>>>>") | |||||
fmt.Println(msg) | |||||
if db.DBs[msg.DbName] == nil { | |||||
return nil | |||||
} | |||||
engine := db.DBs[msg.DbName] | |||||
err = svc2.OrderSaveCreateUpdate(engine, msg.Pvd, msg.OrderData, msg.Opts, msg.DbName) | |||||
if err != nil { | |||||
fmt.Println("order_track_create_for_Pdd_ERR:::::::::::>>>>>>>>>", err.Error()) | |||||
return err | |||||
} | |||||
return nil | |||||
} |