Browse Source

update

master
dengbiao 4 months ago
parent
commit
2506cac17a
1 changed files with 28 additions and 30 deletions
  1. +28
    -30
      consume/one_circles_sign_in_consume.go

+ 28
- 30
consume/one_circles_sign_in_consume.go View File

@@ -12,8 +12,6 @@ import (
"errors" "errors"
"fmt" "fmt"
"github.com/streadway/amqp" "github.com/streadway/amqp"
"os"
"runtime/pprof"
) )


func OneCirclesSignInGreenEnergyConsume(queue md.MqQueue) { func OneCirclesSignInGreenEnergyConsume(queue md.MqQueue) {
@@ -34,44 +32,44 @@ func OneCirclesSignInGreenEnergyConsume(queue md.MqQueue) {


var res amqp.Delivery var res amqp.Delivery
var ok bool var ok bool
//for {
res, ok = <-delivery
if ok == true {
err = handleOneCirclesSignInGreenEnergy(ch, res.Body)
if err != nil {
fmt.Println("err ::: ", err)
utils.FilePutContents("OneCirclesSignInGreenEnergyConsume_ERR", "[err]:"+err.Error())
_ = res.Reject(true)
//_ = res.Reject(false)
////TODO::重新推回队列末尾,避免造成队列堵塞
//var msg *md.OneCirclesStructForSignIn
//json.Unmarshal(res.Body, &msg)
//ch.Publish(queue.ExchangeName, msg, queue.RoutKey)
for {
res, ok = <-delivery
if ok == true {
err = handleOneCirclesSignInGreenEnergy(ch, res.Body)
if err != nil {
fmt.Println("err ::: ", err)
utils.FilePutContents("OneCirclesSignInGreenEnergyConsume_ERR", "[err]:"+err.Error())
_ = res.Reject(true)
//_ = res.Reject(false)
////TODO::重新推回队列末尾,避免造成队列堵塞
//var msg *md.OneCirclesStructForSignIn
//json.Unmarshal(res.Body, &msg)
//ch.Publish(queue.ExchangeName, msg, queue.RoutKey)
} else {
_ = res.Ack(true)
}
} else { } else {
_ = res.Ack(true)
panic(errors.New("error getting message"))
} }
} else {
panic(errors.New("error getting message"))
} }
//}
fmt.Println("get msg done") fmt.Println("get msg done")
} }


func handleOneCirclesSignInGreenEnergy(ch *rabbit.Channel, msgData []byte) error { func handleOneCirclesSignInGreenEnergy(ch *rabbit.Channel, msgData []byte) error {
cpuProfile, err := os.Create("./cpu_profile")
if err != nil {
fmt.Printf("创建文件失败:%s", err.Error())
return err
}
defer cpuProfile.Close()
//采集CPU信息
pprof.StartCPUProfile(cpuProfile)
defer pprof.StopCPUProfile()
//cpuProfile, err := os.Create("./cpu_profile")
//if err != nil {
// fmt.Printf("创建文件失败:%s", err.Error())
// return err
//}
//defer cpuProfile.Close()
//
////采集CPU信息
//pprof.StartCPUProfile(cpuProfile)
//defer pprof.StopCPUProfile()


//1、解析mq中queue的数据结构体 //1、解析mq中queue的数据结构体
var msg *md.OneCirclesStructForSignIn var msg *md.OneCirclesStructForSignIn
err = json.Unmarshal(msgData, &msg)
err := json.Unmarshal(msgData, &msg)
if err != nil { if err != nil {
return err return err
} }


Loading…
Cancel
Save