272 lines
7.1 KiB
Go
272 lines
7.1 KiB
Go
package bitget
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"go-admin/app/admin/models"
|
||
"go-admin/common/const/rediskey"
|
||
"go-admin/common/database"
|
||
"go-admin/common/global"
|
||
"go-admin/common/helper"
|
||
"go-admin/common/storage"
|
||
"go-admin/config/serverinit"
|
||
"go-admin/pkg/utility"
|
||
"go-admin/services/bitgetservice"
|
||
"go-admin/services/fileservice"
|
||
"os"
|
||
"os/signal"
|
||
"strconv"
|
||
"time"
|
||
|
||
ext "go-admin/config"
|
||
|
||
"github.com/bytedance/sonic"
|
||
"github.com/go-admin-team/go-admin-core/config/source/file"
|
||
"github.com/go-admin-team/go-admin-core/logger"
|
||
"github.com/go-admin-team/go-admin-core/sdk"
|
||
"github.com/go-admin-team/go-admin-core/sdk/config"
|
||
"github.com/go-admin-team/go-admin-core/sdk/pkg"
|
||
"github.com/spf13/cobra"
|
||
"gorm.io/gorm"
|
||
)
|
||
|
||
var (
|
||
configYml string
|
||
StartCmd = &cobra.Command{
|
||
Use: "bitget",
|
||
Short: "Start Bitget service (market data + user private)",
|
||
Example: "go-admin bitget -c config/settings.yml",
|
||
SilenceUsage: true,
|
||
PreRun: func(cmd *cobra.Command, args []string) {
|
||
setup()
|
||
},
|
||
RunE: func(cmd *cobra.Command, args []string) error {
|
||
return run()
|
||
},
|
||
}
|
||
)
|
||
|
||
func init() {
|
||
StartCmd.PersistentFlags().StringVarP(&configYml, "config", "c", "config/settings.yml", "Start server with provided configuration file")
|
||
}
|
||
|
||
func setup() {
|
||
// 注入配置扩展项
|
||
config.ExtendConfig = &ext.ExtConfig
|
||
//1. 读取配置
|
||
config.Setup(
|
||
file.NewSource(file.WithPath(configYml)),
|
||
database.Setup,
|
||
storage.Setup,
|
||
)
|
||
//注册监听函数
|
||
queue := sdk.Runtime.GetMemoryQueue("")
|
||
queue.Register(global.LoginLog, models.SaveLoginLog)
|
||
queue.Register(global.OperateLog, models.SaveOperaLog)
|
||
go queue.Run()
|
||
|
||
usageStr := `starting Bitget service (market data + user private)...`
|
||
logger.Info(usageStr)
|
||
}
|
||
|
||
func run() error {
|
||
ctx, cancel := context.WithCancel(context.Background())
|
||
defer cancel()
|
||
|
||
var db *gorm.DB
|
||
dbs := sdk.Runtime.GetDb()
|
||
|
||
for _, item := range dbs {
|
||
db = item
|
||
break
|
||
}
|
||
|
||
defaultInit(db, ctx)
|
||
|
||
utility.SafeGo(func() {
|
||
// 启动定时任务
|
||
clearLogJob(db, ctx)
|
||
})
|
||
|
||
//自动重启websocket
|
||
utility.SafeGo(func() {
|
||
reconnect(ctx)
|
||
})
|
||
|
||
//动态用户连接管理
|
||
utility.SafeGo(func() {
|
||
manageBitgetUserConnections(ctx)
|
||
})
|
||
|
||
// 等待中断信号以优雅地关闭服务器（设置 5 秒的超时时间）
|
||
quit := make(chan os.Signal, 1)
|
||
signal.Notify(quit, os.Interrupt)
|
||
<-quit
|
||
|
||
fmt.Printf("%s Shutdown Server ... \r\n", pkg.GetCurrentTimeStr())
|
||
logger.Info("Server exiting")
|
||
|
||
return nil
|
||
}
|
||
|
||
func defaultInit(db *gorm.DB, ctx context.Context) {
|
||
//初始化 默认redis
|
||
helper.InitDefaultRedis(ext.ExtConfig.Redis.Addr, ext.ExtConfig.Redis.Password, ext.ExtConfig.Redis.Db)
|
||
helper.InitLockRedisConn(ext.ExtConfig.Redis.Addr, ext.ExtConfig.Redis.Password, strconv.Itoa(ext.ExtConfig.Redis.Db))
|
||
|
||
err := helper.DefaultRedis.Ping()
|
||
|
||
if err != nil {
|
||
logger.Error("初始化redis失败!请检查配置")
|
||
_, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||
cancel()
|
||
} else {
|
||
logger.Info("redis初始化成功")
|
||
}
|
||
|
||
//初始化Bitget服务（行情 + 用户私有）
|
||
logger.Info("=== 启动Bitget完整服务 ===")
|
||
|
||
// 启动Bitget行情服务
|
||
logger.Info("启动Bitget行情服务...")
|
||
if err := serverinit.BitgetMarketInit(); err != nil {
|
||
logger.Errorf("Bitget行情服务启动失败: %v", err)
|
||
}
|
||
|
||
// 初始化Bitget用户连接
|
||
logger.Info("启动Bitget用户服务...")
|
||
initBitgetUserConnections(db, ctx)
|
||
|
||
logger.Info("=== Bitget完整服务已启动 ===")
|
||
}
|
||
|
||
// 初始化Bitget用户连接
|
||
func initBitgetUserConnections(db *gorm.DB, ctx context.Context) {
|
||
var list []models.LineApiUser
|
||
err := db.Model(&models.LineApiUser{}).Where("open_status=1 AND exchange_type =?", global.EXCHANGE_BITGET).Find(&list).Error
|
||
|
||
if err != nil {
|
||
logger.Infof("获取Bitget用户api失败: %v", err)
|
||
return
|
||
}
|
||
|
||
logger.Infof("找到 %d 个Bitget用户，开始建立连接...", len(list))
|
||
for _, item := range list {
|
||
// 遵循私有WebSocket密钥传递规范：外部传入API密钥参数
|
||
createBitgetUserConnection(item)
|
||
}
|
||
}
|
||
|
||
// 创建Bitget用户连接（遵循密钥传递规范）
|
||
func createBitgetUserConnection(apiUser models.LineApiUser) {
|
||
logger.Infof("为Bitget用户 %d 建立私有连接", apiUser.Id)
|
||
|
||
// 现货WebSocket (wsType=0)
|
||
spotManager := bitgetservice.GetOrCreateBitgetWebSocketManager(
|
||
0, // 现货
|
||
apiUser.ApiKey,
|
||
apiUser.ApiSecret,
|
||
apiUser.Passphrase,
|
||
"", // proxyType - 从配置获取
|
||
"", // proxyAddress - 从配置获取
|
||
)
|
||
spotManager.Start()
|
||
|
||
// 合约WebSocket (wsType=1)
|
||
futuresManager := bitgetservice.GetOrCreateBitgetWebSocketManager(
|
||
1, // 合约
|
||
apiUser.ApiKey,
|
||
apiUser.ApiSecret,
|
||
apiUser.Passphrase,
|
||
"", // proxyType
|
||
"", // proxyAddress
|
||
)
|
||
futuresManager.Start()
|
||
|
||
logger.Infof("Bitget用户 %d 私有连接已建立", apiUser.Id)
|
||
}
|
||
|
||
// 动态Bitget用户连接管理
|
||
func manageBitgetUserConnections(ctx context.Context) {
|
||
ticker := time.NewTicker(time.Second * 5)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
logger.Info("Bitget用户连接管理器退出")
|
||
return
|
||
case <-ticker.C:
|
||
deleteKeys, _ := helper.DefaultRedis.GetAllList(rediskey.ApiUserDeleteList)
|
||
activeApis, _ := helper.DefaultRedis.GetAllList(rediskey.ApiUserActiveList)
|
||
|
||
//移除Bitget连接
|
||
for _, item := range deleteKeys {
|
||
logger.Infof("移除Bitget用户连接: %s", item)
|
||
|
||
// 停止现货WebSocket
|
||
bitgetservice.StopBitgetWebSocketManager(0, item)
|
||
// 停止合约WebSocket
|
||
bitgetservice.StopBitgetWebSocketManager(1, item)
|
||
|
||
if _, err := helper.DefaultRedis.LRem(rediskey.ApiUserDeleteList, item); err != nil {
|
||
logger.Errorf("移除待关闭websocket失败: %v", err)
|
||
}
|
||
}
|
||
|
||
//连接Bitget用户websocket
|
||
var apiUser models.LineApiUser
|
||
for _, item := range activeApis {
|
||
if item == "" {
|
||
continue
|
||
}
|
||
|
||
if err := sonic.Unmarshal([]byte(item), &apiUser); err != nil {
|
||
logger.Errorf("解析用户信息失败: %v", err)
|
||
continue
|
||
}
|
||
|
||
// 只处理Bitget用户
|
||
if apiUser.Id > 0 && apiUser.ExchangeType == global.EXCHANGE_BITGET {
|
||
logger.Info("为Bitget用户 %d 建立连接", apiUser.Id)
|
||
createBitgetUserConnection(apiUser)
|
||
|
||
if _, err := helper.DefaultRedis.LRem(rediskey.ApiUserActiveList, item); err != nil {
|
||
logger.Errorf("删除待触发的apiUser失败: %v", err)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// 定时清理日志
|
||
func clearLogJob(db *gorm.DB, ctx context.Context) {
|
||
ticker := time.NewTicker(time.Hour * 1)
|
||
defer ticker.Stop()
|
||
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
fileservice.ClearLogs(db)
|
||
}
|
||
}
|
||
|
||
// 定时重连websocket
|
||
func reconnect(ctx context.Context) error {
|
||
ticker := time.NewTicker(time.Hour * 1)
|
||
defer ticker.Stop()
|
||
|
||
select {
|
||
case <-ctx.Done():
|
||
return nil
|
||
case <-ticker.C:
|
||
// 这里可以添加Bitget连接的重连逻辑
|
||
// 类似于Binance的重连机制
|
||
logger.Info("检查Bitget连接状态...")
|
||
}
|
||
|
||
return nil
|
||
}
|