xnsq

Documentation

Overview

NSQ队列消息 此模块依赖xlog模块，所以使用时请注意使用 `xlog.Regedit()` 来注册日志配置

Index

• func AddNsq(rou string, fun func(param []byte) error) error
• func AddNsqConf(rou string, cnf *NsqFuncConf) error
• func GetNsqStatus() map[string]*NsqFuncConf
• func Listen()
• func Regedit(c *Config)
• func RemoveRegedit()
• func Router(router string, fun func(param []byte) error) error
• func Set(funs string, param any) error
• func SetDef(fun string, param any, t time.Duration) error
• func Start(get, set string) error
• func Startx() error
• func Status() uint8
• func Stop() error
• type Config
• type HttpsQuestStruct
• type NSQ
• type NsqFuncConf

Functions

func AddNsq

func AddNsq(rou string, fun func(param []byte) error) error

添加NSQ队列监听任务

rou	队列下标标识
fun	队列处理函数

func AddNsqConf

func AddNsqConf(rou string, cnf *NsqFuncConf) error

添加NSQ队列监听任务

rou	队列下标标识
cnf	队列任务监听配置，即x时间内最大运行x次

func GetNsqStatus

func GetNsqStatus() map[string]*NsqFuncConf

获取NSQ队列状态

func Listen

func Listen()

NSQ队列监听 导出函数，需要在入口处进行执行 `go xnsq.Listen()` 来进行同步监听

func Regedit

func Regedit(c *Config)

注入配置项

func RemoveRegedit

func RemoveRegedit()

移除队列监听

func Router

func Router(router string, fun func(param []byte) error) error

注册NSQ路由信息 如果路由中已存在，则会返回错误信息

router	注册路由
fun		执行的函数

func Set

func Set(funs string, param any) error

写入队列【若未开启队列的话，将不管队列方法执行的结果，即程序会运行0~1次，即使是失败，所以建议开启NSQ队列，用以保证任务能成功执行】

fun		写入的服务名称，对应读取时的名称配置
param	传入的参数，对应读取时的参数结构

func SetDef

func SetDef(fun string, param any, t time.Duration) error

写入延时队列

fun		写入的服务名称，对应读取时的名称配置
param	传入的参数，对应读取时的参数结构
t		延时时间【目前配置禁止超过1小时的延时任务，若某程序需要延迟1天的话请使用30min进行循环写入的形式来进行实现】

func Start

func Start(get, set string) error

开启服务 - 指定新的服务地址/服务迁移

get	读取XNSQ的地址
set	设置XNSQ的地址

func Startx

func Startx() error

开启服务 - 直接使用原地址进行开启服务

func Status

func Status() uint8

获取当前服务状态

返回值判断：1-停止中 2-已停止 3-运行中

func Stop

func Stop() error

停止服务

Types

type Config

type Config struct {
	Set            string // 写入队列的nsqd服务域名端口，一般为127.0.0.1:4150
	SetTopic       string // 写入队列的topic名称
	Get            string // 读取队列的nsqlookupd域名端口，一般为127.0.0.1:4161，空表示不监听此服务
	GetTopic       string // 读取队列的get_topic服务
	GetChannel     string // 读取队列的get_channel服务
	GetMaxInFlight int    // 最大并发消费数配置，建议配置为nsqd服务的倍数，默认为1
	MaxAttempts    uint16 // 错误最大重试次数
}

配置项信息

type HttpsQuestStruct

type HttpsQuestStruct struct {
	URL    string            `json:"url"`    // 请求网址
	Method string            `json:"method"` // 请求方式：GET/POST/POSTJSON
	Param  any               `json:"param"`  // 请求携带参数，参考https.Post或者https.PostJson中的param
	Header map[string]string `json:"header"` // 请求携带的header头
}

延时HTTP请求

type NSQ

type NSQ struct {
	Func    string `json:"func"`     // 调用方法
	Time    string `json:"time"`     // 发送时间
	Param   any    `json:"param"`    // 传入参数
	RunTime string `json:"run_time"` // 期望运行时间，若此值不为空则表示再次延时
}

NSQ的message消息结构，用于json解构及参数传递

type NsqFuncConf

type NsqFuncConf struct {
	Key           string        // 下标标识
	Name          string        // 队列任务名称
	Desc          string        // 队列任务描述
	RunSuccessNum uint          // 成功运行次数
	RunErrorNum   uint          // 失败运行次数
	PreviousError string        // 上次运行失败的原因
	PreviousTime  string        // 上次运行时间
	TimeSlot      time.Duration // 时间段内
	RunMax        uint          // 允许执行的数量，即到此数量且时间超出后就进行归零

	Func func(param []byte) error `json:"-"` // 执行的函数操作
	// contains filtered or unexported fields
}

队列配置