|
|
@@ -46,6 +46,38 @@ func ZhiosTikTokUpdate(queue md.MqQueue) { |
|
|
|
} |
|
|
|
fmt.Println("get msg done") |
|
|
|
} |
|
|
|
func ZhiosTikTokAllUpdate(queue md.MqQueue) { |
|
|
|
fmt.Println(">>>>>>>>>>>>>>>>>>>>>>>>") |
|
|
|
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(15) |
|
|
|
delivery := ch.Consume(queue.Name, false) |
|
|
|
|
|
|
|
var res amqp.Delivery |
|
|
|
var ok bool |
|
|
|
for { |
|
|
|
res, ok = <-delivery |
|
|
|
if ok == true { |
|
|
|
//fmt.Println(string(res.Body)) |
|
|
|
fmt.Println(">>>>>>>>>>>>>>>>CanalOrderConsume<<<<<<<<<<<<<<<<<<<<<<<<<") |
|
|
|
err = handleZhiosTikTokGoodsUpdate(res.Body) |
|
|
|
//_ = res.Reject(false) |
|
|
|
//if err == nil { |
|
|
|
_ = res.Ack(true) |
|
|
|
//} |
|
|
|
} else { |
|
|
|
panic(errors.New("error getting message")) |
|
|
|
} |
|
|
|
} |
|
|
|
fmt.Println("get msg done") |
|
|
|
} |
|
|
|
|
|
|
|
func handleZhiosTikTokGoodsUpdate(msg []byte) error { |
|
|
|
//1、解析canal采集至mq中queue的数据结构体 |
|
|
|