huangjiajun
6 months ago
1 changed files with 3 additions and 4 deletions
Unified View
Diff Options
+ 3
- 4
consume/canal_user_virtual_coin_flow_consume.go
View File
| @@ -14,7 +14,6 @@ import ( | |||||
| ) | ) | ||||
| func CanalOneOrengeUserVirtualCoinFlowConsume(queue md.MqQueue) { | func CanalOneOrengeUserVirtualCoinFlowConsume(queue md.MqQueue) { | ||||
| fmt.Println(">>>>>>>>>>>>CanalUserVirtualCoinFlowConsume>>>>>>>>>>>>") | |||||
| ch, err := rabbit.Cfg.Pool.GetChannel() | ch, err := rabbit.Cfg.Pool.GetChannel() | ||||
| if err != nil { | if err != nil { | ||||
| logx.Error(err) | logx.Error(err) | ||||
| @@ -24,7 +23,7 @@ func CanalOneOrengeUserVirtualCoinFlowConsume(queue md.MqQueue) { | |||||
| //1、将自己绑定到交换机上 | //1、将自己绑定到交换机上 | ||||
| ch.Bind(queue.Name, queue.ExchangeName, queue.RoutKey) | ch.Bind(queue.Name, queue.ExchangeName, queue.RoutKey) | ||||
| //2、取出数据进行消费 | //2、取出数据进行消费 | ||||
| ch.Qos(1) | |||||
| ch.Qos(1000) | |||||
| delivery := ch.Consume(queue.Name, false) | delivery := ch.Consume(queue.Name, false) | ||||
| var res amqp.Delivery | var res amqp.Delivery | ||||
| @@ -33,10 +32,10 @@ func CanalOneOrengeUserVirtualCoinFlowConsume(queue md.MqQueue) { | |||||
| res, ok = <-delivery | res, ok = <-delivery | ||||
| if ok == true { | if ok == true { | ||||
| //fmt.Println(string(res.Body)) | //fmt.Println(string(res.Body)) | ||||
| fmt.Println(">>>>>>>>>>>>>>>>CanalUserVirtualCoinFlowConsume<<<<<<<<<<<<<<<<<<<<<<<<<") | |||||
| fmt.Println(">>>>>>>>>>>>>>>>CanalOneOrengeUserVirtualCoinFlowConsume<<<<<<<<<<<<<<<<<<<<<<<<<") | |||||
| err = handleCanalOneOrengeUserVirtualCoinFlow(res.Body) | err = handleCanalOneOrengeUserVirtualCoinFlow(res.Body) | ||||
| if err != nil { | if err != nil { | ||||
| fmt.Println("handleCanalUserVirtualCoinFlow_ERR:::::", err.Error()) | |||||
| fmt.Println("CanalOneOrengeUserVirtualCoinFlowConsume_ERR:::::", err.Error()) | |||||
| } | } | ||||
| //_ = res.Reject(false) | //_ = res.Reject(false) | ||||
| err = res.Ack(true) | err = res.Ack(true) | ||||