init
This commit is contained in:
qiyunfanbo126.com
commit
141682065f
|
|
@ -0,0 +1,62 @@
|
||||||
|
# toml配置文件
|
||||||
|
# Wiki：https://github.com/toml-lang/toml
|
||||||
|
ServiceName = "snow"
|
||||||
|
Debug = true
|
||||||
|
Env = "local" # local-本地 develop-开发 beta-预发布 production-线上
|
||||||
|
PrometheusCollectEnable = true
|
||||||
|
SkyWalkingOapServer = "127.0.0.1:11800"
|
||||||
|
|
||||||
|
[Log]
|
||||||
|
Handler = "file"
|
||||||
|
Dir = "./logs"
|
||||||
|
Level = "info"
|
||||||
|
|
||||||
|
[Db]
|
||||||
|
Driver = "mysql"
|
||||||
|
|
||||||
|
[Db.Option]
|
||||||
|
MaxConns = 128
|
||||||
|
MaxIdle = 32
|
||||||
|
IdleTimeout = 180 # second
|
||||||
|
Charset = "utf8mb4"
|
||||||
|
ConnectTimeout = 3 # second
|
||||||
|
|
||||||
|
[Db.Master]
|
||||||
|
Host = "127.0.0.1"
|
||||||
|
Port = 3306
|
||||||
|
User = "root"
|
||||||
|
Password = "123456"
|
||||||
|
DBName = "test"
|
||||||
|
|
||||||
|
[[Db.Slaves]] # 支持多个从库
|
||||||
|
Host = "127.0.0.1"
|
||||||
|
Port = 3306
|
||||||
|
User = "root"
|
||||||
|
Password = "123456"
|
||||||
|
DBName = "test"
|
||||||
|
|
||||||
|
[Api]
|
||||||
|
Host = "0.0.0.0"
|
||||||
|
Port = 8080
|
||||||
|
|
||||||
|
[Cache]
|
||||||
|
Driver = "redis"
|
||||||
|
|
||||||
|
[Redis.Master]
|
||||||
|
Host = "127.0.0.1"
|
||||||
|
Port = 6379
|
||||||
|
#Password = ""
|
||||||
|
#DB = 0
|
||||||
|
|
||||||
|
#[Redis.Option]
|
||||||
|
#MaxIdle = 64
|
||||||
|
#MaxConns = 256
|
||||||
|
#IdleTimeout = 180 # second
|
||||||
|
#ConnectTimeout = 1
|
||||||
|
#ReadTimeout = 1
|
||||||
|
#WriteTimeout = 1
|
||||||
|
|
||||||
|
[AliMns]
|
||||||
|
Url = ""
|
||||||
|
AccessKeyId = ""
|
||||||
|
AccessKeySecret = ""
|
||||||
|
|
@ -0,0 +1,4 @@
|
||||||
|
/.idea
|
||||||
|
/vendor
|
||||||
|
/.env
|
||||||
|
!/.env.example
|
||||||
|
|
@ -0,0 +1,22 @@
|
||||||
|
## Snow
|
||||||
|
Snow是一套简单易用的Go语言业务框架，整体逻辑设计简洁，支持HTTP服务、队列调度和任务调度等常用业务场景模式。
|
||||||
|
|
||||||
|
## Quick start
|
||||||
|
|
||||||
|
### Build
|
||||||
|
sh build/shell/build.sh
|
||||||
|
|
||||||
|
### Run
|
||||||
|
```shell
|
||||||
|
1. build/bin/snow -a api #启动Api服务
|
||||||
|
2. build/bin/snow -a cron #启动Cron定时任务服务
|
||||||
|
3. build/bin/snow -a job #启动队列调度服务
|
||||||
|
4. build/bin/snow -a command -m test #执行名称为test的脚本任务
|
||||||
|
```
|
||||||
|
|
||||||
|
## Documents
|
||||||
|
|
||||||
|
- [项目地址](https://github.com/qit-team/snow)
|
||||||
|
- [中文文档](https://github.com/qit-team/snow/wiki)
|
||||||
|
- [changelog](https://github.com/qit-team/snow/blob/master/CHANGLOG.md)
|
||||||
|
- [xorm](http://gobook.io/read/github.com/go-xorm/manual-zh-CN/)
|
||||||
|
|
@ -0,0 +1,34 @@
|
||||||
|
package bannerlistcache
|
||||||
|
|
||||||
|
import (
|
||||||
|
"sync"
|
||||||
|
|
||||||
|
"quenue/app/caches"
|
||||||
|
|
||||||
|
"github.com/qit-team/snow-core/cache"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
prefix = caches.BannerList //缓存key的前缀
|
||||||
|
)
|
||||||
|
|
||||||
|
var (
|
||||||
|
instance *bannerListCache
|
||||||
|
once sync.Once
|
||||||
|
)
|
||||||
|
|
||||||
|
type bannerListCache struct {
|
||||||
|
cache.BaseCache
|
||||||
|
}
|
||||||
|
|
||||||
|
//单例模式
|
||||||
|
func GetInstance() *bannerListCache {
|
||||||
|
once.Do(func() {
|
||||||
|
instance = new(bannerListCache)
|
||||||
|
instance.Prefix = prefix
|
||||||
|
//instance.DiName = redis.SingletonMain //设置缓存依赖的实例别名
|
||||||
|
//instance.DriverType = cache.DriverTypeRedis //设置缓存驱动的类型,默认redis
|
||||||
|
//instance.SeTTL(86400) 设置默认缓存时间 默认86400
|
||||||
|
})
|
||||||
|
return instance
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,59 @@
|
||||||
|
package bannerlistcache
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"quenue/config"
|
||||||
|
|
||||||
|
"github.com/qit-team/snow-core/cache"
|
||||||
|
_ "github.com/qit-team/snow-core/cache/rediscache"
|
||||||
|
"github.com/qit-team/snow-core/redis"
|
||||||
|
)
|
||||||
|
|
||||||
|
func init() {
|
||||||
|
//加载配置文件
|
||||||
|
conf, err := config.Load("../../../.env")
|
||||||
|
if err != nil {
|
||||||
|
fmt.Println(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
//注册redis类
|
||||||
|
err = redis.Pr.Register(cache.DefaultDiName, conf.Redis)
|
||||||
|
if err != nil {
|
||||||
|
fmt.Println(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func Test_GetMulti(t *testing.T) {
|
||||||
|
ctx := context.TODO()
|
||||||
|
cache := GetInstance()
|
||||||
|
res, _ := cache.Set(ctx, "1000", "a")
|
||||||
|
if res != true {
|
||||||
|
t.Errorf("set key:%s is error", "1000")
|
||||||
|
}
|
||||||
|
|
||||||
|
keys := []string{"1000", "-2000", "9999"}
|
||||||
|
cacheList, err := cache.GetMulti(ctx, keys...)
|
||||||
|
if err != nil {
|
||||||
|
t.Errorf("getMulti error:%s", err.Error())
|
||||||
|
}
|
||||||
|
fmt.Println(cacheList)
|
||||||
|
i := 0
|
||||||
|
for k, v := range cacheList {
|
||||||
|
i++
|
||||||
|
if k == "1000" {
|
||||||
|
if v != "a" {
|
||||||
|
t.Errorf("value of key:%s is error %v", k, v)
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
if v != "" {
|
||||||
|
t.Errorf("value of key:%s is error %v", k, v)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if i != len(keys) {
|
||||||
|
t.Errorf("count of cache key is error: %d", i)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,8 @@
|
||||||
|
package caches
|
||||||
|
|
||||||
|
//缓存前缀key，不同的业务使用不同的前缀，避免了业务之间的重用冲突
|
||||||
|
const (
|
||||||
|
Cookie = "ck:"
|
||||||
|
Copy = "cp:"
|
||||||
|
BannerList = "bl:"
|
||||||
|
)
|
||||||
|
|
@ -0,0 +1,9 @@
|
||||||
|
package console
|
||||||
|
|
||||||
|
import (
|
||||||
|
"github.com/qit-team/snow-core/command"
|
||||||
|
)
|
||||||
|
|
||||||
|
func RegisterCommand(c *command.Command) {
|
||||||
|
c.AddFunc("test", test)
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,15 @@
|
||||||
|
package console
|
||||||
|
|
||||||
|
import (
|
||||||
|
"github.com/robfig/cron"
|
||||||
|
)
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 配置执行计划
|
||||||
|
* @wiki https://godoc.org/github.com/robfig/cron
|
||||||
|
*/
|
||||||
|
func RegisterSchedule(c *cron.Cron) {
|
||||||
|
//c.AddFunc("0 30 * * * *", test)
|
||||||
|
//c.AddFunc("@hourly", test)
|
||||||
|
c.AddFunc("@every 10s", test)
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,7 @@
|
||||||
|
package console
|
||||||
|
|
||||||
|
import "fmt"
|
||||||
|
|
||||||
|
func test() {
|
||||||
|
fmt.Println("run test")
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,14 @@
|
||||||
|
package common
|
||||||
|
|
||||||
|
const (
|
||||||
|
|
||||||
|
//IM
|
||||||
|
MQ_NATS = "nats"
|
||||||
|
MQ_RABBIT = "rabbitmq"
|
||||||
|
MQ_NSQ = "nsq"
|
||||||
|
MQ_KFK = "kafka"
|
||||||
|
MQ_KFK_V2 = "kafka_v2"
|
||||||
|
|
||||||
|
//QUNUE
|
||||||
|
ORDER_RESEND_TOPICAL = "platform_all"
|
||||||
|
)
|
||||||
|
|
@ -0,0 +1,37 @@
|
||||||
|
package errorcode
|
||||||
|
|
||||||
|
const (
|
||||||
|
//成功
|
||||||
|
Success = 200
|
||||||
|
|
||||||
|
//参数错误
|
||||||
|
ParamError = 400
|
||||||
|
|
||||||
|
//未经授权
|
||||||
|
NotAuth = 401
|
||||||
|
|
||||||
|
//请求被禁止
|
||||||
|
Forbidden = 403
|
||||||
|
|
||||||
|
//找不到页面
|
||||||
|
NotFound = 404
|
||||||
|
|
||||||
|
//系统错误
|
||||||
|
SystemError = 500
|
||||||
|
)
|
||||||
|
|
||||||
|
var MsgEN = map[int]string{
|
||||||
|
Success: "success",
|
||||||
|
ParamError: "param error",
|
||||||
|
NotAuth: "not authorized",
|
||||||
|
Forbidden: "forbidden",
|
||||||
|
NotFound: "not found",
|
||||||
|
SystemError: "system error",
|
||||||
|
}
|
||||||
|
|
||||||
|
func GetMsg(code int) string {
|
||||||
|
if msg, ok := MsgEN[code]; ok {
|
||||||
|
return msg
|
||||||
|
}
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,7 @@
|
||||||
|
package logtype
|
||||||
|
|
||||||
|
const (
|
||||||
|
Message = "message"
|
||||||
|
GoPanic = "go.panic"
|
||||||
|
HTTP = "http"
|
||||||
|
)
|
||||||
|
|
@ -0,0 +1,17 @@
|
||||||
|
package event
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
common2 "snow-im/app/constants/common"
|
||||||
|
"snow-im/app/utils"
|
||||||
|
"snow-im/app/utils/common"
|
||||||
|
)
|
||||||
|
|
||||||
|
type RegisterCacheEventHandler struct {
|
||||||
|
}
|
||||||
|
|
||||||
|
func (this *RegisterCacheEventHandler) Hand(param map[string]interface{}) error {
|
||||||
|
var paramData,_ = json.Marshal(param)
|
||||||
|
common.CacheHelper.Cache(utils.GetRealKey(common2.USER)+param["id"].(string),string(paramData),0,true)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,34 @@
|
||||||
|
package im
|
||||||
|
|
||||||
|
import (
|
||||||
|
"github.com/go-netty/go-netty"
|
||||||
|
"snow-im/app/utils"
|
||||||
|
"snow-im/app/utils/netool"
|
||||||
|
)
|
||||||
|
|
||||||
|
var ImLogic ImHandler
|
||||||
|
|
||||||
|
func Init() {
|
||||||
|
ImLogic = ImHandler{
|
||||||
|
ImRouter,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
type ImHandler struct {
|
||||||
|
netool.Router
|
||||||
|
}
|
||||||
|
|
||||||
|
//消息处理
|
||||||
|
func (this *ImHandler) HandMessage(message []byte, connect netty.InboundContext) error {
|
||||||
|
defer func() {
|
||||||
|
if err := recover();err!=nil{
|
||||||
|
utils.Log(nil,"hand message",err)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
var msgId, msg = utils.UnpackMessage(message)
|
||||||
|
err := this.Router.HandleMsg(msgId, msg, connect)
|
||||||
|
if err != nil{
|
||||||
|
utils.Log(nil,"message err",err)
|
||||||
|
}
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,218 @@
|
||||||
|
package im
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"github.com/go-netty/go-netty"
|
||||||
|
redis2 "github.com/go-redis/redis"
|
||||||
|
"github.com/qit-team/snow-core/redis"
|
||||||
|
"go.mongodb.org/mongo-driver/bson"
|
||||||
|
"snow-im/app/constants/common"
|
||||||
|
"snow-im/app/constants/errorcode"
|
||||||
|
"snow-im/app/constants/msgid"
|
||||||
|
"snow-im/app/http/controllers"
|
||||||
|
"snow-im/app/http/entities"
|
||||||
|
"snow-im/app/models"
|
||||||
|
"snow-im/app/utils"
|
||||||
|
common2 "snow-im/app/utils/common"
|
||||||
|
"snow-im/app/utils/mq"
|
||||||
|
"snow-im/app/utils/netool"
|
||||||
|
"snow-im/app/utils/serialize"
|
||||||
|
"snow-im/config"
|
||||||
|
"strconv"
|
||||||
|
"strings"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
var ImRouter netool.Router
|
||||||
|
|
||||||
|
func init() {
|
||||||
|
ImRouter = netool.Router{}
|
||||||
|
ImRouter.AddRouter(msgid.SINGLE_MSG, SendMsgtoUser)
|
||||||
|
ImRouter.AddRouter(msgid.GROUP_MSG, SendMsgtoGroup)
|
||||||
|
ImRouter.AddRouter(msgid.MULT_MSG, SendMsgtoMultUser)
|
||||||
|
ImRouter.AddRouter(msgid.HEART_BEAT, HeartBeat)
|
||||||
|
ImRouter.AddRouter(msgid.AUTH, Auth)
|
||||||
|
ImRouter.AddRouter(msgid.ACK_MSG, Ack)
|
||||||
|
}
|
||||||
|
|
||||||
|
func getRsp(err error) entities.ImRsp {
|
||||||
|
var rsp entities.ImRsp
|
||||||
|
if err == nil {
|
||||||
|
rsp = entities.ImRsp{
|
||||||
|
Code: errorcode.Success,
|
||||||
|
Msg: errorcode.GetMsg(errorcode.Success, ""),
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
rsp = entities.ImRsp{
|
||||||
|
Code: errorcode.SystemError,
|
||||||
|
Msg: errorcode.GetMsg(errorcode.SystemError, ""),
|
||||||
|
}
|
||||||
|
utils.Log(nil, "err", err)
|
||||||
|
}
|
||||||
|
return rsp
|
||||||
|
}
|
||||||
|
|
||||||
|
//验证
|
||||||
|
func Auth(msg []byte, connect netty.InboundContext) error {
|
||||||
|
var authReq entities.AuthReq
|
||||||
|
err := serialize.SerializeTool.UnSerialize(config.GetConf().Serialize,msg,&authReq)
|
||||||
|
if err == nil {
|
||||||
|
err = controllers.Validate(authReq)
|
||||||
|
if err == nil {
|
||||||
|
uid, _ := redis.GetRedis().Get(utils.GetRealKey(common.TOKEN_PRE) + authReq.AppId + ":" + authReq.Token)
|
||||||
|
if uid != "" {
|
||||||
|
netool.GetConnManagger().SaveConnection(uid, connect.Channel())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
//单聊
|
||||||
|
func SendMsgtoUser(msg []byte, connect netty.InboundContext) error {
|
||||||
|
var singReq entities.SingTalkReq
|
||||||
|
utils.Log(nil, string(msg))
|
||||||
|
err := serialize.SerializeTool.UnSerialize(config.GetConf().Serialize,msg,&singReq)
|
||||||
|
if err == nil {
|
||||||
|
err = controllers.Validate(singReq)
|
||||||
|
if err == nil {
|
||||||
|
singReq.Id = utils.GenUniqId(singReq.SenderId)
|
||||||
|
singReq.Timestamp = time.Now().Unix()
|
||||||
|
var uid, _ = strconv.Atoi(singReq.To)
|
||||||
|
var msgFrom = common2.CacheHelper.GetCache(utils.GetRealKey(common.USER)+singReq.SenderId,false).(map[string]interface{})
|
||||||
|
singReq.Avatar = msgFrom["avatar"].(string)
|
||||||
|
singReq.SenderName = msgFrom["nick_name"].(string)
|
||||||
|
utils.Log(nil, "err", utils.GetUnderLineKey(common.SINGLE_TALK)+strconv.Itoa(uid%config.GetConf().Im.ImworkNum))
|
||||||
|
err = rabbitmq.MqManager.GetMqByName(common.MQ_KFK).Produce(utils.GetUnderLineKey(common.SINGLE_TALK)+strconv.Itoa(uid%config.GetConf().Im.ImworkNum), singReq, 0)
|
||||||
|
//生成会话
|
||||||
|
var msgData,_ = json.Marshal(singReq)
|
||||||
|
if err == nil {
|
||||||
|
var msg = models.Msg{}
|
||||||
|
msg.ParseFromParam(singReq.Msg)
|
||||||
|
msg.Conversion = utils.SortKeys([]string{singReq.SenderId,singReq.To})
|
||||||
|
msg.SaveMsg()
|
||||||
|
if singReq.RoomId == ""{
|
||||||
|
//优化管道操作
|
||||||
|
var pipline = common2.PikaTool.GetPipe()
|
||||||
|
var to = redis2.Z{
|
||||||
|
Score: float64(singReq.Timestamp),
|
||||||
|
Member: singReq.SenderId,
|
||||||
|
}
|
||||||
|
var from = redis2.Z{
|
||||||
|
Score: float64(singReq.Timestamp),
|
||||||
|
Member: singReq.SenderId,
|
||||||
|
}
|
||||||
|
pipline.ZAdd(utils.GetRealKey(common.CONVERSION)+singReq.To, to,from)
|
||||||
|
var msgFrom = common2.CacheHelper.GetCache(utils.GetRealKey(common.USER)+singReq.To,false).(map[string]interface{})
|
||||||
|
singReq.Avatar = msgFrom["avatar"].(string)
|
||||||
|
singReq.SenderName = msgFrom["nick_name"].(string)
|
||||||
|
msgDataFrom,_ := json.Marshal(singReq)
|
||||||
|
pipline.MSet(utils.GetRealKey(common.CONVERSION)+singReq.To+":"+singReq.SenderId, msgData,utils.GetRealKey(common.CONVERSION)+singReq.SenderId+":"+singReq.To, msgDataFrom)
|
||||||
|
pipline.Incr(utils.GetRealKey(common.CONVERSION)+singReq.To+":"+singReq.SenderId+":num")
|
||||||
|
pipline.Incr(utils.GetRealKey(common.CONVERSION)+singReq.SenderId+":"+singReq.To+":num")
|
||||||
|
pipline.Exec()
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
//群聊
|
||||||
|
func SendMsgtoGroup(msg []byte, connect netty.InboundContext) error {
|
||||||
|
var singReq entities.SingTalkReq
|
||||||
|
err := serialize.SerializeTool.UnSerialize(config.GetConf().Serialize,msg,&singReq)
|
||||||
|
if err == nil {
|
||||||
|
err = controllers.Validate(singReq)
|
||||||
|
if err == nil {
|
||||||
|
singReq.Id = utils.GenUniqId(singReq.SenderId)
|
||||||
|
singReq.Timestamp = time.Now().Unix()
|
||||||
|
var userInfo map[string]interface{}
|
||||||
|
var rs = common2.CacheHelper.GetCache(utils.GetRealKey(common.USER)+singReq.SenderId,false)
|
||||||
|
userInfo = rs.(map[string]interface{})
|
||||||
|
var room = models.Room{}
|
||||||
|
room = room.GetRoom(bson.M{"_id":singReq.RoomId})
|
||||||
|
singReq.SenderName = userInfo["nick_name"].(string)
|
||||||
|
singReq.Avatar = userInfo["avatar"].(string)
|
||||||
|
singReq.RoomAvatar = room.Avatar
|
||||||
|
singReq.RoomName = room.Name
|
||||||
|
var uid, _ = strconv.Atoi(singReq.RoomId)
|
||||||
|
err = rabbitmq.MqManager.GetMqByName(common.MQ_KFK).Produce(utils.GetUnderLineKey(common.GROUP_TALK)+strconv.Itoa(uid%config.GetConf().Im.ImGroupNum), singReq, 0)
|
||||||
|
if err == nil {
|
||||||
|
var msg = models.Msg{}
|
||||||
|
msg.ParseFromParam(singReq.Msg)
|
||||||
|
msg.Conversion = singReq.RoomId
|
||||||
|
msg.SaveMsg()
|
||||||
|
var pipie = common2.PikaTool.GetPipe()
|
||||||
|
var roomMembers = common2.PikaTool.SetGet(utils.GetRealKey(common.ROOM) + singReq.RoomId)
|
||||||
|
for _,v := range roomMembers{
|
||||||
|
var msgData,_ = json.Marshal(singReq)
|
||||||
|
var member =redis2.Z{
|
||||||
|
Member:singReq.RoomId,
|
||||||
|
Score:float64(singReq.Timestamp),
|
||||||
|
}
|
||||||
|
pipie.ZAdd(utils.GetRealKey(common.CONVERSION)+v, member)
|
||||||
|
pipie.Set(utils.GetRealKey(common.CONVERSION)+v+":"+singReq.RoomId,msgData, 0)
|
||||||
|
}
|
||||||
|
_,err = pipie.Exec()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
//组聊
|
||||||
|
func SendMsgtoMultUser(msg []byte, connect netty.InboundContext) error {
|
||||||
|
var singReq entities.SingTalkReq
|
||||||
|
err := serialize.SerializeTool.UnSerialize(config.GetConf().Serialize,msg,&singReq)
|
||||||
|
if err == nil {
|
||||||
|
err = controllers.Validate(singReq)
|
||||||
|
if err == nil {
|
||||||
|
singReq.Id = utils.GenUniqId(singReq.SenderId)
|
||||||
|
var userInfo map[string]string
|
||||||
|
json.Unmarshal([]byte(common2.PikaTool.HashGet(utils.GetRealKey("user_info"), singReq.SenderId)), &userInfo)
|
||||||
|
singReq.SenderName = userInfo["nick_name"]
|
||||||
|
singReq.Avatar = userInfo["avatar"]
|
||||||
|
var uids = strings.Split(singReq.To, ",")
|
||||||
|
var pipie = common2.PikaTool.GetPipe()
|
||||||
|
for _, v := range uids {
|
||||||
|
var uid, _ = strconv.Atoi(v)
|
||||||
|
singReq.To = v
|
||||||
|
err = rabbitmq.MqManager.GetMqByName(common.MQ_KFK).Produce(utils.GetUnderLineKey(common.SINGLE_TALK)+strconv.Itoa(uid%config.GetConf().Im.ImworkNum), singReq, 0)
|
||||||
|
//生成会话
|
||||||
|
if err == nil {
|
||||||
|
var member =redis2.Z{
|
||||||
|
Member:singReq.RoomId,
|
||||||
|
Score:float64(singReq.Timestamp),
|
||||||
|
}
|
||||||
|
pipie.ZAdd(utils.GetRealKey(common.CONVERSION)+v, member)
|
||||||
|
pipie.Set(utils.GetRealKey(common.CONVERSION)+singReq.To+":"+singReq.RoomId, msg,0)
|
||||||
|
}
|
||||||
|
_,err = pipie.Exec()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
//消息ack
|
||||||
|
func Ack(msg []byte, connect netty.InboundContext) error {
|
||||||
|
var req entities.MsgId
|
||||||
|
err := json.Unmarshal(msg, &req)
|
||||||
|
if err == nil {
|
||||||
|
|
||||||
|
}
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
//心跳
|
||||||
|
func HeartBeat(msg []byte, connect netty.InboundContext) error {
|
||||||
|
if connect.Channel().IsActive() {
|
||||||
|
netool.GetConnManagger().SetBeat(connect.Channel().ID())
|
||||||
|
connect.Write(utils.PackMsg([]byte("Ping"), msgid.HEART_BEAT))
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,12 @@
|
||||||
|
package mq
|
||||||
|
|
||||||
|
import (
|
||||||
|
"quenue/app/http/tcppool"
|
||||||
|
"quenue/app/utils"
|
||||||
|
)
|
||||||
|
|
||||||
|
// 调用充值
|
||||||
|
func OrderCharge(tag uint64, ch interface{}, msg []byte) error {
|
||||||
|
utils.Log(nil, "消息", tag, ch, string(msg))
|
||||||
|
return tcppool.TcpFactory.SendMsg(msg)
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,34 @@
|
||||||
|
package mq
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"quenue/app/constants/common"
|
||||||
|
mqs "quenue/app/utils/mq"
|
||||||
|
"quenue/config"
|
||||||
|
"strconv"
|
||||||
|
)
|
||||||
|
|
||||||
|
func init() {
|
||||||
|
//初始化工作队列
|
||||||
|
//utils.JobQueue = make(chan utils.Job, utils.MaxQueue)
|
||||||
|
}
|
||||||
|
func startQunue(name string, method interface{}, mqTp string, tp int, exhange string, i int) {
|
||||||
|
if tp == 0 {
|
||||||
|
go mqs.MqManager.GetMqByName(mqTp).Consume(name, method, i)
|
||||||
|
} else {
|
||||||
|
//go rabbitmq.DelayConsume(name, exhange, method.(func(tag uint64, ch *amqp.Channel, msg []byte)))
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
// 普通队列
|
||||||
|
func StartQunueServer() error {
|
||||||
|
if config.GetConf().StartQunue == 1 {
|
||||||
|
for i := 0; i < 1; i++ {
|
||||||
|
fmt.Println("对列" + strconv.Itoa(i))
|
||||||
|
startQunue(common.ORDER_RESEND_TOPICAL, OrderCharge, common.MQ_KFK_V2, 0, "", i) //单聊
|
||||||
|
}
|
||||||
|
}
|
||||||
|
select {}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,92 @@
|
||||||
|
package controllers
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"encoding/json"
|
||||||
|
"io/ioutil"
|
||||||
|
"net/http"
|
||||||
|
|
||||||
|
"quenue/app/constants/errorcode"
|
||||||
|
|
||||||
|
"github.com/gin-gonic/gin"
|
||||||
|
"github.com/qit-team/snow-core/log/logger"
|
||||||
|
"gopkg.in/go-playground/validator.v9"
|
||||||
|
)
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 成功时返回
|
||||||
|
*/
|
||||||
|
func Success(c *gin.Context, data interface{}) {
|
||||||
|
c.JSON(http.StatusOK, gin.H{
|
||||||
|
"code": errorcode.Success,
|
||||||
|
"message": "ok",
|
||||||
|
"request_uri": c.Request.URL.Path,
|
||||||
|
"data": data,
|
||||||
|
})
|
||||||
|
c.Abort()
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 失败时返回
|
||||||
|
*/
|
||||||
|
func Error(c *gin.Context, code int, msg ...string) {
|
||||||
|
message := ""
|
||||||
|
if len(msg) > 0 {
|
||||||
|
message = msg[0]
|
||||||
|
} else {
|
||||||
|
message = errorcode.GetMsg(code)
|
||||||
|
}
|
||||||
|
|
||||||
|
c.JSON(http.StatusOK, gin.H{
|
||||||
|
"code": code,
|
||||||
|
"message": message,
|
||||||
|
"request_uri": c.Request.URL.Path,
|
||||||
|
"data": make(map[string]string),
|
||||||
|
})
|
||||||
|
c.Abort()
|
||||||
|
}
|
||||||
|
|
||||||
|
func Error404(c *gin.Context) {
|
||||||
|
Error(c, errorcode.NotFound, "路由不存在")
|
||||||
|
}
|
||||||
|
|
||||||
|
func Error500(c *gin.Context) {
|
||||||
|
Error(c, errorcode.SystemError)
|
||||||
|
}
|
||||||
|
|
||||||
|
type HTTPError struct {
|
||||||
|
Code int `json:"code" example:"400"`
|
||||||
|
Message string `json:"message" example:"status bad request"`
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 将请求的body转换为request数据结构
|
||||||
|
* @param c
|
||||||
|
* @param request 传入request数据结构的指针 如 new(TestRequest)
|
||||||
|
*/
|
||||||
|
func GenRequest(c *gin.Context, request interface{}) (err error) {
|
||||||
|
body, err := ReadBody(c)
|
||||||
|
if err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
err = json.Unmarshal(body, request)
|
||||||
|
if (err == nil) {
|
||||||
|
validate := validator.New()
|
||||||
|
errValidate := validate.Struct(request)
|
||||||
|
if errValidate != nil {
|
||||||
|
logger.Error(c, "param_validator_exception:" + c.Request.URL.Path, errValidate)
|
||||||
|
return errValidate
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
//重复读取body
|
||||||
|
func ReadBody(c *gin.Context) (body []byte, err error) {
|
||||||
|
body, err = ioutil.ReadAll(c.Request.Body)
|
||||||
|
if err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
c.Request.Body = ioutil.NopCloser(bytes.NewBuffer(body))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,113 @@
|
||||||
|
package controllers
|
||||||
|
|
||||||
|
import (
|
||||||
|
"strconv"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"quenue/app/constants/errorcode"
|
||||||
|
"quenue/app/http/entities"
|
||||||
|
"quenue/app/http/formatters/bannerformatter"
|
||||||
|
"quenue/app/services/bannerservice"
|
||||||
|
"quenue/app/utils/httpclient"
|
||||||
|
|
||||||
|
"github.com/gin-gonic/gin"
|
||||||
|
"github.com/qit-team/snow-core/log/logger"
|
||||||
|
)
|
||||||
|
|
||||||
|
// hello示例
|
||||||
|
func HandleHello(c *gin.Context) {
|
||||||
|
logger.Debug(c, "hello", "test message")
|
||||||
|
client := httpclient.NewClient(c.Request.Context())
|
||||||
|
resposne, err := client.R().Get("https://www.baidu.com")
|
||||||
|
if err != nil {
|
||||||
|
Error(c, errorcode.SystemError, err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
logger.Info(c, "HandleHello", resposne.String())
|
||||||
|
Success(c, "hello world!")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// request和response的示例
|
||||||
|
// HandleTest godoc
|
||||||
|
// @Summary request和response的示例
|
||||||
|
// @Description request和response的示例
|
||||||
|
// @Tags snow
|
||||||
|
// @Accept json
|
||||||
|
// @Produce json
|
||||||
|
// @Param test body entities.TestRequest true "test request"
|
||||||
|
// @Success 200 {array} entities.TestResponse
|
||||||
|
// @Failure 400 {object} controllers.HTTPError
|
||||||
|
// @Failure 404 {object} controllers.HTTPError
|
||||||
|
// @Failure 500 {object} controllers.HTTPError
|
||||||
|
// @Router /test [post]
|
||||||
|
func HandleTest(c *gin.Context) {
|
||||||
|
request := new(entities.TestRequest)
|
||||||
|
err := GenRequest(c, request)
|
||||||
|
if err != nil {
|
||||||
|
Error(c, errorcode.ParamError)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
response := new(entities.TestResponse)
|
||||||
|
response.Name = request.Name
|
||||||
|
response.Url = request.Url
|
||||||
|
response.Id = time.Now().Unix()
|
||||||
|
Success(c, response)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// 测试数据库服务示例
|
||||||
|
func GetBannerList(c *gin.Context) {
|
||||||
|
pageStr := c.Query("page")
|
||||||
|
limitStr := c.DefaultQuery("limit", "20")
|
||||||
|
|
||||||
|
page, _ := strconv.Atoi(pageStr)
|
||||||
|
if page <= 0 {
|
||||||
|
page = 1
|
||||||
|
}
|
||||||
|
|
||||||
|
limit, _ := strconv.Atoi(limitStr)
|
||||||
|
if limit <= 0 {
|
||||||
|
limit = 20
|
||||||
|
}
|
||||||
|
|
||||||
|
list, err := bannerservice.GetListByPid(1, limit, page)
|
||||||
|
if err != nil {
|
||||||
|
Error500(c)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
data := map[string]interface{}{
|
||||||
|
"page": page,
|
||||||
|
"limit": limit,
|
||||||
|
"data": bannerformatter.FormatList(list),
|
||||||
|
}
|
||||||
|
|
||||||
|
Success(c, data)
|
||||||
|
}
|
||||||
|
|
||||||
|
// validator的示例
|
||||||
|
// HandleTestValidator godoc
|
||||||
|
// @Summary HandleTestValidator的示例
|
||||||
|
// @Description HandleTestValidator的示例
|
||||||
|
// @Tags snow
|
||||||
|
// @Accept json
|
||||||
|
// @Produce json
|
||||||
|
// @Param testValidator body entities.TestValidatorRequest true "example of validator"
|
||||||
|
// @Success 200 {array} entities.TestValidatorRequest
|
||||||
|
// @Failure 400 {object} controllers.HTTPError
|
||||||
|
// @Failure 404 {object} controllers.HTTPError
|
||||||
|
// @Failure 500 {object} controllers.HTTPError
|
||||||
|
// @Router /test_validator [post]
|
||||||
|
func HandleTestValidator(c *gin.Context) {
|
||||||
|
request := new(entities.TestValidatorRequest)
|
||||||
|
err := GenRequest(c, request)
|
||||||
|
if err != nil {
|
||||||
|
Error(c, errorcode.ParamError)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
Success(c, request)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,7 @@
|
||||||
|
package entities
|
||||||
|
|
||||||
|
type MqMessage struct {
|
||||||
|
Body map[string]interface{} `json:"body"`
|
||||||
|
Property map[string]interface{} `json:"property"`
|
||||||
|
Key string `json:"serial_number"`
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,43 @@
|
||||||
|
package entities
|
||||||
|
|
||||||
|
//请求数据结构
|
||||||
|
type TestRequest struct {
|
||||||
|
Name string `json:"name" example:"snow"`
|
||||||
|
Url string `json:"url" example:"github.com/qit-team/snow"`
|
||||||
|
}
|
||||||
|
|
||||||
|
//返回数据结构
|
||||||
|
type TestResponse struct {
|
||||||
|
Id int64 `json:"id" example:"1"`
|
||||||
|
Name string `json:"name" example:"snow"`
|
||||||
|
Url string `json:"url" example:"github.com/qit-team/snow"`
|
||||||
|
}
|
||||||
|
|
||||||
|
/*
|
||||||
|
* validator.v9文档
|
||||||
|
* 地址https://godoc.org/gopkg.in/go-playground/validator.v9
|
||||||
|
* 列了几个大家可能会用到的，如有遗漏，请看上面文档
|
||||||
|
*/
|
||||||
|
|
||||||
|
//请求数据结构
|
||||||
|
type TestValidatorRequest struct {
|
||||||
|
//tips，因为组件required不管是没传值或者传 0 or "" 都通过不了，但是如果用指针类型，那么0就是0，而nil无法通过校验
|
||||||
|
Id *int64 `json:"id" validate:"required" example:"1"`
|
||||||
|
Age int `json:"age" validate:"required,gte=0,lte=130" example:"20"`
|
||||||
|
Name *string `json:"name" validate:"required" example:"snow"`
|
||||||
|
Email string `json:"email" validate:"required,email" example:"snow@github.com"`
|
||||||
|
Url string `json:"url" validate:"required" example:"github.com/qit-team/snow"`
|
||||||
|
Mobile string `json:"mobile" validate:"required" example:"snow"`
|
||||||
|
RangeNum int `json:"range_num" validate:"max=10,min=1" example:"3"`
|
||||||
|
TestNum *int `json:"test_num" validate:"required,oneof=5 7 9" example:"7"`
|
||||||
|
Content *string `json:"content" example:"snow"`
|
||||||
|
Addresses []*Address `json:"addresses" validate:"required,dive,required" `
|
||||||
|
}
|
||||||
|
|
||||||
|
// Address houses a users address information
|
||||||
|
type Address struct {
|
||||||
|
Street string `json:"street" validate:"required" example:"huandaodonglu"`
|
||||||
|
City string `json:"city" validate:"required" example:"xiamen"`
|
||||||
|
Planet string `json:"planet" validate:"required" example:"snow"`
|
||||||
|
Phone string `json:"phone" validate:"required" example:"snow"`
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,34 @@
|
||||||
|
package bannerformatter
|
||||||
|
|
||||||
|
import (
|
||||||
|
"quenue/app/models/bannermodel"
|
||||||
|
)
|
||||||
|
|
||||||
|
type BannerFormatter struct {
|
||||||
|
Id int `json:"id"`
|
||||||
|
Title string `json:"title"`
|
||||||
|
Img string `json:"image"`
|
||||||
|
Url string `json:"url"`
|
||||||
|
}
|
||||||
|
|
||||||
|
func FormatList(bannerList []*bannermodel.Banner) (res []*BannerFormatter) {
|
||||||
|
res = make([]*BannerFormatter, len(bannerList))
|
||||||
|
|
||||||
|
for k, banner := range bannerList {
|
||||||
|
one := FormatOne(banner)
|
||||||
|
res[k] = one
|
||||||
|
}
|
||||||
|
|
||||||
|
return res
|
||||||
|
}
|
||||||
|
|
||||||
|
//单条消息的格式化，
|
||||||
|
func FormatOne(banner *bannermodel.Banner) (res *BannerFormatter) {
|
||||||
|
res = &BannerFormatter{
|
||||||
|
Id: int(banner.Id),
|
||||||
|
Title: banner.Title,
|
||||||
|
Img: banner.ImageUrl,
|
||||||
|
Url: banner.Url,
|
||||||
|
}
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,45 @@
|
||||||
|
package bannerformatter
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"quenue/app/models/bannermodel"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TesFormatOne(t *testing.T) {
|
||||||
|
a := &bannermodel.Banner{
|
||||||
|
Id: 1,
|
||||||
|
Title: "test",
|
||||||
|
ImageUrl: "http://x/1.jpg",
|
||||||
|
Url: "http://x",
|
||||||
|
Status: "1",
|
||||||
|
}
|
||||||
|
b := FormatOne(a)
|
||||||
|
if b.Title != a.Title || b.Img != a.ImageUrl || b.Url != a.Url {
|
||||||
|
t.Error("FormatOne not same")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TesFormatList(t *testing.T) {
|
||||||
|
a := make([]*bannermodel.Banner, 2)
|
||||||
|
a[0] = &bannermodel.Banner{
|
||||||
|
Id: 1,
|
||||||
|
Title: "test",
|
||||||
|
ImageUrl: "http://x1/1.jpg",
|
||||||
|
Url: "http://x1",
|
||||||
|
Status: "1",
|
||||||
|
}
|
||||||
|
a[1] = &bannermodel.Banner{
|
||||||
|
Id: 2,
|
||||||
|
Title: "test2",
|
||||||
|
ImageUrl: "http://x/2.jpg",
|
||||||
|
Url: "http://x2",
|
||||||
|
Status: "2",
|
||||||
|
}
|
||||||
|
b := FormatList(a)
|
||||||
|
for k, v := range b {
|
||||||
|
if v.Title != a[k].Title || v.Img != a[k].ImageUrl || v.Url != a[k].Url {
|
||||||
|
t.Error("FormatList not same")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,77 @@
|
||||||
|
package metric
|
||||||
|
|
||||||
|
import (
|
||||||
|
"net/http"
|
||||||
|
|
||||||
|
"quenue/app/utils/metric"
|
||||||
|
|
||||||
|
"github.com/prometheus/client_golang/prometheus"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
HOST = "host"
|
||||||
|
PATH = "path" // 路径
|
||||||
|
METHOD = "method" // 方法
|
||||||
|
CODE = "code" // 错误码
|
||||||
|
|
||||||
|
// metric
|
||||||
|
ALL_REQ_TOTAL_COUNT = "all_req_total_count" // 所有URL总请求数
|
||||||
|
ALL_REQ_COST_TIME = "all_req_cost_time" // 所有URL请求耗时
|
||||||
|
|
||||||
|
REQ_TOTAL_COUNT = "req_total_count" // 每个URL总请求数
|
||||||
|
REQ_COST_TIME = "req_cost_time" // 每个URL请求耗时
|
||||||
|
)
|
||||||
|
|
||||||
|
func init() {
|
||||||
|
metric.RegisterCollector(reqTotalCounter, reqCostTimeObserver, allReqTotalCounter, allReqCostTimeObserver)
|
||||||
|
}
|
||||||
|
|
||||||
|
var (
|
||||||
|
reqTotalCounter = prometheus.NewCounterVec(prometheus.CounterOpts{
|
||||||
|
Name: REQ_TOTAL_COUNT,
|
||||||
|
}, []string{PATH, METHOD})
|
||||||
|
|
||||||
|
reqCostTimeObserver = prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
||||||
|
Name: REQ_COST_TIME,
|
||||||
|
Buckets: []float64{
|
||||||
|
100,
|
||||||
|
200,
|
||||||
|
500,
|
||||||
|
1000,
|
||||||
|
3000,
|
||||||
|
5000,
|
||||||
|
},
|
||||||
|
}, []string{PATH, METHOD})
|
||||||
|
|
||||||
|
allReqTotalCounter = prometheus.NewCounterVec(prometheus.CounterOpts{
|
||||||
|
Name: ALL_REQ_TOTAL_COUNT,
|
||||||
|
}, []string{HOST})
|
||||||
|
|
||||||
|
allReqCostTimeObserver = prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
||||||
|
Name: ALL_REQ_COST_TIME,
|
||||||
|
Buckets: []float64{
|
||||||
|
100,
|
||||||
|
200,
|
||||||
|
500,
|
||||||
|
1000,
|
||||||
|
3000,
|
||||||
|
5000,
|
||||||
|
},
|
||||||
|
}, []string{HOST})
|
||||||
|
)
|
||||||
|
|
||||||
|
func AddReqCount(req *http.Request) {
|
||||||
|
reqTotalCounter.WithLabelValues(req.URL.Path, req.Method).Inc()
|
||||||
|
}
|
||||||
|
|
||||||
|
func CollectReqCostTime(req *http.Request, ms int64) {
|
||||||
|
reqCostTimeObserver.WithLabelValues(req.URL.Path, req.Method).Observe(float64(ms))
|
||||||
|
}
|
||||||
|
|
||||||
|
func AddAllReqCount(req *http.Request) {
|
||||||
|
allReqTotalCounter.WithLabelValues(req.Host).Inc()
|
||||||
|
}
|
||||||
|
|
||||||
|
func CollectAllReqCostTime(req *http.Request, ms int64) {
|
||||||
|
allReqCostTimeObserver.WithLabelValues(req.Host).Observe(float64(ms))
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,22 @@
|
||||||
|
package middlewares
|
||||||
|
|
||||||
|
import (
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"quenue/app/http/metric"
|
||||||
|
|
||||||
|
"github.com/gin-gonic/gin"
|
||||||
|
)
|
||||||
|
|
||||||
|
func CollectMetric() gin.HandlerFunc {
|
||||||
|
return func(ctx *gin.Context) {
|
||||||
|
start := time.Now()
|
||||||
|
ctx.Next()
|
||||||
|
dur := time.Now().Sub(start).Milliseconds()
|
||||||
|
|
||||||
|
metric.AddAllReqCount(ctx.Request)
|
||||||
|
metric.CollectAllReqCostTime(ctx.Request, dur)
|
||||||
|
metric.AddReqCount(ctx.Request)
|
||||||
|
metric.CollectReqCostTime(ctx.Request, dur)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,50 @@
|
||||||
|
package middlewares
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
syslog "log"
|
||||||
|
"net/http/httputil"
|
||||||
|
"runtime/debug"
|
||||||
|
|
||||||
|
"quenue/app/constants/logtype"
|
||||||
|
"quenue/config"
|
||||||
|
|
||||||
|
"github.com/gin-gonic/gin"
|
||||||
|
"github.com/qit-team/snow-core/log/logger"
|
||||||
|
)
|
||||||
|
|
||||||
|
func ServerRecovery() gin.HandlerFunc {
|
||||||
|
return func(c *gin.Context) {
|
||||||
|
|
||||||
|
defer func() {
|
||||||
|
if err := recover(); err != nil {
|
||||||
|
httpRequest, _ := httputil.DumpRequest(c.Request, false)
|
||||||
|
msg := map[string]interface{}{
|
||||||
|
"error": err,
|
||||||
|
"request": string(httpRequest),
|
||||||
|
"stack": string(debug.Stack()),
|
||||||
|
}
|
||||||
|
msgJson, _ := json.Marshal(msg)
|
||||||
|
logger.GetLogger().Error(string(msgJson), logtype.GoPanic, c)
|
||||||
|
|
||||||
|
if config.IsDebug() {
|
||||||
|
//本地开发 debug 模式开启时输出错误信息到shell
|
||||||
|
syslog.Println(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
c.JSON(500, gin.H{
|
||||||
|
"code": 500,
|
||||||
|
"msg": "system error",
|
||||||
|
"request_uri": c.Request.URL.Path,
|
||||||
|
"data": make(map[string]string),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
|
//before request
|
||||||
|
|
||||||
|
c.Next()
|
||||||
|
|
||||||
|
//after request
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,55 @@
|
||||||
|
package middlewares
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"strconv"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"quenue/app/http/trace"
|
||||||
|
|
||||||
|
"github.com/SkyAPM/go2sky"
|
||||||
|
"github.com/SkyAPM/go2sky/propagation"
|
||||||
|
v3 "github.com/SkyAPM/go2sky/reporter/grpc/language-agent"
|
||||||
|
"github.com/gin-gonic/gin"
|
||||||
|
"github.com/qit-team/snow-core/log/logger"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
componentIDGOHttpServer = 5004
|
||||||
|
)
|
||||||
|
|
||||||
|
func Trace() gin.HandlerFunc {
|
||||||
|
return func(c *gin.Context) {
|
||||||
|
tracer, err := trace.Tracer()
|
||||||
|
if err != nil {
|
||||||
|
logger.Error(c, "Trace", err.Error())
|
||||||
|
c.Next()
|
||||||
|
return
|
||||||
|
}
|
||||||
|
r := c.Request
|
||||||
|
operationName := fmt.Sprintf("/%s%s", r.Method, r.URL.Path)
|
||||||
|
span, ctx, err := tracer.CreateEntrySpan(c, operationName, func() (string, error) {
|
||||||
|
// 从http头部捞取上一层的调用链信息, 当前使用v3版本的协议
|
||||||
|
// https://github.com/apache/skywalking/blob/master/docs/en/protocols/Skywalking-Cross-Process-Propagation-Headers-Protocol-v3.md
|
||||||
|
return r.Header.Get(propagation.Header), nil
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
logger.Error(c, "Trace", err.Error())
|
||||||
|
c.Next()
|
||||||
|
return
|
||||||
|
}
|
||||||
|
span.SetComponent(componentIDGOHttpServer)
|
||||||
|
// 可以自定义tag
|
||||||
|
span.Tag(go2sky.TagHTTPMethod, r.Method)
|
||||||
|
span.Tag(go2sky.TagURL, fmt.Sprintf("%s%s", r.Host, r.URL.Path))
|
||||||
|
span.SetSpanLayer(v3.SpanLayer_Http)
|
||||||
|
c.Request = c.Request.WithContext(ctx)
|
||||||
|
c.Next()
|
||||||
|
code := c.Writer.Status()
|
||||||
|
if code >= 400 {
|
||||||
|
span.Error(time.Now(), fmt.Sprintf("Error on handling request, statusCode: %d", code))
|
||||||
|
}
|
||||||
|
span.Tag(go2sky.TagStatusCode, strconv.Itoa(code))
|
||||||
|
span.End()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,55 @@
|
||||||
|
package routes
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 配置路由
|
||||||
|
*/
|
||||||
|
import (
|
||||||
|
"quenue/app/http/controllers"
|
||||||
|
"quenue/app/http/middlewares"
|
||||||
|
"quenue/app/http/trace"
|
||||||
|
"quenue/app/utils/metric"
|
||||||
|
"quenue/config"
|
||||||
|
|
||||||
|
"github.com/gin-gonic/gin"
|
||||||
|
"github.com/qit-team/snow-core/http/middleware"
|
||||||
|
"github.com/qit-team/snow-core/log/logger"
|
||||||
|
"github.com/swaggo/gin-swagger"
|
||||||
|
"github.com/swaggo/gin-swagger/swaggerFiles"
|
||||||
|
)
|
||||||
|
|
||||||
|
//api路由配置
|
||||||
|
func RegisterRoute(router *gin.Engine) {
|
||||||
|
//middleware: 服务错误处理 => 生成请求id => access log
|
||||||
|
router.Use(middlewares.ServerRecovery(), middleware.GenRequestId, middleware.GenContextKit, middleware.AccessLog())
|
||||||
|
|
||||||
|
if config.GetConf().PrometheusCollectEnable && config.IsEnvEqual(config.ProdEnv) {
|
||||||
|
router.Use(middlewares.CollectMetric())
|
||||||
|
metric.Init(metric.EnableRuntime(), metric.EnableProcess())
|
||||||
|
metricHandler := metric.Handler()
|
||||||
|
router.GET("/metrics", func(ctx *gin.Context) {
|
||||||
|
metricHandler.ServeHTTP(ctx.Writer, ctx.Request)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
if len(config.GetConf().SkyWalkingOapServer) > 0 && config.IsEnvEqual(config.ProdEnv) {
|
||||||
|
err := trace.InitTracer(config.GetConf().ServiceName, config.GetConf().SkyWalkingOapServer)
|
||||||
|
if err != nil {
|
||||||
|
logger.Error(nil, "InitTracer", err.Error())
|
||||||
|
} else {
|
||||||
|
router.Use(middlewares.Trace())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
router.NoRoute(controllers.Error404)
|
||||||
|
router.GET("/hello", controllers.HandleHello)
|
||||||
|
router.POST("/test", controllers.HandleTest)
|
||||||
|
router.POST("/test_validator", controllers.HandleTestValidator)
|
||||||
|
|
||||||
|
//api版本
|
||||||
|
v1 := router.Group("/v1")
|
||||||
|
{
|
||||||
|
v1.GET("/banner_list", controllers.GetBannerList)
|
||||||
|
}
|
||||||
|
|
||||||
|
router.GET("/swagger/*any", ginSwagger.WrapHandler(swaggerFiles.Handler))
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,106 @@
|
||||||
|
package tcppool
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bufio"
|
||||||
|
"github.com/nange/easypool"
|
||||||
|
_ "github.com/nange/easypool"
|
||||||
|
"io"
|
||||||
|
"net"
|
||||||
|
"quenue/app/utils"
|
||||||
|
"quenue/config"
|
||||||
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
var lock sync.Once
|
||||||
|
var TcpPoolFactory = &TcpPool{Full: new(int32)}
|
||||||
|
|
||||||
|
type TcpPool struct {
|
||||||
|
client easypool.Pool
|
||||||
|
Full *int32
|
||||||
|
lastTime int64
|
||||||
|
isDie bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func (t *TcpPool) initPool(port string) *TcpPool {
|
||||||
|
lock.Do(func() {
|
||||||
|
factory := func() (net.Conn, error) {
|
||||||
|
var conn, err = net.Dial("tcp", "192.168.110.50:"+port)
|
||||||
|
return conn, err
|
||||||
|
}
|
||||||
|
config := &easypool.PoolConfig{
|
||||||
|
InitialCap: 1,
|
||||||
|
MaxCap: 2,
|
||||||
|
MaxIdle: 1,
|
||||||
|
Idletime: 30 * time.Second,
|
||||||
|
MaxLifetime: 10 * time.Minute,
|
||||||
|
Factory: factory,
|
||||||
|
}
|
||||||
|
|
||||||
|
pool, err := easypool.NewHeapPool(config)
|
||||||
|
if err == nil {
|
||||||
|
t.isDie = false
|
||||||
|
t.client = pool
|
||||||
|
} else {
|
||||||
|
utils.Log(nil, "tcp err", err)
|
||||||
|
time.Sleep(3 * time.Second)
|
||||||
|
}
|
||||||
|
|
||||||
|
})
|
||||||
|
return t
|
||||||
|
}
|
||||||
|
func (t *TcpPool) handRead(conn net.Conn) {
|
||||||
|
//defer func() {
|
||||||
|
// if err := recover(); err != nil {
|
||||||
|
// utils.Log(nil, "tcp read err", err)
|
||||||
|
// }
|
||||||
|
//}()
|
||||||
|
for {
|
||||||
|
if time.Now().Unix()-t.lastTime > 10 {
|
||||||
|
//t.isDie = true
|
||||||
|
//t.client.Close()
|
||||||
|
//t.client = nil
|
||||||
|
//return
|
||||||
|
}
|
||||||
|
reader := bufio.NewReader(conn)
|
||||||
|
var buffer [256]byte
|
||||||
|
// 持续读取数据
|
||||||
|
n, err := reader.Read(buffer[:])
|
||||||
|
if err == io.EOF {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
t.isDie = true
|
||||||
|
if t.client != nil {
|
||||||
|
t.client.Close()
|
||||||
|
t.client = nil
|
||||||
|
}
|
||||||
|
return
|
||||||
|
utils.Log(nil, "Error reading data:", err)
|
||||||
|
}
|
||||||
|
recvStr := string(buffer[:n])
|
||||||
|
if recvStr == "1" {
|
||||||
|
atomic.AddInt32(t.Full, 1)
|
||||||
|
} else if recvStr == "ping" {
|
||||||
|
conn.Write([]byte("pong"))
|
||||||
|
t.lastTime = time.Now().Unix()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
func (t *TcpPool) watch(conn net.Conn) {
|
||||||
|
go t.handRead(conn)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (t *TcpPool) SendMsg(msg []byte) error {
|
||||||
|
conn, err := t.initPool(config.GetConf().OrderPort).client.Get()
|
||||||
|
if err != nil {
|
||||||
|
utils.Log(nil, "get tcp err", err)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
//var data, _ = json.Marshal(msg)
|
||||||
|
t.watch(conn)
|
||||||
|
_, err = conn.Write(msg)
|
||||||
|
conn.Close()
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,103 @@
|
||||||
|
package tcppool
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bufio"
|
||||||
|
"fmt"
|
||||||
|
"io"
|
||||||
|
"net"
|
||||||
|
"quenue/app/utils"
|
||||||
|
"quenue/config"
|
||||||
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
var (
|
||||||
|
full int32 = 0
|
||||||
|
TcpFactory = TcpHelper{Full: &full}
|
||||||
|
lockSingle sync.Once
|
||||||
|
)
|
||||||
|
|
||||||
|
type TcpHelper struct {
|
||||||
|
client net.Conn
|
||||||
|
lastTime int64
|
||||||
|
Full *int32
|
||||||
|
}
|
||||||
|
|
||||||
|
func (t *TcpHelper) Init(port string) *TcpHelper {
|
||||||
|
lockSingle.Do(func() {
|
||||||
|
var conn, err = net.Dial("tcp", "192.168.110.50:"+port)
|
||||||
|
if err == nil {
|
||||||
|
t.client = conn
|
||||||
|
//t.watch(t.client)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
return t
|
||||||
|
}
|
||||||
|
func (t *TcpHelper) handRead(conn net.Conn) {
|
||||||
|
//defer func() {
|
||||||
|
// if err := recover(); err != nil {
|
||||||
|
// utils.Log(nil, "tcp read err", err)
|
||||||
|
// }
|
||||||
|
//}()
|
||||||
|
for {
|
||||||
|
if time.Now().Unix()-t.lastTime > 10 {
|
||||||
|
//t.isDie = true
|
||||||
|
//t.client.Close()
|
||||||
|
//t.client = nil
|
||||||
|
//return
|
||||||
|
}
|
||||||
|
fmt.Println("read")
|
||||||
|
reader := bufio.NewReader(conn)
|
||||||
|
var buffer [256]byte
|
||||||
|
// 持续读取数据
|
||||||
|
n, err := reader.Read(buffer[:])
|
||||||
|
if err == io.EOF {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
//if t.client != nil {
|
||||||
|
// t.client.Close()
|
||||||
|
// t.client = nil
|
||||||
|
//}
|
||||||
|
utils.Log(nil, "Error reading data:", err)
|
||||||
|
continue
|
||||||
|
|
||||||
|
}
|
||||||
|
recvStr := string(buffer[:n])
|
||||||
|
if recvStr == "1" {
|
||||||
|
atomic.AddInt32(t.Full, 1)
|
||||||
|
} else if recvStr == "ping" {
|
||||||
|
conn.Write([]byte("pong"))
|
||||||
|
t.lastTime = time.Now().Unix()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
select {}
|
||||||
|
}
|
||||||
|
func (t *TcpHelper) SendMsg(msg []byte) error {
|
||||||
|
_, err := t.Init(config.GetConf().OrderPort).client.Write(msg)
|
||||||
|
|
||||||
|
var buffer [256]byte
|
||||||
|
// 持续读取数据
|
||||||
|
n, err := t.client.Read(buffer[:])
|
||||||
|
if err == nil {
|
||||||
|
if n > 0 {
|
||||||
|
recvStr := string(buffer[:n])
|
||||||
|
fmt.Println("结果：recvStr:", recvStr)
|
||||||
|
if recvStr == "1" {
|
||||||
|
fmt.Println("满了")
|
||||||
|
atomic.AddInt32(t.Full, 1)
|
||||||
|
} else if recvStr == "2" {
|
||||||
|
atomic.AddInt32(t.Full, -1)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
func (t *TcpHelper) Close(conn net.Conn) {
|
||||||
|
t.client.Close()
|
||||||
|
}
|
||||||
|
func (t *TcpHelper) watch(conn net.Conn) {
|
||||||
|
go t.handRead(conn)
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,42 @@
|
||||||
|
package trace
|
||||||
|
|
||||||
|
import (
|
||||||
|
"sync"
|
||||||
|
|
||||||
|
"quenue/config"
|
||||||
|
|
||||||
|
"github.com/SkyAPM/go2sky"
|
||||||
|
"github.com/SkyAPM/go2sky/reporter"
|
||||||
|
)
|
||||||
|
|
||||||
|
var (
|
||||||
|
tracer *go2sky.Tracer
|
||||||
|
lock sync.Mutex
|
||||||
|
)
|
||||||
|
|
||||||
|
func Tracer() (*go2sky.Tracer, error) {
|
||||||
|
if tracer == nil {
|
||||||
|
// 有err, 不适合用sync.Once做单例
|
||||||
|
lock.Lock()
|
||||||
|
defer lock.Unlock()
|
||||||
|
if tracer == nil {
|
||||||
|
err := InitTracer(config.GetConf().ServiceName, config.GetConf().SkyWalkingOapServer)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return tracer, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func InitTracer(serviceName, skyWalkingOapServer string) error {
|
||||||
|
report, err := reporter.NewGRPCReporter(skyWalkingOapServer)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
tracer, err = go2sky.NewTracer(serviceName, go2sky.WithReporter(report))
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,66 @@
|
||||||
|
package basejob
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"sync"
|
||||||
|
|
||||||
|
"github.com/qit-team/work"
|
||||||
|
)
|
||||||
|
|
||||||
|
var (
|
||||||
|
jb *work.Job
|
||||||
|
register func(job *work.Job)
|
||||||
|
mu sync.RWMutex
|
||||||
|
)
|
||||||
|
|
||||||
|
func SetJob(job *work.Job) {
|
||||||
|
if jb == nil {
|
||||||
|
jb = job
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func SetJobRegister(r func(*work.Job)) {
|
||||||
|
register = r
|
||||||
|
}
|
||||||
|
|
||||||
|
func GetJob() *work.Job {
|
||||||
|
if jb == nil {
|
||||||
|
if register != nil {
|
||||||
|
mu.Lock()
|
||||||
|
defer mu.Unlock()
|
||||||
|
jb = work.New()
|
||||||
|
register(jb)
|
||||||
|
} else {
|
||||||
|
panic("job register is nil")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return jb
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 消息入队 -- 原始message
|
||||||
|
*/
|
||||||
|
func Enqueue(ctx context.Context, topic string, message string, args ...interface{}) (isOk bool, err error) {
|
||||||
|
return GetJob().Enqueue(ctx, topic, message, args...)
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 消息入队 -- Task数据结构
|
||||||
|
*/
|
||||||
|
func EnqueueWithTask(ctx context.Context, topic string, task work.Task, args ...interface{}) (isOk bool, err error) {
|
||||||
|
return GetJob().EnqueueWithTask(ctx, topic, task, args...)
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 消息批量入队 -- 原始message
|
||||||
|
*/
|
||||||
|
func BatchEnqueue(ctx context.Context, topic string, messages []string, args ...interface{}) (isOk bool, err error) {
|
||||||
|
return GetJob().BatchEnqueue(ctx, topic, messages, args...)
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 消息批量入队 -- Task数据结构
|
||||||
|
*/
|
||||||
|
func BatchEnqueueWithTask(ctx context.Context, topic string, tasks []work.Task, args ...interface{}) (isOk bool, err error) {
|
||||||
|
return GetJob().BatchEnqueueWithTask(ctx, topic, tasks, args...)
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,56 @@
|
||||||
|
package jobs
|
||||||
|
|
||||||
|
import (
|
||||||
|
"strings"
|
||||||
|
|
||||||
|
"quenue/app/jobs/basejob"
|
||||||
|
"quenue/config"
|
||||||
|
|
||||||
|
"github.com/qit-team/snow-core/log/logger"
|
||||||
|
"github.com/qit-team/snow-core/queue"
|
||||||
|
"github.com/qit-team/snow-core/redis"
|
||||||
|
"github.com/qit-team/work"
|
||||||
|
)
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 配置队列任务
|
||||||
|
*/
|
||||||
|
func RegisterWorker(job *work.Job) {
|
||||||
|
basejob.SetJob(job)
|
||||||
|
|
||||||
|
//设置worker的任务投递回调函数
|
||||||
|
job.AddFunc("topic-test", test)
|
||||||
|
//设置worker的任务投递回调函数，和并发数
|
||||||
|
job.AddFunc("topic-test1", test, 2)
|
||||||
|
//使用worker结构进行注册
|
||||||
|
job.AddWorker("topic-test2", &work.Worker{Call: work.MyWorkerFunc(test), MaxConcurrency: 1})
|
||||||
|
|
||||||
|
RegisterQueueDriver(job)
|
||||||
|
SetOptions(job)
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 给topic注册对应的队列服务
|
||||||
|
*/
|
||||||
|
func RegisterQueueDriver(job *work.Job) {
|
||||||
|
//设置队列服务，需要实现work.Queue接口的方法
|
||||||
|
q := queue.GetQueue(redis.SingletonMain, queue.DriverTypeRedis)
|
||||||
|
//针对topic设置相关的queue
|
||||||
|
job.AddQueue(q, "topic-test1", "topic-test2")
|
||||||
|
//设置默认的queue, 没有设置过的topic会使用默认的queue
|
||||||
|
job.AddQueue(q)
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 设置配置参数
|
||||||
|
*/
|
||||||
|
func SetOptions(job *work.Job) {
|
||||||
|
//设置logger，需要实现work.Logger接口的方法
|
||||||
|
job.SetLogger(logger.GetLogger())
|
||||||
|
|
||||||
|
//设置启用的topic，未设置表示启用全部注册过topic
|
||||||
|
if config.GetOptions().Queue != "" {
|
||||||
|
topics := strings.Split(config.GetOptions().Queue, ",")
|
||||||
|
job.SetEnableTopics(topics...)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,24 @@
|
||||||
|
package jobs
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/qit-team/work"
|
||||||
|
)
|
||||||
|
|
||||||
|
func test(task work.Task) (work.TaskResult) {
|
||||||
|
time.Sleep(time.Millisecond * 5)
|
||||||
|
s, err := work.JsonEncode(task)
|
||||||
|
if err != nil {
|
||||||
|
//work.StateFailed 不会进行ack确认
|
||||||
|
//work.StateFailedWithAck 会进行actk确认
|
||||||
|
//return work.TaskResult{Id: task.Id, State: work.StateFailed}
|
||||||
|
return work.TaskResult{Id: task.Id, State: work.StateFailedWithAck}
|
||||||
|
} else {
|
||||||
|
//work.StateSucceed 会进行ack确认
|
||||||
|
fmt.Println("do task", s)
|
||||||
|
return work.TaskResult{Id: task.Id, State: work.StateSucceed}
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,57 @@
|
||||||
|
package bannermodel
|
||||||
|
|
||||||
|
import (
|
||||||
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/qit-team/snow-core/db"
|
||||||
|
)
|
||||||
|
|
||||||
|
var (
|
||||||
|
once sync.Once
|
||||||
|
m *bannerModel
|
||||||
|
)
|
||||||
|
/**
|
||||||
|
* Banner实体
|
||||||
|
*/
|
||||||
|
type Banner struct {
|
||||||
|
Id int64 `xorm:"pk autoincr"` //注：使用getOne 或者ID() 需要设置主键
|
||||||
|
Pid int
|
||||||
|
Title string
|
||||||
|
ImageUrl string `xorm:"'img_url'"`
|
||||||
|
Url string
|
||||||
|
Status string
|
||||||
|
CreatedAt time.Time
|
||||||
|
UpdatedAt time.Time
|
||||||
|
DeletedAt time.Time `xorm:"deleted"` //此特性会激发软删除
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 表名规则
|
||||||
|
* @wiki http://gobook.io/read/github.com/go-xorm/manual-zh-CN/chapter-02/3.tags.html
|
||||||
|
*/
|
||||||
|
func (m *Banner) TableName() string {
|
||||||
|
return "banner"
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 私有化，防止被外部new
|
||||||
|
*/
|
||||||
|
type bannerModel struct {
|
||||||
|
db.Model //组合基础Model，集成基础Model的属性和方法
|
||||||
|
}
|
||||||
|
|
||||||
|
//单例模式
|
||||||
|
func GetInstance() *bannerModel {
|
||||||
|
once.Do(func() {
|
||||||
|
m = new(bannerModel)
|
||||||
|
//m.DiName = "" //设置数据库实例连接，默认db.SingletonMain
|
||||||
|
})
|
||||||
|
return m
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *bannerModel) GetListByPid(pid int, limits ...int) (banners []*Banner, err error) {
|
||||||
|
banners = make([]*Banner, 0)
|
||||||
|
err = m.GetList(&banners, "pid = ?", []interface{}{pid}, limits)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,54 @@
|
||||||
|
package bannermodel
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
_ "github.com/go-sql-driver/mysql"
|
||||||
|
"github.com/qit-team/snow-core/config"
|
||||||
|
"github.com/qit-team/snow-core/db"
|
||||||
|
"github.com/qit-team/snow-core/utils"
|
||||||
|
)
|
||||||
|
|
||||||
|
func init() {
|
||||||
|
m := config.DbBaseConfig{
|
||||||
|
Host: "127.0.0.1",
|
||||||
|
Port: 3306,
|
||||||
|
User: "root",
|
||||||
|
Password: "123456",
|
||||||
|
DBName: "test",
|
||||||
|
}
|
||||||
|
dbConf := config.DbConfig{
|
||||||
|
Driver: "mysql",
|
||||||
|
Master: m,
|
||||||
|
}
|
||||||
|
|
||||||
|
err := db.Pr.Register("db", dbConf, true)
|
||||||
|
if err != nil {
|
||||||
|
fmt.Println(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestGetOne(t *testing.T) {
|
||||||
|
bannerModel := GetInstance()
|
||||||
|
banner := new(Banner)
|
||||||
|
res, err := bannerModel.GetOne(1, banner)
|
||||||
|
if err != nil {
|
||||||
|
t.Error(err)
|
||||||
|
} else if res != true {
|
||||||
|
t.Error("missing banner data")
|
||||||
|
} else if banner.Id == 0 {
|
||||||
|
t.Error("missing banner data")
|
||||||
|
}
|
||||||
|
fmt.Println(utils.JsonEncode(banner))
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestGetList(t *testing.T) {
|
||||||
|
bannerModel := GetInstance()
|
||||||
|
banners := make([]*Banner, 0)
|
||||||
|
err := bannerModel.GetList(&banners, "pid >= ?", []interface{}{1}, []int{10}, "status desc, id desc")
|
||||||
|
if err != nil {
|
||||||
|
t.Error(err)
|
||||||
|
}
|
||||||
|
fmt.Println(utils.JsonEncode(banners))
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,25 @@
|
||||||
|
package bannerservice
|
||||||
|
|
||||||
|
import (
|
||||||
|
"quenue/app/models/bannermodel"
|
||||||
|
)
|
||||||
|
|
||||||
|
func GetListByPid(pid int, limit int, page int) (banners []*bannermodel.Banner, err error) {
|
||||||
|
limitStart := GetLimitStart(limit, page)
|
||||||
|
banners, err = bannermodel.GetInstance().GetListByPid(pid, limitStart...)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
func GetLimitStart(limit int, page int) (arr []int) {
|
||||||
|
arr = make([]int, 2)
|
||||||
|
if limit <= 0 {
|
||||||
|
limit = 20
|
||||||
|
}
|
||||||
|
arr[0] = limit
|
||||||
|
if page > 0 {
|
||||||
|
arr[1] = (page - 1) * limit
|
||||||
|
} else {
|
||||||
|
arr[1] = 0
|
||||||
|
}
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,190 @@
|
||||||
|
package httpclient
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"net/http"
|
||||||
|
"strconv"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"quenue/app/http/trace"
|
||||||
|
"quenue/config"
|
||||||
|
|
||||||
|
"github.com/SkyAPM/go2sky"
|
||||||
|
"github.com/SkyAPM/go2sky/propagation"
|
||||||
|
v3 "github.com/SkyAPM/go2sky/reporter/grpc/language-agent"
|
||||||
|
"github.com/go-resty/resty/v2"
|
||||||
|
"github.com/qit-team/snow-core/log/logger"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
RetryCounts = 2
|
||||||
|
RetryInterval = 3 * time.Second
|
||||||
|
)
|
||||||
|
|
||||||
|
const componentIDGOHttpClient = 5005
|
||||||
|
|
||||||
|
type ClientConfig struct {
|
||||||
|
ctx context.Context
|
||||||
|
client *resty.Client
|
||||||
|
tracer *go2sky.Tracer
|
||||||
|
extraTags map[string]string
|
||||||
|
}
|
||||||
|
|
||||||
|
type ClientOption func(*ClientConfig)
|
||||||
|
|
||||||
|
func WithClientTag(key string, value string) ClientOption {
|
||||||
|
return func(c *ClientConfig) {
|
||||||
|
if c.extraTags == nil {
|
||||||
|
c.extraTags = make(map[string]string)
|
||||||
|
}
|
||||||
|
c.extraTags[key] = value
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func WithClient(client *resty.Client) ClientOption {
|
||||||
|
return func(c *ClientConfig) {
|
||||||
|
c.client = client
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func WithContext(ctx context.Context) ClientOption {
|
||||||
|
return func(c *ClientConfig) {
|
||||||
|
c.ctx = ctx
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
type transport struct {
|
||||||
|
*ClientConfig
|
||||||
|
delegated http.RoundTripper
|
||||||
|
}
|
||||||
|
|
||||||
|
func (t *transport) RoundTrip(req *http.Request) (resp *http.Response, err error) {
|
||||||
|
span, err := t.tracer.CreateExitSpan(t.ctx, fmt.Sprintf("/%s%s", req.Method, req.URL.Path), req.Host, func(header string) error {
|
||||||
|
// 将本层的调用链信息写入http头部, 传入到下一层调用, 当前使用v3版本的协议
|
||||||
|
// https://github.com/apache/skywalking/blob/master/docs/en/protocols/Skywalking-Cross-Process-Propagation-Headers-Protocol-v3.md
|
||||||
|
req.Header.Set(propagation.Header, header)
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
return t.delegated.RoundTrip(req)
|
||||||
|
}
|
||||||
|
defer span.End()
|
||||||
|
span.SetComponent(componentIDGOHttpClient)
|
||||||
|
for k, v := range t.extraTags {
|
||||||
|
span.Tag(go2sky.Tag(k), v)
|
||||||
|
}
|
||||||
|
span.Tag(go2sky.TagHTTPMethod, req.Method)
|
||||||
|
span.Tag(go2sky.TagURL, req.URL.String())
|
||||||
|
span.SetSpanLayer(v3.SpanLayer_Http)
|
||||||
|
resp, err = t.delegated.RoundTrip(req)
|
||||||
|
if err != nil {
|
||||||
|
span.Error(time.Now(), err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
span.Tag(go2sky.TagStatusCode, strconv.Itoa(resp.StatusCode))
|
||||||
|
if resp.StatusCode >= http.StatusBadRequest {
|
||||||
|
span.Error(time.Now(), "Errors on handling client")
|
||||||
|
}
|
||||||
|
return resp, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewClient(ctx context.Context, options ...ClientOption) (client *resty.Client) {
|
||||||
|
client = resty.New()
|
||||||
|
if config.IsDebug() {
|
||||||
|
client.SetDebug(true).EnableTrace()
|
||||||
|
}
|
||||||
|
|
||||||
|
var (
|
||||||
|
tracer *go2sky.Tracer
|
||||||
|
err error
|
||||||
|
)
|
||||||
|
if len(config.GetConf().SkyWalkingOapServer) > 0 && config.IsEnvEqual(config.ProdEnv) {
|
||||||
|
tracer, err = trace.Tracer()
|
||||||
|
if err != nil {
|
||||||
|
logger.Error(ctx, "NewClient:Tracer", err.Error())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if tracer != nil {
|
||||||
|
co := &ClientConfig{ctx: ctx, tracer: tracer}
|
||||||
|
for _, option := range options {
|
||||||
|
option(co)
|
||||||
|
}
|
||||||
|
if co.client == nil {
|
||||||
|
co.client = client
|
||||||
|
}
|
||||||
|
tp := &transport{
|
||||||
|
ClientConfig: co,
|
||||||
|
delegated: http.DefaultTransport,
|
||||||
|
}
|
||||||
|
if co.client.GetClient().Transport != nil {
|
||||||
|
tp.delegated = co.client.GetClient().Transport
|
||||||
|
}
|
||||||
|
co.client.SetTransport(tp)
|
||||||
|
}
|
||||||
|
|
||||||
|
client.OnBeforeRequest(func(ct *resty.Client, req *resty.Request) error {
|
||||||
|
//req.SetContext(c)
|
||||||
|
logger.Info(ctx, "OnBeforeRequest", logger.NewWithField("url", req.URL))
|
||||||
|
return nil // if its success otherwise return error
|
||||||
|
})
|
||||||
|
// Registering Response Middleware
|
||||||
|
client.OnAfterResponse(func(ct *resty.Client, resp *resty.Response) error {
|
||||||
|
logger.Info(ctx, "OnAfterResponse", logger.NewWithField("url", resp.Request.URL), logger.NewWithField("request", resp.Request.RawRequest), logger.NewWithField("response", resp.String()))
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
return client
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewClientWithRetry(ctx context.Context, retryCounts int, retryInterval time.Duration, options ...ClientOption) (client *resty.Client) {
|
||||||
|
client = resty.New()
|
||||||
|
if config.IsDebug() {
|
||||||
|
client.SetDebug(true).EnableTrace()
|
||||||
|
}
|
||||||
|
if retryCounts == 0 {
|
||||||
|
retryCounts = RetryCounts
|
||||||
|
}
|
||||||
|
if retryInterval.Seconds() == 0.0 {
|
||||||
|
retryInterval = RetryInterval
|
||||||
|
}
|
||||||
|
client.SetRetryCount(retryCounts).SetRetryMaxWaitTime(retryInterval)
|
||||||
|
|
||||||
|
var (
|
||||||
|
tracer *go2sky.Tracer
|
||||||
|
err error
|
||||||
|
)
|
||||||
|
if len(config.GetConf().SkyWalkingOapServer) > 0 && config.IsEnvEqual(config.ProdEnv) {
|
||||||
|
tracer, err = trace.Tracer()
|
||||||
|
if err != nil {
|
||||||
|
logger.Error(ctx, "NewClient:Tracer", err.Error())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if tracer != nil {
|
||||||
|
co := &ClientConfig{ctx: ctx, tracer: tracer}
|
||||||
|
for _, option := range options {
|
||||||
|
option(co)
|
||||||
|
}
|
||||||
|
if co.client == nil {
|
||||||
|
co.client = client
|
||||||
|
}
|
||||||
|
tp := &transport{
|
||||||
|
ClientConfig: co,
|
||||||
|
delegated: http.DefaultTransport,
|
||||||
|
}
|
||||||
|
if co.client.GetClient().Transport != nil {
|
||||||
|
tp.delegated = co.client.GetClient().Transport
|
||||||
|
}
|
||||||
|
co.client.SetTransport(tp)
|
||||||
|
}
|
||||||
|
|
||||||
|
client.OnBeforeRequest(func(ct *resty.Client, req *resty.Request) error {
|
||||||
|
logger.Info(ctx, "OnBeforeRequest", logger.NewWithField("url", req.URL))
|
||||||
|
return nil // if its success otherwise return error
|
||||||
|
})
|
||||||
|
// Registering Response Middleware
|
||||||
|
client.OnAfterResponse(func(ct *resty.Client, resp *resty.Response) error {
|
||||||
|
logger.Info(ctx, "OnAfterResponse", logger.NewWithField("url", resp.Request.URL), logger.NewWithField("request", resp.Request.RawRequest), logger.NewWithField("response", resp.String()))
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
return client
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,177 @@
|
||||||
|
package metric
|
||||||
|
|
||||||
|
// prometheus metric：unique identifier: name and optional key-value pairs called labels
|
||||||
|
// 1. name regexp: [a-zA-Z_:][a-zA-Z0-9_:]*
|
||||||
|
// 2. label name regexp: [a-zA-Z_][a-zA-Z0-9_]*
|
||||||
|
// 3. Label names beginning with __ are reserved for internal use.
|
||||||
|
// 4. Label values may contain any Unicode characters.
|
||||||
|
// 5. notation: <metric name>{<label name>=<label value>, ...}
|
||||||
|
// for example: api_http_requests_total{method="POST", handler="/messages"}
|
||||||
|
// A label with an empty label value is considered equivalent to a label that does not exist.
|
||||||
|
|
||||||
|
// each sample consists of :
|
||||||
|
// - a float64 value
|
||||||
|
// - a millisecond-precision timestamp
|
||||||
|
|
||||||
|
// metric type:
|
||||||
|
// - Counter
|
||||||
|
// A cumulative metric that represents a single monotonically increasing counter whose value can only increase or be reset to zero on restart.
|
||||||
|
// - Gauge
|
||||||
|
// A gauge is a metric that represents a single numerical value that can arbitrarily go up and down.
|
||||||
|
// - Histogram
|
||||||
|
// A histogram samples observations (usually things like request durations or response sizes) and counts them in configurable buckets. It also provides a sum of all observed values.
|
||||||
|
// - Summary
|
||||||
|
// Similar to a histogram, a summary samples observations (usually things like request durations and response sizes). While it also provides a total count of observations and a sum of all observed values, it calculates configurable quantiles over a sliding time window.
|
||||||
|
// metric:
|
||||||
|
// Counter:
|
||||||
|
// - req_total_count
|
||||||
|
// - req_failed_count
|
||||||
|
// Gauge:
|
||||||
|
// - heap_inuse_size
|
||||||
|
// - heap_total_size
|
||||||
|
// - heap_object_num
|
||||||
|
// - goroutine_num
|
||||||
|
// Histogram:
|
||||||
|
// - req_cost_time
|
||||||
|
// Summary:
|
||||||
|
// - req_cost_time
|
||||||
|
|
||||||
|
import (
|
||||||
|
"net/http"
|
||||||
|
"sync"
|
||||||
|
|
||||||
|
"github.com/prometheus/client_golang/prometheus"
|
||||||
|
"github.com/prometheus/client_golang/prometheus/promhttp"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
//ENV = "env"
|
||||||
|
APP = "snow"
|
||||||
|
VER = "ver"
|
||||||
|
)
|
||||||
|
|
||||||
|
var (
|
||||||
|
collectors = []prometheus.Collector{}
|
||||||
|
)
|
||||||
|
|
||||||
|
func RegisterCollector(c ...prometheus.Collector) {
|
||||||
|
collectors = append(collectors, c...)
|
||||||
|
}
|
||||||
|
|
||||||
|
type Options struct {
|
||||||
|
labels map[string]string
|
||||||
|
processEnable bool
|
||||||
|
runtimeEnable bool
|
||||||
|
}
|
||||||
|
|
||||||
|
type Option func(opt *Options)
|
||||||
|
|
||||||
|
// 添加App和Ver label
|
||||||
|
func AppVer(app, ver string) Option {
|
||||||
|
return func(opt *Options) {
|
||||||
|
if app != "" {
|
||||||
|
opt.labels[APP] = app
|
||||||
|
}
|
||||||
|
if ver != "" {
|
||||||
|
opt.labels[VER] = ver
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 添加额外label
|
||||||
|
func WithLabel(key, val string) Option {
|
||||||
|
return func(opt *Options) {
|
||||||
|
if key != "" && val != "" {
|
||||||
|
opt.labels[key] = val
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 收集进程信息
|
||||||
|
func EnableProcess() Option {
|
||||||
|
return func(opt *Options) {
|
||||||
|
opt.processEnable = true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func EnableRuntime() Option {
|
||||||
|
return func(opt *Options) {
|
||||||
|
opt.runtimeEnable = true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
type Reporter struct {
|
||||||
|
opts Options
|
||||||
|
collectors []prometheus.Collector
|
||||||
|
// registerer
|
||||||
|
registerer prometheus.Registerer
|
||||||
|
gatherer prometheus.Gatherer
|
||||||
|
}
|
||||||
|
|
||||||
|
var (
|
||||||
|
once sync.Once
|
||||||
|
reporter Reporter
|
||||||
|
)
|
||||||
|
|
||||||
|
func Init(opts ...Option) {
|
||||||
|
_opts := Options{
|
||||||
|
labels: map[string]string{},
|
||||||
|
}
|
||||||
|
for _, opt := range opts {
|
||||||
|
opt(&_opts)
|
||||||
|
}
|
||||||
|
|
||||||
|
once.Do(func() {
|
||||||
|
cs := collectors
|
||||||
|
if _opts.processEnable {
|
||||||
|
cs = append(cs, prometheus.NewProcessCollector(prometheus.ProcessCollectorOpts{}))
|
||||||
|
}
|
||||||
|
|
||||||
|
if _opts.runtimeEnable {
|
||||||
|
cs = append(cs, prometheus.NewGoCollector())
|
||||||
|
}
|
||||||
|
|
||||||
|
reporter = Reporter{
|
||||||
|
opts: _opts,
|
||||||
|
collectors: cs,
|
||||||
|
}
|
||||||
|
|
||||||
|
registry := prometheus.NewRegistry()
|
||||||
|
|
||||||
|
reporter.registerer = prometheus.WrapRegistererWith(reporter.opts.labels, registry)
|
||||||
|
reporter.gatherer = registry
|
||||||
|
|
||||||
|
reporter.registerer.MustRegister(reporter.collectors...)
|
||||||
|
|
||||||
|
})
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *Reporter) newCounterVec(metric string, labels []string) *prometheus.CounterVec {
|
||||||
|
counterVec := prometheus.NewCounterVec(prometheus.CounterOpts{
|
||||||
|
Name: metric,
|
||||||
|
}, labels)
|
||||||
|
|
||||||
|
return counterVec
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *Reporter) newGaugeVec(metric string, labels []string) *prometheus.GaugeVec {
|
||||||
|
gaugeVec := prometheus.NewGaugeVec(prometheus.GaugeOpts{
|
||||||
|
Name: metric,
|
||||||
|
}, labels)
|
||||||
|
return gaugeVec
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *Reporter) newHistogramVec(metric string, labels []string, buckets []float64) *prometheus.HistogramVec {
|
||||||
|
histogramVec := prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
||||||
|
Name: metric,
|
||||||
|
Buckets: buckets,
|
||||||
|
}, labels)
|
||||||
|
return histogramVec
|
||||||
|
}
|
||||||
|
|
||||||
|
func Handler() http.Handler {
|
||||||
|
return promhttp.InstrumentMetricHandler(
|
||||||
|
reporter.registerer, promhttp.HandlerFor(reporter.gatherer, promhttp.HandlerOpts{}),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,6 @@
|
||||||
|
package common
|
||||||
|
|
||||||
|
type Imq interface {
|
||||||
|
Produce(name string, log interface{}, delayTime int, args ...interface{}) error
|
||||||
|
Consume(name string, hand interface{}, i int)
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,95 @@
|
||||||
|
package mqs
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"github.com/IBM/sarama"
|
||||||
|
"github.com/qit-team/snow-core/redis"
|
||||||
|
"quenue/app/utils"
|
||||||
|
"quenue/config"
|
||||||
|
"strconv"
|
||||||
|
"sync"
|
||||||
|
)
|
||||||
|
|
||||||
|
type KafkaMq struct {
|
||||||
|
config *sarama.Config
|
||||||
|
}
|
||||||
|
|
||||||
|
func (n *KafkaMq) Init(name string) *KafkaMq {
|
||||||
|
kafConfig := sarama.NewConfig()
|
||||||
|
kafConfig.Producer.RequiredAcks = sarama.WaitForAll // 发送完数据需要leader和follow都确认
|
||||||
|
kafConfig.Producer.Partitioner = sarama.NewRandomPartitioner // 新选出一个partition
|
||||||
|
kafConfig.Producer.Return.Successes = true
|
||||||
|
n.config = kafConfig
|
||||||
|
return n
|
||||||
|
}
|
||||||
|
|
||||||
|
// 同步
|
||||||
|
func (n KafkaMq) Produce(name string, log interface{}, delayTime int, args ...interface{}) error {
|
||||||
|
// 成功交付的消息将在success channel返回
|
||||||
|
// 构造一个消息
|
||||||
|
msg := &sarama.ProducerMessage{}
|
||||||
|
msg.Topic = name
|
||||||
|
|
||||||
|
var data, _ = json.Marshal(log)
|
||||||
|
msg.Value = sarama.StringEncoder(string(data))
|
||||||
|
msg.Partition = 2
|
||||||
|
// 连接kafka
|
||||||
|
client, err := sarama.NewSyncProducer([]string{config.GetConf().KafkaUrl}, n.config)
|
||||||
|
if err != nil {
|
||||||
|
fmt.Println("producer closed, err:", err)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
defer client.Close()
|
||||||
|
// 发送消息
|
||||||
|
pid, offset, err := client.SendMessage(msg)
|
||||||
|
if err != nil {
|
||||||
|
utils.Log(nil, "send msg failed, err:", err, pid, offset)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
fmt.Println("send msg success, pid:", pid, offset)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
func (n KafkaMq) Consume(name string, hand interface{}, i int) {
|
||||||
|
utils.Log(nil, "hand", i)
|
||||||
|
consumer, err := sarama.NewConsumer([]string{config.GetConf().KafkaUrl}, nil)
|
||||||
|
if err != nil {
|
||||||
|
utils.Log(nil, "kafka comsume", err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
partitionList, err := consumer.Partitions(name) // 根据topic取到所有的分区
|
||||||
|
if err != nil {
|
||||||
|
utils.Log(nil, "kafka comsume", err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
//utils.Log(nil,"kafka comsume",name,partitionList)
|
||||||
|
for partition := range partitionList { // 遍历所有的分区
|
||||||
|
// 针对每个分区创建一个对应的分区消费者
|
||||||
|
var offsetReDis, _ = redis.GetRedis().Incr(context.Background(), "kafka_consume:"+strconv.Itoa(int(partition))).Result() //保证多消费者不重复消费
|
||||||
|
var offset int64 = sarama.OffsetNewest
|
||||||
|
if offsetReDis > 0 {
|
||||||
|
offset = int64(offsetReDis)
|
||||||
|
}
|
||||||
|
pc, err := consumer.ConsumePartition(name, int32(partition), offset)
|
||||||
|
if err != nil {
|
||||||
|
fmt.Printf("failed to start consumer for partition %d,err:%v\n", partition, err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
defer pc.AsyncClose()
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
wg.Add(1)
|
||||||
|
// 异步从每个分区消费信息
|
||||||
|
go func(sarama.PartitionConsumer) {
|
||||||
|
for msg := range pc.Messages() {
|
||||||
|
defer wg.Done()
|
||||||
|
var handler = hand.(func(tag uint64, ch interface{}, msg []byte))
|
||||||
|
redis.GetRedis().Set(context.Background(), "kafka_consume:"+strconv.Itoa(int(partition)), msg.Offset, 0)
|
||||||
|
handler(0, nil, msg.Value)
|
||||||
|
//utils.Log(nil,"hand msg",string(msg.Value),msg.Offset)
|
||||||
|
}
|
||||||
|
}(pc)
|
||||||
|
wg.Wait()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,114 @@
|
||||||
|
package mqs
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"github.com/confluentinc/confluent-kafka-go/kafka"
|
||||||
|
"quenue/app/http/entities"
|
||||||
|
"quenue/app/http/tcppool"
|
||||||
|
"quenue/app/utils"
|
||||||
|
config "quenue/config"
|
||||||
|
"sync/atomic"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
type KafkaV2Mq struct {
|
||||||
|
}
|
||||||
|
|
||||||
|
func (kk KafkaV2Mq) Produce(name string, log interface{}, delayTime int, args ...interface{}) error {
|
||||||
|
kfconfig := &kafka.ConfigMap{
|
||||||
|
"bootstrap.servers": config.GetConf().KafkaUrl, // Kafka服务器地址
|
||||||
|
}
|
||||||
|
|
||||||
|
producer, err := kafka.NewProducer(kfconfig)
|
||||||
|
if err != nil {
|
||||||
|
utils.Log(nil, "kafka productor init error", err)
|
||||||
|
}
|
||||||
|
defer producer.Close()
|
||||||
|
var logData, _ = json.Marshal(log)
|
||||||
|
message := &kafka.Message{
|
||||||
|
TopicPartition: kafka.TopicPartition{Topic: &name, Partition: kafka.PartitionAny},
|
||||||
|
Value: []byte(logData),
|
||||||
|
}
|
||||||
|
err = producer.Produce(message, nil)
|
||||||
|
if err != nil {
|
||||||
|
utils.Log(nil, "Failed to produce message:", err)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
utils.Log(nil, "kafka produce", err)
|
||||||
|
producer.Flush(15 * 1000)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
func (kk KafkaV2Mq) Consume(name string, hand interface{}, ci int) {
|
||||||
|
utils.Log(nil, "hand", ci)
|
||||||
|
kfconfig := &kafka.ConfigMap{
|
||||||
|
"bootstrap.servers": config.GetConf().KafkaUrl, // Kafka服务器地址
|
||||||
|
"group.id": config.GetConf().KafkaGroup, // 消费者组ID
|
||||||
|
"auto.offset.reset": "earliest", // 自动从最早的消息开始消费
|
||||||
|
"heartbeat.interval.ms": 3000,
|
||||||
|
"session.timeout.ms": 8000,
|
||||||
|
"enable.auto.commit": false,
|
||||||
|
}
|
||||||
|
|
||||||
|
consumer, err := kafka.NewConsumer(kfconfig)
|
||||||
|
|
||||||
|
if err != nil {
|
||||||
|
utils.Log(nil, "comsume", err)
|
||||||
|
}
|
||||||
|
defer consumer.Close()
|
||||||
|
err = consumer.Subscribe(name, nil)
|
||||||
|
if err != nil {
|
||||||
|
utils.Log(nil, "Failed to subscribe:", err)
|
||||||
|
}
|
||||||
|
for i := 0; i < config.GetConf().Num; i++ {
|
||||||
|
go func() {
|
||||||
|
for {
|
||||||
|
if atomic.LoadInt32(tcppool.TcpPoolFactory.Full) == 1 {
|
||||||
|
utils.Log(nil, "tcp is die")
|
||||||
|
time.Sleep(1 * time.Second)
|
||||||
|
}
|
||||||
|
msg, err := consumer.ReadMessage(-1)
|
||||||
|
fmt.Println("read msg", ci)
|
||||||
|
if err == nil {
|
||||||
|
utils.Log(nil, "offset", msg.TopicPartition.Offset)
|
||||||
|
var handler = hand.(func(tag uint64, ch interface{}, msg []byte) error)
|
||||||
|
var mqsg = entities.MqMessage{}
|
||||||
|
mqsg.Key = string(msg.Key)
|
||||||
|
mqsg.Property = make(map[string]interface{})
|
||||||
|
if msg.Headers != nil {
|
||||||
|
for _, v := range msg.Headers {
|
||||||
|
if v.Key == "property" {
|
||||||
|
json.Unmarshal(v.Value, &mqsg.Property)
|
||||||
|
}
|
||||||
|
mqsg.Property[string(v.Key)] = string(v.Value)
|
||||||
|
}
|
||||||
|
err = json.Unmarshal(msg.Value, &mqsg.Body)
|
||||||
|
} else {
|
||||||
|
mqsg.Property = map[string]interface{}{"reseller_id": 20001, "order_product_id": 104, "platform_product_id": 592, "serial_number": "100000001", "platform_tag": "InternalTest", "tag": "PInternalTest"}
|
||||||
|
mqsg.Body = map[string]interface{}{"reseller_id": 20001, "order_price": 12.01, "account_type": 0}
|
||||||
|
}
|
||||||
|
var data, _ = json.Marshal(mqsg)
|
||||||
|
err = handler(0, nil, data)
|
||||||
|
if err == nil {
|
||||||
|
//手动提交编译量
|
||||||
|
kk.commitOffset(consumer, msg.TopicPartition)
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
utils.Log(nil, "Error while consuming: %v\n", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
select {}
|
||||||
|
}
|
||||||
|
func (kk *KafkaV2Mq) commitOffset(consumer *kafka.Consumer, tp kafka.TopicPartition) {
|
||||||
|
// 创建一个偏移量提交请求
|
||||||
|
offsets := []kafka.TopicPartition{tp}
|
||||||
|
commit, err := consumer.CommitOffsets(offsets)
|
||||||
|
if err != nil {
|
||||||
|
utils.Log(nil, "Failed to commit offset: %v", err)
|
||||||
|
} else {
|
||||||
|
utils.Log(nil, "Committed offset: %v", commit)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,28 @@
|
||||||
|
package mqs
|
||||||
|
|
||||||
|
import (
|
||||||
|
"quenue/app/constants/common"
|
||||||
|
common2 "quenue/app/utils/mq/common"
|
||||||
|
"sync"
|
||||||
|
)
|
||||||
|
|
||||||
|
var (
|
||||||
|
MqManager = CMqManager{}
|
||||||
|
once sync.Once
|
||||||
|
)
|
||||||
|
|
||||||
|
type CMqManager struct {
|
||||||
|
mqs map[string]common2.Imq
|
||||||
|
}
|
||||||
|
|
||||||
|
func (this *CMqManager) InitMq() {
|
||||||
|
this.mqs = make(map[string]common2.Imq)
|
||||||
|
this.mqs[common.MQ_KFK] = KafkaMq{}
|
||||||
|
this.mqs[common.MQ_KFK_V2] = KafkaV2Mq{}
|
||||||
|
}
|
||||||
|
func (this *CMqManager) GetMqByName(name string) common2.Imq {
|
||||||
|
once.Do(func() {
|
||||||
|
this.InitMq()
|
||||||
|
})
|
||||||
|
return this.mqs[name]
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,15 @@
|
||||||
|
package utils
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"github.com/gin-gonic/gin"
|
||||||
|
"runtime"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
func Log(c *gin.Context, name string, msg ...interface{}) {
|
||||||
|
_, file, line, _ := runtime.Caller(1)
|
||||||
|
timeLayout := "2006-01-01 03:04:05" //转化所需模板
|
||||||
|
var datetime = time.Unix(time.Now().Unix(), 0).Format(timeLayout)
|
||||||
|
fmt.Println(name, msg, file, line, datetime)
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,63 @@
|
||||||
|
package bootstrap
|
||||||
|
|
||||||
|
import (
|
||||||
|
"quenue/app/jobs"
|
||||||
|
"quenue/app/jobs/basejob"
|
||||||
|
"quenue/config"
|
||||||
|
|
||||||
|
"github.com/qit-team/snow-core/db"
|
||||||
|
"github.com/qit-team/snow-core/kernel/close"
|
||||||
|
"github.com/qit-team/snow-core/kernel/container"
|
||||||
|
"github.com/qit-team/snow-core/log/accesslogger"
|
||||||
|
"github.com/qit-team/snow-core/log/logger"
|
||||||
|
"github.com/qit-team/snow-core/redis"
|
||||||
|
)
|
||||||
|
|
||||||
|
//全局变量
|
||||||
|
var App *container.Container
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 服务引导程序
|
||||||
|
*/
|
||||||
|
func Bootstrap(conf *config.Config) (err error) {
|
||||||
|
//容器
|
||||||
|
App = container.App
|
||||||
|
|
||||||
|
//注册db服务
|
||||||
|
//第一个参数为注入别名，第二个参数为配置，第三个参数可选为是否懒加载
|
||||||
|
err = db.Pr.Register(db.SingletonMain, conf.Db)
|
||||||
|
if err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
//注册redis服务
|
||||||
|
err = redis.Pr.Register(redis.SingletonMain, conf.Redis)
|
||||||
|
if err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
//注册mns服务
|
||||||
|
//err = alimns.Pr.Register(alimns.SingletonMain, conf.Mns, true)
|
||||||
|
//if err != nil {
|
||||||
|
// return
|
||||||
|
//}
|
||||||
|
|
||||||
|
//注册日志类服务
|
||||||
|
err = logger.Pr.Register(logger.SingletonMain, conf.Log, true)
|
||||||
|
if err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
//注册access log服务
|
||||||
|
err = accesslogger.Pr.Register(accesslogger.SingletonMain, conf.Log)
|
||||||
|
if err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
//注册应用停止时调用的关闭服务
|
||||||
|
close.MultiRegister(db.Pr, redis.Pr)
|
||||||
|
|
||||||
|
//注册job register，为了非job模式的消息入队调用
|
||||||
|
basejob.SetJobRegister(jobs.RegisterWorker)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,2 @@
|
||||||
|
*
|
||||||
|
!.gitignore
|
||||||
|
|
@ -0,0 +1,9 @@
|
||||||
|
#/bin/bash
|
||||||
|
os=$1 #系统linux
|
||||||
|
arch=$2 #架构amd64
|
||||||
|
|
||||||
|
#回到根目录
|
||||||
|
rootPath=$(cd `dirname $0`/../../; pwd)
|
||||||
|
|
||||||
|
#编译
|
||||||
|
GOOS=$os GOARCH=$arch go build -o build/bin/snow main.go
|
||||||
|
|
@ -0,0 +1,78 @@
|
||||||
|
package config
|
||||||
|
|
||||||
|
import (
|
||||||
|
"os"
|
||||||
|
|
||||||
|
"github.com/BurntSushi/toml"
|
||||||
|
"github.com/qit-team/snow-core/config"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
ProdEnv = "production" //线上环境
|
||||||
|
BetaEnv = "beta" //beta环境
|
||||||
|
DevEnv = "develop" //开发环境
|
||||||
|
LocalEnv = "local" //本地环境
|
||||||
|
)
|
||||||
|
|
||||||
|
var srvConf *Config
|
||||||
|
|
||||||
|
// ------------------------配置文件解析
|
||||||
|
type Config struct {
|
||||||
|
ServiceName string `toml:"ServiceName"`
|
||||||
|
Env string `toml:"Env"`
|
||||||
|
Debug bool `toml:"Debug"`
|
||||||
|
PrometheusCollectEnable bool `toml:"PrometheusCollectEnable"`
|
||||||
|
SkyWalkingOapServer string `toml:"SkyWalkingOapServer"`
|
||||||
|
Log config.LogConfig `toml:"Log"`
|
||||||
|
Redis config.RedisConfig `toml:"Redis"`
|
||||||
|
Mns config.MnsConfig `toml:"AliMns"`
|
||||||
|
Db config.DbConfig `toml:"Db"`
|
||||||
|
Api config.ApiConfig `toml:"Api"`
|
||||||
|
StartQunue int32 `toml:"StartQunue"`
|
||||||
|
KafkaUrl string `toml:"KafkaUrl"`
|
||||||
|
KafkaGroup string `toml:"KafkaGroup"`
|
||||||
|
OrderPort string `toml:"OrderPort"`
|
||||||
|
Num int `toml:"Num"`
|
||||||
|
}
|
||||||
|
|
||||||
|
func newConfig() *Config {
|
||||||
|
return new(Config)
|
||||||
|
}
|
||||||
|
|
||||||
|
// ------------------------ 加载配置 ------------------------//
|
||||||
|
func Load(path string) (*Config, error) {
|
||||||
|
_, err := os.Stat(path)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
conf := newConfig()
|
||||||
|
if _, err := toml.DecodeFile(path, conf); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
srvConf = conf
|
||||||
|
return conf, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// 当前配置
|
||||||
|
func GetConf() *Config {
|
||||||
|
return srvConf
|
||||||
|
}
|
||||||
|
|
||||||
|
// 是否调试模式
|
||||||
|
func IsDebug() bool {
|
||||||
|
return srvConf.Debug
|
||||||
|
}
|
||||||
|
|
||||||
|
// 当前环境，默认本地开发
|
||||||
|
func GetEnv() string {
|
||||||
|
if srvConf.Env == "" {
|
||||||
|
return LocalEnv
|
||||||
|
}
|
||||||
|
return srvConf.Env
|
||||||
|
}
|
||||||
|
|
||||||
|
// 是否当前环境
|
||||||
|
func IsEnvEqual(env string) bool {
|
||||||
|
return GetEnv() == env
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,45 @@
|
||||||
|
package config
|
||||||
|
|
||||||
|
import (
|
||||||
|
"flag"
|
||||||
|
"strings"
|
||||||
|
)
|
||||||
|
|
||||||
|
var options *Options
|
||||||
|
|
||||||
|
//------------------------启动命令配置
|
||||||
|
type Options struct {
|
||||||
|
ShowVersion bool
|
||||||
|
Cmd string
|
||||||
|
ConfFile string
|
||||||
|
App string
|
||||||
|
PidDir string
|
||||||
|
Queue string
|
||||||
|
Command string
|
||||||
|
}
|
||||||
|
|
||||||
|
func parseOptions() *Options {
|
||||||
|
opts := new(Options)
|
||||||
|
flag.BoolVar(&opts.ShowVersion, "v", false, "show version")
|
||||||
|
flag.StringVar(&opts.App, "a", "api", "application to run")
|
||||||
|
flag.StringVar(&opts.Cmd, "k", "", "status|stop|restart")
|
||||||
|
flag.StringVar(&opts.ConfFile, "c", ".env", "conf file path")
|
||||||
|
flag.StringVar(&opts.PidDir, "p", "/var/run/", "pid directory")
|
||||||
|
flag.StringVar(&opts.Queue, "queue", "", "topic of queue is enable")
|
||||||
|
flag.StringVar(&opts.Command, "m", "", "command name")
|
||||||
|
flag.Parse()
|
||||||
|
return opts
|
||||||
|
}
|
||||||
|
|
||||||
|
//获取启动命令配置
|
||||||
|
func GetOptions() *Options {
|
||||||
|
if options == nil {
|
||||||
|
options = parseOptions()
|
||||||
|
}
|
||||||
|
return options
|
||||||
|
}
|
||||||
|
|
||||||
|
//pid进程号的保存路径
|
||||||
|
func (opts *Options) GenPidFile() string {
|
||||||
|
return strings.TrimRight(opts.PidDir, "/") + "/" + opts.App + ".pid"
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,48 @@
|
||||||
|
package docs
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"encoding/json"
|
||||||
|
|
||||||
|
"github.com/alecthomas/template"
|
||||||
|
"github.com/swaggo/swag"
|
||||||
|
)
|
||||||
|
|
||||||
|
var doc =``
|
||||||
|
|
||||||
|
type swaggerInfo struct {
|
||||||
|
Version string
|
||||||
|
Host string
|
||||||
|
BasePath string
|
||||||
|
Schemes []string
|
||||||
|
Title string
|
||||||
|
Description string
|
||||||
|
}
|
||||||
|
|
||||||
|
// SwaggerInfo holds exported Swagger Info so clients can modify it
|
||||||
|
var SwaggerInfo = swaggerInfo{ Schemes: []string{}}
|
||||||
|
|
||||||
|
type s struct{}
|
||||||
|
|
||||||
|
func (s *s) ReadDoc() string {
|
||||||
|
t, err := template.New("swagger_info").Funcs(template.FuncMap{
|
||||||
|
"marshal": func(v interface {}) string {
|
||||||
|
a, _ := json.Marshal(v)
|
||||||
|
return string(a)
|
||||||
|
},
|
||||||
|
}).Parse(doc)
|
||||||
|
if err != nil {
|
||||||
|
return doc
|
||||||
|
}
|
||||||
|
|
||||||
|
var tpl bytes.Buffer
|
||||||
|
if err := t.Execute(&tpl, SwaggerInfo); err != nil {
|
||||||
|
return doc
|
||||||
|
}
|
||||||
|
|
||||||
|
return tpl.String()
|
||||||
|
}
|
||||||
|
|
||||||
|
func init() {
|
||||||
|
swag.Register(swag.Name, &s{})
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,107 @@
|
||||||
|
module quenue
|
||||||
|
|
||||||
|
go 1.21
|
||||||
|
|
||||||
|
toolchain go1.22.2
|
||||||
|
|
||||||
|
require (
|
||||||
|
github.com/BurntSushi/toml v0.4.1
|
||||||
|
github.com/IBM/sarama v1.45.0
|
||||||
|
github.com/SkyAPM/go2sky v0.6.0
|
||||||
|
github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751
|
||||||
|
github.com/bwmarrin/snowflake v0.3.0
|
||||||
|
github.com/confluentinc/confluent-kafka-go v1.9.2
|
||||||
|
github.com/gin-gonic/gin v1.7.7
|
||||||
|
github.com/go-netty/go-netty v1.6.7
|
||||||
|
github.com/go-redis/redis v6.15.9+incompatible
|
||||||
|
github.com/go-resty/resty/v2 v2.7.0
|
||||||
|
github.com/go-sql-driver/mysql v1.6.0
|
||||||
|
github.com/nange/easypool v1.3.1
|
||||||
|
github.com/prometheus/client_golang v1.11.0
|
||||||
|
github.com/qit-team/snow-core v0.1.28
|
||||||
|
github.com/qit-team/work v0.3.11
|
||||||
|
github.com/robfig/cron v1.2.0
|
||||||
|
github.com/swaggo/gin-swagger v1.3.3
|
||||||
|
github.com/swaggo/swag v1.7.6
|
||||||
|
go.mongodb.org/mongo-driver v1.17.2
|
||||||
|
gopkg.in/go-playground/validator.v9 v9.31.0
|
||||||
|
)
|
||||||
|
|
||||||
|
require (
|
||||||
|
github.com/KyleBanks/depth v1.2.1 // indirect
|
||||||
|
github.com/PuerkitoBio/purell v1.1.1 // indirect
|
||||||
|
github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 // indirect
|
||||||
|
github.com/apache/rocketmq-client-go/v2 v2.1.0 // indirect
|
||||||
|
github.com/beorn7/perks v1.0.1 // indirect
|
||||||
|
github.com/cespare/xxhash/v2 v2.1.2 // indirect
|
||||||
|
github.com/davecgh/go-spew v1.1.1 // indirect
|
||||||
|
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
|
||||||
|
github.com/eapache/go-resiliency v1.7.0 // indirect
|
||||||
|
github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 // indirect
|
||||||
|
github.com/eapache/queue v1.1.0 // indirect
|
||||||
|
github.com/emirpasic/gods v1.12.0 // indirect
|
||||||
|
github.com/gin-contrib/sse v0.1.0 // indirect
|
||||||
|
github.com/go-openapi/jsonpointer v0.19.5 // indirect
|
||||||
|
github.com/go-openapi/jsonreference v0.19.6 // indirect
|
||||||
|
github.com/go-openapi/spec v0.20.3 // indirect
|
||||||
|
github.com/go-openapi/swag v0.19.15 // indirect
|
||||||
|
github.com/go-playground/locales v0.14.0 // indirect
|
||||||
|
github.com/go-playground/universal-translator v0.18.0 // indirect
|
||||||
|
github.com/go-playground/validator/v10 v10.9.0 // indirect
|
||||||
|
github.com/go-redis/redis/v8 v8.11.4 // indirect
|
||||||
|
github.com/goccy/go-json v0.8.1 // indirect
|
||||||
|
github.com/golang/mock v1.6.0 // indirect
|
||||||
|
github.com/golang/protobuf v1.5.2 // indirect
|
||||||
|
github.com/golang/snappy v0.0.4 // indirect
|
||||||
|
github.com/google/uuid v1.3.0 // indirect
|
||||||
|
github.com/hashicorp/errwrap v1.0.0 // indirect
|
||||||
|
github.com/hashicorp/go-multierror v1.1.1 // indirect
|
||||||
|
github.com/hashicorp/go-uuid v1.0.3 // indirect
|
||||||
|
github.com/hetiansu5/accesslog v1.0.0 // indirect
|
||||||
|
github.com/hetiansu5/cores v1.0.0 // indirect
|
||||||
|
github.com/jcmturner/aescts/v2 v2.0.0 // indirect
|
||||||
|
github.com/jcmturner/dnsutils/v2 v2.0.0 // indirect
|
||||||
|
github.com/jcmturner/gofork v1.7.6 // indirect
|
||||||
|
github.com/jcmturner/gokrb5/v8 v8.4.4 // indirect
|
||||||
|
github.com/jcmturner/rpc/v2 v2.0.3 // indirect
|
||||||
|
github.com/josharian/intern v1.0.0 // indirect
|
||||||
|
github.com/json-iterator/go v1.1.12 // indirect
|
||||||
|
github.com/klauspost/compress v1.17.11 // indirect
|
||||||
|
github.com/leodido/go-urn v1.2.1 // indirect
|
||||||
|
github.com/lestrrat-go/file-rotatelogs v2.4.0+incompatible // indirect
|
||||||
|
github.com/lestrrat-go/strftime v1.0.5 // indirect
|
||||||
|
github.com/mailru/easyjson v0.7.7 // indirect
|
||||||
|
github.com/mattn/go-isatty v0.0.14 // indirect
|
||||||
|
github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
|
||||||
|
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
|
||||||
|
github.com/modern-go/reflect2 v1.0.2 // indirect
|
||||||
|
github.com/pierrec/lz4/v4 v4.1.22 // indirect
|
||||||
|
github.com/pkg/errors v0.9.1 // indirect
|
||||||
|
github.com/prometheus/client_model v0.2.0 // indirect
|
||||||
|
github.com/prometheus/common v0.26.0 // indirect
|
||||||
|
github.com/prometheus/procfs v0.6.0 // indirect
|
||||||
|
github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
|
||||||
|
github.com/rifflock/lfshook v0.0.0-20180920164130-b9218ef580f5 // indirect
|
||||||
|
github.com/sirupsen/logrus v1.8.1 // indirect
|
||||||
|
github.com/syndtr/goleveldb v1.0.0 // indirect
|
||||||
|
github.com/tidwall/gjson v1.12.1 // indirect
|
||||||
|
github.com/tidwall/match v1.1.1 // indirect
|
||||||
|
github.com/tidwall/pretty v1.2.0 // indirect
|
||||||
|
github.com/ugorji/go/codec v1.2.6 // indirect
|
||||||
|
github.com/valyala/bytebufferpool v1.0.0 // indirect
|
||||||
|
github.com/valyala/fasttemplate v1.2.1 // indirect
|
||||||
|
go.uber.org/atomic v1.9.0 // indirect
|
||||||
|
golang.org/x/crypto v0.32.0 // indirect
|
||||||
|
golang.org/x/net v0.34.0 // indirect
|
||||||
|
golang.org/x/sys v0.29.0 // indirect
|
||||||
|
golang.org/x/text v0.21.0 // indirect
|
||||||
|
golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d // indirect
|
||||||
|
google.golang.org/genproto v0.0.0-20220503193339-ba3ae3f07e29 // indirect
|
||||||
|
google.golang.org/grpc v1.46.0 // indirect
|
||||||
|
google.golang.org/protobuf v1.28.0 // indirect
|
||||||
|
gopkg.in/yaml.v2 v2.4.0 // indirect
|
||||||
|
stathat.com/c/consistent v1.0.0 // indirect
|
||||||
|
xorm.io/builder v0.3.9 // indirect
|
||||||
|
xorm.io/core v0.7.3 // indirect
|
||||||
|
xorm.io/xorm v1.2.5 // indirect
|
||||||
|
)
|
||||||
|
|
@ -0,0 +1,2 @@
|
||||||
|
*
|
||||||
|
!.gitignore
|
||||||
|
|
@ -0,0 +1,146 @@
|
||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"errors"
|
||||||
|
"flag"
|
||||||
|
"fmt"
|
||||||
|
_ "github.com/go-sql-driver/mysql"
|
||||||
|
_ "github.com/qit-team/snow-core/cache/rediscache"
|
||||||
|
"github.com/qit-team/snow-core/kernel/server"
|
||||||
|
_ "github.com/qit-team/snow-core/queue/redisqueue"
|
||||||
|
"os"
|
||||||
|
"quenue/app/console"
|
||||||
|
"quenue/app/handlers/mq"
|
||||||
|
"quenue/app/http/routes"
|
||||||
|
"quenue/app/jobs"
|
||||||
|
"quenue/bootstrap"
|
||||||
|
"quenue/config"
|
||||||
|
_ "quenue/docs"
|
||||||
|
)
|
||||||
|
|
||||||
|
// @title Swagger Example API
|
||||||
|
// @version 1.0
|
||||||
|
// @description This is a sample server celler server.
|
||||||
|
// @termsOfService http://swagger.io/terms/
|
||||||
|
|
||||||
|
// @contact.name API Support
|
||||||
|
// @contact.url http://www.swagger.io/support
|
||||||
|
// @contact.email support@swagger.io
|
||||||
|
|
||||||
|
// @license.name Apache 2.0
|
||||||
|
// @license.url http://www.apache.org/licenses/LICENSE-2.0.html
|
||||||
|
|
||||||
|
// @host localhost:8080
|
||||||
|
// @BasePath /
|
||||||
|
|
||||||
|
// @securityDefinitions.basic BasicAuth
|
||||||
|
|
||||||
|
// @securityDefinitions.apikey ApiKeyAuth
|
||||||
|
// @in header
|
||||||
|
// @name Authorization
|
||||||
|
|
||||||
|
// @securitydefinitions.oauth2.application OAuth2Application
|
||||||
|
// @tokenUrl https://example.com/oauth/token
|
||||||
|
// @scope.write Grants write access
|
||||||
|
// @scope.admin Grants read and write access to administrative information
|
||||||
|
|
||||||
|
// @securitydefinitions.oauth2.implicit OAuth2Implicit
|
||||||
|
// @authorizationUrl https://example.com/oauth/authorize
|
||||||
|
// @scope.write Grants write access
|
||||||
|
// @scope.admin Grants read and write access to administrative information
|
||||||
|
|
||||||
|
// @securitydefinitions.oauth2.password OAuth2Password
|
||||||
|
// @tokenUrl https://example.com/oauth/token
|
||||||
|
// @scope.read Grants read access
|
||||||
|
// @scope.write Grants write access
|
||||||
|
// @scope.admin Grants read and write access to administrative information
|
||||||
|
|
||||||
|
// @securitydefinitions.oauth2.accessCode OAuth2AccessCode
|
||||||
|
// @tokenUrl https://example.com/oauth/token
|
||||||
|
// @authorizationUrl https://example.com/oauth/authorize
|
||||||
|
// @scope.admin Grants read and write access to administrative information
|
||||||
|
|
||||||
|
var (
|
||||||
|
port string
|
||||||
|
kafka string
|
||||||
|
num int
|
||||||
|
)
|
||||||
|
|
||||||
|
func initConfig() {
|
||||||
|
flag.StringVar(&kafka, "kafka", "", "kafka url")
|
||||||
|
flag.StringVar(&port, "port", "9502", "tcp port")
|
||||||
|
flag.IntVar(&num, "num", 10, "进程数量")
|
||||||
|
|
||||||
|
}
|
||||||
|
func main() {
|
||||||
|
//解析启动命令
|
||||||
|
initConfig()
|
||||||
|
opts := config.GetOptions()
|
||||||
|
|
||||||
|
if opts.ShowVersion {
|
||||||
|
fmt.Printf("%s\ncommit %s\nbuilt on %s\n", server.Version, server.BuildCommit, server.BuildDate)
|
||||||
|
os.Exit(0)
|
||||||
|
}
|
||||||
|
|
||||||
|
handleCmd(opts)
|
||||||
|
|
||||||
|
err := startServer(opts)
|
||||||
|
if err != nil {
|
||||||
|
fmt.Printf("server start error, %s\n", err)
|
||||||
|
os.Exit(1)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 执行(status|stop|restart)命令
|
||||||
|
func handleCmd(opts *config.Options) {
|
||||||
|
if opts.Cmd != "" {
|
||||||
|
pidFile := opts.GenPidFile()
|
||||||
|
err := server.HandleUserCmd(opts.Cmd, pidFile)
|
||||||
|
if err != nil {
|
||||||
|
fmt.Printf("Handle user command(%s) error, %s\n", opts.Cmd, err)
|
||||||
|
} else {
|
||||||
|
fmt.Printf("Handle user command(%s) succ \n ", opts.Cmd)
|
||||||
|
}
|
||||||
|
os.Exit(0)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func startServer(opts *config.Options) (err error) {
|
||||||
|
//加载配置
|
||||||
|
conf, err := config.Load(opts.ConfFile)
|
||||||
|
if err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
//引导程序
|
||||||
|
err = bootstrap.Bootstrap(conf)
|
||||||
|
if err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
//for i := 0; i < 2; i++ {
|
||||||
|
// mqs.MqManager.GetMqByName(common.MQ_KFK).Produce(common.ORDER_RESEND_TOPICAL, time.Now().Format(time.DateTime), 0)
|
||||||
|
//}
|
||||||
|
////
|
||||||
|
//select {}
|
||||||
|
pidFile := opts.GenPidFile()
|
||||||
|
config.GetConf().KafkaUrl = kafka
|
||||||
|
config.GetConf().OrderPort = port
|
||||||
|
config.GetConf().Num = num
|
||||||
|
fmt.Println(port, kafka, num)
|
||||||
|
//根据启动命令行参数，决定启动哪种服务模式
|
||||||
|
switch opts.App {
|
||||||
|
case "api":
|
||||||
|
err = server.StartHttp(pidFile, conf.Api, routes.RegisterRoute)
|
||||||
|
case "cron":
|
||||||
|
err = server.StartConsole(pidFile, console.RegisterSchedule)
|
||||||
|
case "job":
|
||||||
|
err = server.StartJob(pidFile, jobs.RegisterWorker)
|
||||||
|
case "command":
|
||||||
|
err = server.ExecuteCommand(opts.Command, console.RegisterCommand)
|
||||||
|
case "mq":
|
||||||
|
mq.StartQunueServer() //消费消息
|
||||||
|
default:
|
||||||
|
err = errors.New("no server start")
|
||||||
|
}
|
||||||
|
return
|
||||||
|
}
|
||||||
Loading…
Reference in New Issue