完善功能,添加日志,添加nsq tcp上报

gigibox

2023-06-19 942f3416b333304bde50f0dca5581595f397eafa

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
package nsqclient
 
import (
    "context"
    "fmt"
 
    "kingdee-dbapi/config"
)
 
var producerCli Producer
var consumeCli NsqConsumer
 
func InitNsqProducer() error {
    var err error
    producerCli, err = NewProducer(config.Options.NsqServer)
    if err != nil {
        fmt.Println(err.Error())
    }
 
    return err
}
 
func Produce(topic string, msg []byte) bool {
    if producerCli == nil {
        err := InitNsqProducer()
        if err != nil {
            fmt.Println("Init Nsq Client error:" + err.Error())
            return false
        }
    }
 
    err := producerCli.Publish(topic, msg)
    if err != nil {
        fmt.Println("Publish error:" + err.Error())
    }
 
    return err == nil
}
 
func InitNsqConsumer(topic, channel string, handle func(data []byte) error) {
    if c, err := NewNsqConsumer(context.Background(), topic, channel); err != nil {
        fmt.Println("NewNsqConsumer failed", err)
        return
    } else {
        c.AddHandler(handle)
        if err := c.Run(config.Options.NsqServer, 1); err != nil {
            fmt.Println("run consumer failed", err)
        }
 
    }
}