@@ -25,6 +25,7 @@ const ( | |||||
PVD_JD = "jd" | PVD_JD = "jd" | ||||
PVD_JDOwn = "jd_own" | PVD_JDOwn = "jd_own" | ||||
PVD_JDUNIONOWN = "jd_union_own" | PVD_JDUNIONOWN = "jd_union_own" | ||||
PVD_PDDUNIONOWN = "pdd_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" | ||||
@@ -245,7 +245,7 @@ func parsePids(eg *xorm.Engine, masterId, pvd string, pids []string) map[int]*md | |||||
orderType := md.OrderTypeSelf | orderType := md.OrderTypeSelf | ||||
types := "self" | types := "self" | ||||
isContinue := 1 | isContinue := 1 | ||||
if len(s) == 2 && (pvd == md.PVD_JD || pvd == md.PVD_JDOwn) { | |||||
if len(s) == 2 { | |||||
id = utils.StrToInt(s[1]) | id = utils.StrToInt(s[1]) | ||||
types = s[0] | types = s[0] | ||||
isContinue = 0 | isContinue = 0 | ||||
@@ -348,7 +348,9 @@ func OrderSaveUpdate(eg *xorm.Engine, pvd string, ordData *[]md.OrderInfo, opts | |||||
_ = db.DbsOrderUpdate(eg, v.OrdId, &v) | _ = db.DbsOrderUpdate(eg, v.OrdId, &v) | ||||
} | } | ||||
if v.State != ords[v.PvdOid].State && (v.State != 4 || v.Pvd == md.PVD_JD || v.Pvd == md.PVD_JDOwn || v.Pvd == md.PVD_ELM) { | if v.State != ords[v.PvdOid].State && (v.State != 4 || v.Pvd == md.PVD_JD || v.Pvd == md.PVD_JDOwn || v.Pvd == md.PVD_ELM) { | ||||
v.Reason = ords[v.PvdOid].Reason | |||||
if ords[v.PvdOid].Reason != "" { | |||||
v.Reason = ords[v.PvdOid].Reason | |||||
} | |||||
v.State = ords[v.PvdOid].State | v.State = ords[v.PvdOid].State | ||||
if ords[v.PvdOid].State != 4 && (v.State == 5 || v.SettleAt > 0) { //如果不是失效 平台结算状态都不改了 | if ords[v.PvdOid].State != 4 && (v.State == 5 || v.SettleAt > 0) { //如果不是失效 平台结算状态都不改了 | ||||
v.State = 5 | v.State = 5 | ||||
@@ -22,8 +22,9 @@ 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.ZhiosOrderTrackCreateForJdConsumeFunName] = ZhiosOrderTrackCreateForJdConsume //京东-创建订单 | |||||
jobs[consumeMd.ZhiosOrderTrackUpdateForJdConsumeFunName] = ZhiosOrderTrackUpdateForJdConsume //京东-更新订单 | |||||
jobs[consumeMd.ZhiosOrderTrackCreateForPddConsumeFunName] = ZhiosOrderTrackCreateForPddConsume //拼多多-全部订单 | |||||
} | } | ||||
@@ -47,6 +47,15 @@ var RabbitMqQueueKeyList = []*MqQueue{ | |||||
BindKey: "", | BindKey: "", | ||||
ConsumeFunName: "ZhiosOrderTrackCreateForJdConsume", | ConsumeFunName: "ZhiosOrderTrackCreateForJdConsume", | ||||
}, | }, | ||||
{ | |||||
ExchangeName: "zhios.order_track_create.exchange", | |||||
Name: "order_track_create_for_pdd", | |||||
Type: DirectQueueType, | |||||
IsPersistent: false, | |||||
RoutKey: "pdd", | |||||
BindKey: "", | |||||
ConsumeFunName: "ZhiosOrderTrackCreateForPddConsume", | |||||
}, | |||||
{ | { | ||||
ExchangeName: "zhios.order_track_refund.exchange", | ExchangeName: "zhios.order_track_refund.exchange", | ||||
Name: "order_track_refund_for_tb_all", | Name: "order_track_refund_for_tb_all", | ||||
@@ -71,6 +80,7 @@ const ( | |||||
ZhiosOrderTrackUpdateForTbConsumeFunName = "ZhiosOrderTrackUpdateForTbConsume" | ZhiosOrderTrackUpdateForTbConsumeFunName = "ZhiosOrderTrackUpdateForTbConsume" | ||||
ZhiosOrderTrackCreateForTbConsumeFunName = "ZhiosOrderTrackCreateForTbConsume" | ZhiosOrderTrackCreateForTbConsumeFunName = "ZhiosOrderTrackCreateForTbConsume" | ||||
ZhiosOrderTrackCreateForJdConsumeFunName = "ZhiosOrderTrackCreateForJdConsume" | ZhiosOrderTrackCreateForJdConsumeFunName = "ZhiosOrderTrackCreateForJdConsume" | ||||
ZhiosOrderTrackCreateForPddConsumeFunName = "ZhiosOrderTrackCreateForPddConsume" | |||||
ZhiosOrderTrackUpdateForJdConsumeFunName = "ZhiosOrderTrackUpdateForJdConsume" | ZhiosOrderTrackUpdateForJdConsumeFunName = "ZhiosOrderTrackUpdateForJdConsume" | ||||
ZhiosOrderTrackRefundForTbAllConsumeFunName = "ZhiosOrderTrackRefundForTbAllConsume" | ZhiosOrderTrackRefundForTbAllConsumeFunName = "ZhiosOrderTrackRefundForTbAllConsume" | ||||
ZhiosOrderTrackRefundForTbPartConsumeFunName = "ZhiosOrderTrackRefundForTbPartConsume" | ZhiosOrderTrackRefundForTbPartConsumeFunName = "ZhiosOrderTrackRefundForTbPartConsume" | ||||
@@ -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 ZhiosOrderTrackCreateForPddConsume(queue md.MqQueue) { | |||||
fmt.Println(">>>>>>>>>>>>ZhiosOrderTrackCreateForPddConsume>>>>>>>>>>>>") | |||||
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 = handleZhiosOrderTrackCreateForPddConsume(res.Body) | |||||
fmt.Println("err ::: ", err) | |||||
if err != nil { | |||||
fmt.Println("ZhiosOrderTrackCreateForPddConsume_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("ZhiosOrderTrackCreateForPddConsume_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 handleZhiosOrderTrackCreateForPddConsume(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 | |||||
} |