From 185cd6f82bdee9acf4da69f692655ba571ddf52d Mon Sep 17 00:00:00 2001
From: "meilin.huang" <954537473@qq.com>
Date: Wed, 14 Jan 2026 20:38:06 +0800
Subject: [PATCH] =?UTF-8?q?fix:=20postgres=E5=AF=BC=E5=87=BA=E8=B0=83?=
 =?UTF-8?q?=E6=95=B4=E7=AD=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 frontend/package.json                         |  2 +-
 frontend/src/common/config.ts                 |  2 +-
 .../src/views/ops/resource/ResourceOp.vue     |  2 +-
 server/go.mod                                 | 12 +--
 server/internal/ai/init/init.go               |  2 +-
 server/internal/db/application/db.go          | 22 ++---
 server/internal/db/application/db_transfer.go |  8 +-
 server/internal/db/dbm/dbi/column.go          | 34 ++++++++
 server/internal/db/dbm/dbi/conn.go            |  3 +-
 server/internal/db/dbm/dbi/transfer.go        |  3 +
 server/internal/db/dbm/mysql/metadata.go      |  2 +-
 server/internal/db/dbm/postgres/column.go     | 13 +--
 server/internal/db/dbm/postgres/helper.go     |  2 +-
 server/internal/db/dbm/postgres/meta.go       |  2 +-
 server/internal/msg/msgx/msgx.go              |  5 +-
 server/internal/msg/msgx/sender/ws.go         |  3 +-
 server/internal/pkg/config/app.go             |  2 +-
 server/internal/redis/rdm/conn.go             |  2 +-
 server/pkg/scheduler/scheduler.go             |  6 +-
 server/pkg/utils/collx/map.go                 | 85 +++++++++++++++++++
 server/pkg/ws/client_manager.go               | 63 +++++---------
 21 files changed, 186 insertions(+), 89 deletions(-)

diff --git a/frontend/package.json b/frontend/package.json
index a5db5ae0..4740ba7d 100644
--- a/frontend/package.json
+++ b/frontend/package.json
@@ -24,7 +24,7 @@
         "crypto-js": "^4.2.0",
         "dayjs": "^1.11.19",
         "echarts": "^6.0.0",
-        "element-plus": "^2.13.0",
+        "element-plus": "^2.13.1",
         "js-base64": "^3.7.8",
         "jsencrypt": "^3.5.4",
         "monaco-editor": "^0.55.1",
diff --git a/frontend/src/common/config.ts b/frontend/src/common/config.ts
index 13a45799..ff40a8e0 100644
--- a/frontend/src/common/config.ts
+++ b/frontend/src/common/config.ts
@@ -15,7 +15,7 @@ const config = {
     baseWsUrl: `${(window as any).globalConfig.BaseWsUrl || `${location.protocol == 'https:' ? 'wss:' : 'ws:'}//${getBaseApiUrl()}`}/api`,
 
     // 系统版本
-    version: 'v1.10.5',
+    version: 'v1.10.6',
 };
 
 export default config;
diff --git a/frontend/src/views/ops/resource/ResourceOp.vue b/frontend/src/views/ops/resource/ResourceOp.vue
index 41dfca72..43ac7369 100644
--- a/frontend/src/views/ops/resource/ResourceOp.vue
+++ b/frontend/src/views/ops/resource/ResourceOp.vue
@@ -2,7 +2,7 @@
     <div class="h-full">
         <el-splitter @resize="onResizeOpPanel">
             <el-splitter-panel size="24%" max="40%">
-                <el-card class="h-full flex" body-class="bg-(--el-bg-color) !p-0 flex flex-col w-full">
+                <el-card class="h-full flex" body-class="!p-0 flex flex-col w-full">
                     <div class="tag-tree-header flex justify-between items-center">
                         <el-input v-model="filterText" :placeholder="$t('tag.tagFilterPlaceholder')" clearable size="small" class="tag-tree-search w-full">
                             <template #prefix>
diff --git a/server/go.mod b/server/go.mod
index e7306050..4c030d86 100644
--- a/server/go.mod
+++ b/server/go.mod
@@ -36,7 +36,7 @@ require (
 	github.com/tidwall/gjson v1.18.0
 	github.com/veops/go-ansiterm v0.0.5
 	go.mongodb.org/mongo-driver/v2 v2.3.0 // mongo
-	golang.org/x/crypto v0.46.0 // ssh
+	golang.org/x/crypto v0.47.0 // ssh
 	golang.org/x/oauth2 v0.34.0
 	golang.org/x/sync v0.19.0
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1
@@ -132,11 +132,11 @@ require (
 	golang.org/x/arch v0.21.0 // indirect
 	golang.org/x/exp v0.0.0-20251002181428-27f1f14c8bb9 // indirect
 	golang.org/x/image v0.31.0 // indirect
-	golang.org/x/mod v0.30.0 // indirect
-	golang.org/x/net v0.47.0 // indirect
-	golang.org/x/sys v0.39.0 // indirect
-	golang.org/x/text v0.32.0 // indirect
-	golang.org/x/tools v0.39.0 // indirect
+	golang.org/x/mod v0.31.0 // indirect
+	golang.org/x/net v0.48.0 // indirect
+	golang.org/x/sys v0.40.0 // indirect
+	golang.org/x/text v0.33.0 // indirect
+	golang.org/x/tools v0.40.0 // indirect
 	google.golang.org/protobuf v1.36.10 // indirect
 	modernc.org/libc v1.66.10 // indirect
 	modernc.org/mathutil v1.7.1 // indirect
diff --git a/server/internal/ai/init/init.go b/server/internal/ai/init/init.go
index 100849b1..3985dd7c 100644
--- a/server/internal/ai/init/init.go
+++ b/server/internal/ai/init/init.go
@@ -1,8 +1,8 @@
 package init
 
 import (
-	"mayfly-go/internal/ai/api"
 	"mayfly-go/initialize"
+	"mayfly-go/internal/ai/api"
 )
 
 func init() {
diff --git a/server/internal/db/application/db.go b/server/internal/db/application/db.go
index 6cd7126f..750c64b3 100644
--- a/server/internal/db/application/db.go
+++ b/server/internal/db/application/db.go
@@ -291,14 +291,16 @@ func (d *dbAppImpl) DumpDb(ctx context.Context, reqParam *dto.DumpDb) error {
 	// 按表名排序
 	sort.Strings(tables)
 	quoteSchema := srcDialect.Quoter().Quote(dbConn.Info.CurrentSchema())
-	dumpHelper := targetDialect.GetDumpHelper()
-
+	targetDumpHelper := targetDialect.GetDumpHelper()
 	targetSqlGenerator := targetDialect.GetSQLGenerator()
-	targetDialectQuote := targetDialect.Quoter().Quote
+	// targetDialectQuote := targetDialect.Quoter().Quote
+
+	srcDialectQuote := srcDialect.Quoter().Quote
 	// 遍历获取每个表的信息
 	for _, tableName := range tables {
 		log(fmt.Sprintf("get table [%s] information...", tableName))
-		quoteTableName := targetDialectQuote(tableName)
+		// targetQuoteTableName := targetDialectQuote(tableName)
+		srcQuoteTableName := srcDialectQuote(tableName)
 
 		// 查询表信息，主要是为了查询表注释
 		tbs, err := srcMeta.GetTables(tableName)
@@ -332,22 +334,22 @@ func (d *dbAppImpl) DumpDb(ctx context.Context, reqParam *dto.DumpDb) error {
 			log(fmt.Sprintf("generate table [%s] DML...", tableName))
 			writer.WriteString(fmt.Sprintf("\n-- ----------------------------\n-- Data: %s \n-- ----------------------------\n", tableName))
 
-			dumpHelper.BeforeInsert(writer, quoteTableName)
+			targetDumpHelper.BeforeInsert(writer, tableName)
 
 			dataCount := 0
 			rows := make([][]any, 0)
-			_, err = dbConn.WalkTableRows(ctx, quoteTableName, func(row map[string]any, _ []*dbi.QueryColumn) error {
+			_, err = dbConn.WalkTableRows(ctx, srcQuoteTableName, func(row map[string]any, _ []*dbi.QueryColumn) error {
 				rowValues := make([]any, len(columns))
 				for i, col := range columns {
 					rowValues[i] = row[col.ColumnName]
 				}
 				rows = append(rows, rowValues)
 				dataCount++
-				if dataCount%500 != 0 {
+				if dataCount%100 != 0 {
 					return nil
 				}
 
-				beforeInsert := dumpHelper.BeforeInsertSql(quoteSchema, quoteTableName)
+				beforeInsert := targetDumpHelper.BeforeInsertSql(quoteSchema, tableName)
 				if beforeInsert != "" {
 					writer.WriteString(beforeInsert)
 				}
@@ -365,7 +367,7 @@ func (d *dbAppImpl) DumpDb(ctx context.Context, reqParam *dto.DumpDb) error {
 			}
 
 			if len(rows) > 0 {
-				beforeInsert := dumpHelper.BeforeInsertSql(quoteSchema, quoteTableName)
+				beforeInsert := targetDumpHelper.BeforeInsertSql(quoteSchema, tableName)
 				if beforeInsert != "" {
 					writer.WriteString(beforeInsert)
 				}
@@ -375,7 +377,7 @@ func (d *dbAppImpl) DumpDb(ctx context.Context, reqParam *dto.DumpDb) error {
 				}
 			}
 
-			dumpHelper.AfterInsert(writer, tableName, columns)
+			targetDumpHelper.AfterInsert(writer, tableName, columns)
 			progress(tableName, dbi.StmtTypeInsert, dataCount, true)
 		}
 
diff --git a/server/internal/db/application/db_transfer.go b/server/internal/db/application/db_transfer.go
index 54824164..5c7214ed 100644
--- a/server/internal/db/application/db_transfer.go
+++ b/server/internal/db/application/db_transfer.go
@@ -240,20 +240,16 @@ func (app *dbTransferAppImpl) transfer2Db(ctx context.Context, taskId uint64, lo
 					},
 				})
 				if err != nil {
+					app.Log(ctx, logId, fmt.Sprintf("db dump failed: %s", err.Error()))
 					pr.CloseWithError(err)
 					return
 				}
 			}()
 
-			if err != nil {
-				pw.CloseWithError(err)
-				app.EndTransfer(ctx, logId, taskId, "transfer table failed", err, nil)
-				return err
-			}
 			tx, _ := targetConn.Begin()
 			err = sqlparser.SQLSplit(pr, ';', func(stmt string) error {
 				if _, err := targetConn.TxExec(tx, stmt); err != nil {
-					app.EndTransfer(ctx, logId, taskId, fmt.Sprintf("执行sql出错: %s", stmt), err, nil)
+					app.Log(ctx, logId, fmt.Sprintf("sql exec failed: %s", stmt), err, nil)
 					pw.CloseWithError(err)
 					return err
 				}
diff --git a/server/internal/db/dbm/dbi/column.go b/server/internal/db/dbm/dbi/column.go
index eda54e11..e1e78bef 100644
--- a/server/internal/db/dbm/dbi/column.go
+++ b/server/internal/db/dbm/dbi/column.go
@@ -43,6 +43,7 @@ var DefaultDbDataType = NewDbDataType("string", DTString).WithCT(CTVarchar)
 type Column struct {
 	TableName     string  `json:"tableName"`     // 表名
 	ColumnName    string  `json:"columnName"`    // 列名
+	ColumnType    string  `json:"columnType"`    // 完整列类型，带有数据类型以及长度、精度等。如varchar(2000)，decimal(20,2)
 	DataType      string  `json:"dataType"`      // 数据类型
 	ColumnComment string  `json:"columnComment"` // 列备注
 	IsPrimaryKey  bool    `json:"isPrimaryKey"`  // 是否为主键
@@ -57,6 +58,10 @@ type Column struct {
 
 // GetColumnType 获取完整的列类型，拼接数据类型与长度等。如varchar(2000)，decimal(20,2)
 func (c *Column) GetColumnType() string {
+	if c.ColumnType != "" {
+		return c.ColumnType
+	}
+
 	if c.CharMaxLength > 0 {
 		return fmt.Sprintf("%s(%d)", c.DataType, c.CharMaxLength)
 	}
@@ -178,6 +183,8 @@ func SQLValueBool(val any) string {
 	return fmt.Sprintf("%v", cast.ToBool(val))
 }
 
+// SQLValueString 转换为SQL字符串值，处理特殊字符与单引号
+// 适用于普通字符串类型，会将双引号等特殊字符进行转义
 func SQLValueString(val any) string {
 	if val == nil {
 		return NULL
@@ -197,6 +204,26 @@ func SQLValueString(val any) string {
 	return fmt.Sprintf("'%s'", quoted)
 }
 
+// SQLValuePreserveSpecialChars 转换为SQL字符串值，保留特殊字符如双引号、换行符等
+// 仅转义单引号为两个单引号（SQL标准转义方式）
+func SQLValuePreserveSpecialChars(val any) string {
+	if val == nil {
+		return NULL
+	}
+
+	strVal, ok := val.(string)
+	if !ok {
+		return fmt.Sprintf("%v", val)
+	}
+
+	// 直接处理 SQL 特殊字符，保留双引号和其他字符
+	// 只转义单引号为两个单引号（SQL 标准转义方式）
+	escapedStr := strings.ReplaceAll(strVal, "'", "''")
+	
+	// 返回 SQL 字符串，保持原始的双引号、换行符等
+	return fmt.Sprintf("'%s'", escapedStr)
+}
+
 var (
 	DTBit = &DataType{
 		Name:     "bit",
@@ -266,6 +293,13 @@ var (
 		SQLValue: SQLValueString,
 	}
 
+	// DTStringPreserveSpecial 用于需要保留双引号换行符等特殊字符的字符串类型
+	DTStringPreserveSpecial = &DataType{
+		Name:     "string",
+		Valuer:   ValuerString,
+		SQLValue: SQLValuePreserveSpecialChars,
+	}
+
 	DTDate = &DataType{
 		Name:     "date",
 		Valuer:   ValuerDate,
diff --git a/server/internal/db/dbm/dbi/conn.go b/server/internal/db/dbm/dbi/conn.go
index 0f7fbe18..fa2d539b 100644
--- a/server/internal/db/dbm/dbi/conn.go
+++ b/server/internal/db/dbm/dbi/conn.go
@@ -49,7 +49,7 @@ func (d *DbConn) Ping() error {
 	stats := d.db.Stats()
 	logx.Debugf("[%s] db stats -> open: %d, idle: %d,  inUse: %d, maxOpen: %d", d.Info.Name, stats.OpenConnections, stats.Idle, stats.InUse, stats.MaxOpenConnections)
 	if stats.OpenConnections == 0 {
-		logx.Info("db stats: no open connections")
+		logx.Infof("[%s] db stats: no open connections", d.Info.Name)
 	}
 
 	return d.db.Ping()
@@ -77,6 +77,7 @@ func (qc *QueryColumn) getValuePtr() any {
 	return qc.valuer.NewValuePtr()
 }
 
+// value 获取列值
 func (qc *QueryColumn) value() any {
 	return qc.valuer.Value()
 }
diff --git a/server/internal/db/dbm/dbi/transfer.go b/server/internal/db/dbm/dbi/transfer.go
index 375d43c4..ebeb6649 100644
--- a/server/internal/db/dbm/dbi/transfer.go
+++ b/server/internal/db/dbm/dbi/transfer.go
@@ -131,6 +131,9 @@ func ConvToTargetDbColumn(srcDbType DbType, targetDbType DbType, targetDialect D
 		return nil
 	}
 
+	// 需要转换至异构数据库时，需要将该字段清空，否则如mysql可以查出该值，其他数据库可能不行，会导致Column.GetColumnType错误。
+	column.ColumnType = ""
+
 	srcMap := commonTypeConverters[srcDbType]
 	if srcMap == nil {
 		return fmt.Errorf("src database type [%s] not suport transfer", srcDbType)
diff --git a/server/internal/db/dbm/mysql/metadata.go b/server/internal/db/dbm/mysql/metadata.go
index ae4c1d69..0374ca23 100644
--- a/server/internal/db/dbm/mysql/metadata.go
+++ b/server/internal/db/dbm/mysql/metadata.go
@@ -97,10 +97,10 @@ func (md *MysqlMetadata) GetColumns(tableNames ...string) ([]dbi.Column, error)
 
 	columns := make([]dbi.Column, 0)
 	for _, re := range res {
-
 		column := dbi.Column{
 			TableName:     cast.ToString(re["tableName"]),
 			ColumnName:    cast.ToString(re["columnName"]),
+			ColumnType:    cast.ToString(re["columnType"]),
 			DataType:      cast.ToString(re["dataType"]),
 			ColumnComment: cast.ToString(re["columnComment"]),
 			Nullable:      cast.ToString(re["nullable"]) == "YES",
diff --git a/server/internal/db/dbm/postgres/column.go b/server/internal/db/dbm/postgres/column.go
index ef2da54e..88aecf44 100644
--- a/server/internal/db/dbm/postgres/column.go
+++ b/server/internal/db/dbm/postgres/column.go
@@ -18,12 +18,13 @@ var (
 
 	Money = dbi.NewDbDataType("money", dbi.DTString).WithCT(dbi.CTVarchar)
 
-	Char    = dbi.NewDbDataType("char", dbi.DTString).WithCT(dbi.CTChar)
-	Nchar   = dbi.NewDbDataType("nchar", dbi.DTString).WithCT(dbi.CTVarchar)
-	Varchar = dbi.NewDbDataType("varchar", dbi.DTString).WithCT(dbi.CTVarchar)
-	Text    = dbi.NewDbDataType("text", dbi.DTString).WithCT(dbi.CTText).WithFixColumn(dbi.ClearCharMaxLength)
-	Json    = dbi.NewDbDataType("json", dbi.DTString).WithCT(dbi.CTJSON).WithFixColumn(dbi.ClearCharMaxLength)
-	Bytea   = dbi.NewDbDataType("bytea", dbi.DTString).WithCT(dbi.CTBinary)
+	Char    = dbi.NewDbDataType("char", dbi.DTStringPreserveSpecial).WithCT(dbi.CTChar)
+	Nchar   = dbi.NewDbDataType("nchar", dbi.DTStringPreserveSpecial).WithCT(dbi.CTVarchar)
+	Varchar = dbi.NewDbDataType("varchar", dbi.DTStringPreserveSpecial).WithCT(dbi.CTVarchar)
+	Text    = dbi.NewDbDataType("text", dbi.DTStringPreserveSpecial).WithCT(dbi.CTText).WithFixColumn(dbi.ClearCharMaxLength)
+	Json    = dbi.NewDbDataType("json", dbi.DTStringPreserveSpecial).WithCT(dbi.CTJSON).WithFixColumn(dbi.ClearCharMaxLength)
+	Jsonb   = dbi.NewDbDataType("jsonb", dbi.DTStringPreserveSpecial).WithCT(dbi.CTJSON).WithFixColumn(dbi.ClearCharMaxLength)
+	Bytea   = dbi.NewDbDataType("bytea", dbi.DTStringPreserveSpecial).WithCT(dbi.CTBinary)
 
 	Date      = dbi.NewDbDataType("date", dbi.DTDate).WithCT(dbi.CTDate).WithFixColumn(dbi.ClearCharMaxLength)
 	Time      = dbi.NewDbDataType("time", dbi.DTTime).WithCT(dbi.CTTime).WithFixColumn(dbi.ClearCharMaxLength)
diff --git a/server/internal/db/dbm/postgres/helper.go b/server/internal/db/dbm/postgres/helper.go
index e7197757..0ab545d0 100644
--- a/server/internal/db/dbm/postgres/helper.go
+++ b/server/internal/db/dbm/postgres/helper.go
@@ -31,7 +31,7 @@ func (dh *DumpHelper) AfterInsert(writer io.Writer, tableName string, columns []
 	// 设置自增序列当前值
 	for _, column := range columns {
 		if column.AutoIncrement {
-			seq := fmt.Sprintf("SELECT setval('%s_%s_seq', (SELECT max(%s) FROM %s));\n", tableName, column.ColumnName, column.ColumnName, tableName)
+			seq := fmt.Sprintf("SELECT setval('%s_%s_seq', (SELECT max(%s) FROM \"%s\"));\n", tableName, column.ColumnName, column.ColumnName, tableName)
 			writer.Write([]byte(seq))
 		}
 	}
diff --git a/server/internal/db/dbm/postgres/meta.go b/server/internal/db/dbm/postgres/meta.go
index 29058cac..87646689 100644
--- a/server/internal/db/dbm/postgres/meta.go
+++ b/server/internal/db/dbm/postgres/meta.go
@@ -101,7 +101,7 @@ func (pm *Meta) GetDbDataTypes() []*dbi.DbDataType {
 	return collx.AsArray(
 		Bool, Int2, Int4, Int8, Numeric, Decimal, Smallserial, Serial, Bigserial, Largeserial,
 		Money,
-		Char, Nchar, Varchar, Text, Json,
+		Char, Nchar, Varchar, Text, Json, Jsonb,
 		Date, Time, Timestamp,
 		Bytea,
 	)
diff --git a/server/internal/msg/msgx/msgx.go b/server/internal/msg/msgx/msgx.go
index 539414ab..1ee51c60 100644
--- a/server/internal/msg/msgx/msgx.go
+++ b/server/internal/msg/msgx/msgx.go
@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"mayfly-go/pkg/utils/collx"
-	"sync"
 )
 
 type MsgType int8
@@ -71,7 +70,7 @@ type MsgSender interface {
 	Send(ctx context.Context, channel *Channel, msg *Msg) error
 }
 
-var messageSenders sync.Map
+var messageSenders *collx.SM[ChannelType, MsgSender] = collx.NewSM[ChannelType, MsgSender]()
 
 // RegisterMsgSender 注册消息发送器
 func RegisterMsgSender(channel ChannelType, sender MsgSender) {
@@ -84,5 +83,5 @@ func GetMsgSender(channel ChannelType) (MsgSender, error) {
 	if !ok {
 		return nil, fmt.Errorf("unsupported message channel %s", channel)
 	}
-	return sender.(MsgSender), nil
+	return sender, nil
 }
diff --git a/server/internal/msg/msgx/sender/ws.go b/server/internal/msg/msgx/sender/ws.go
index fa3819e4..6e973944 100644
--- a/server/internal/msg/msgx/sender/ws.go
+++ b/server/internal/msg/msgx/sender/ws.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"mayfly-go/internal/msg/msgx"
 	"mayfly-go/pkg/i18n"
+	"mayfly-go/pkg/utils/collx"
 	"mayfly-go/pkg/utils/stringx"
 	"mayfly-go/pkg/ws"
 
@@ -27,7 +28,7 @@ func (e WsSender) Send(ctx context.Context, channel *msgx.Channel, msg *msgx.Msg
 		}
 	}
 
-	jsonMsg := msg.TmplExtra
+	jsonMsg := collx.CopyM(msg.TmplExtra)
 	jsonMsg["msg"] = content
 	jsonMsg["title"] = msg.Title
 	jsonMsg["params"] = msg.Params
diff --git a/server/internal/pkg/config/app.go b/server/internal/pkg/config/app.go
index 3829def3..a8cb7e67 100644
--- a/server/internal/pkg/config/app.go
+++ b/server/internal/pkg/config/app.go
@@ -4,7 +4,7 @@ import "fmt"
 
 const (
 	AppName = "mayfly-go"
-	Version = "v1.10.5"
+	Version = "v1.10.6"
 )
 
 func GetAppInfo() string {
diff --git a/server/internal/redis/rdm/conn.go b/server/internal/redis/rdm/conn.go
index 0eab3fab..e4249284 100644
--- a/server/internal/redis/rdm/conn.go
+++ b/server/internal/redis/rdm/conn.go
@@ -49,7 +49,7 @@ func (r *RedisConn) Ping() error {
 	stats := r.Cli.PoolStats()
 	logx.Debugf("[%s] redis stats -> open: %d, idle: %d", r.Info.Name, stats.TotalConns, stats.IdleConns)
 	if stats.TotalConns == 0 {
-		logx.Info("redis stats: no open connections")
+		logx.Infof("[%s] redis stats: no open connections", r.Info.Name)
 	}
 
 	cmd := r.Cli.Ping(context.Background())
diff --git a/server/pkg/scheduler/scheduler.go b/server/pkg/scheduler/scheduler.go
index 7fbbd02e..e9d459a6 100644
--- a/server/pkg/scheduler/scheduler.go
+++ b/server/pkg/scheduler/scheduler.go
@@ -3,7 +3,7 @@ package scheduler
 import (
 	"errors"
 	"mayfly-go/pkg/logx"
-	"sync"
+	"mayfly-go/pkg/utils/collx"
 
 	"github.com/robfig/cron/v3"
 )
@@ -14,7 +14,7 @@ func init() {
 
 var (
 	cronService = cron.New(cron.WithSeconds())
-	key2IdMap   sync.Map
+	key2IdMap  *collx.SM[string, cron.EntryID] = collx.NewSM[string, cron.EntryID]()
 )
 
 func Start() {
@@ -35,7 +35,7 @@ func RemoveByKey(key string) {
 	logx.Debugf("remove cron func => [key = %s]", key)
 	id, ok := key2IdMap.Load(key)
 	if ok {
-		Remove(id.(cron.EntryID))
+		Remove(id)
 		key2IdMap.Delete(key)
 	}
 }
diff --git a/server/pkg/utils/collx/map.go b/server/pkg/utils/collx/map.go
index f6977f0b..d36e0897 100644
--- a/server/pkg/utils/collx/map.go
+++ b/server/pkg/utils/collx/map.go
@@ -4,6 +4,7 @@ import (
 	"database/sql/driver"
 	"encoding/json"
 	"mayfly-go/pkg/utils/anyx"
+	"sync"
 
 	"github.com/spf13/cast"
 )
@@ -50,6 +51,90 @@ func (m M) GetBool(key string) bool {
 	return cast.ToBool(m[key])
 }
 
+// Copy 复制一个新的map，在有线程并发读写时使用，避免并发读写导致数据不一致
+func CopyM(m M) M {
+	newMap := make(M)
+	for k, v := range m {
+		newMap[k] = v
+	}
+	return newMap
+}
+
+/******************* thread-safe map *******************/
+
+// SM 是一个线程安全的 map 实现，基于 sync.Map 并支持泛型
+type SM[K comparable, V any] struct {
+	m sync.Map
+}
+
+// NewSM 创建一个新的线程安全 map 实例
+func NewSM[K comparable, V any]() *SM[K, V] {
+	return &SM[K, V]{}
+}
+
+// Load 方法返回存储在 map 中键为 k 的值，如果不存在则返回零值和 false
+func (sm *SM[K, V]) Load(k K) (V, bool) {
+	v, ok := sm.m.Load(k)
+	if !ok {
+		var zero V
+		return zero, false
+	}
+	return v.(V), true
+}
+
+// Store 在 map 中存储键值对
+func (sm *SM[K, V]) Store(k K, v V) {
+	sm.m.Store(k, v)
+}
+
+// LoadOrStore 返回键 k 对应的值，如果不存在则存储指定的值
+func (sm *SM[K, V]) LoadOrStore(k K, v V) (actual V, loaded bool) {
+	actualVal, loaded := sm.m.LoadOrStore(k, v)
+	return actualVal.(V), loaded
+}
+
+// LoadAndDelete 从 map 中删除键值对并返回之前的值
+func (sm *SM[K, V]) LoadAndDelete(k K) (V, bool) {
+	v, ok := sm.m.LoadAndDelete(k)
+	if !ok {
+		var zero V
+		return zero, false
+	}
+	return v.(V), true
+}
+
+// Delete 从 map 中删除键值对
+func (sm *SM[K, V]) Delete(k K) {
+	sm.m.Delete(k)
+}
+
+// Range 遍历 map 中的所有键值对
+func (sm *SM[K, V]) Range(f func(k K, v V) bool) {
+	sm.m.Range(func(k, v any) bool {
+		return f(k.(K), v.(V))
+	})
+}
+
+// Len 返回 map 中键值对的数量
+func (sm *SM[K, V]) Len() int {
+	count := 0
+	sm.Range(func(_ K, _ V) bool {
+		count++
+		return true
+	})
+	return count
+}
+
+// Values 返回 map 中所有的值
+func (sm *SM[K, V]) Values() []V {
+	values := make([]V, 0)
+	sm.Range(func(_ K, v V) bool {
+		values = append(values, v)
+		return true
+	})
+	return values
+}
+
 /******************* M db driver *******************/
 func (j *M) Scan(value any) error {
 	if v, ok := value.([]byte); ok && len(v) > 0 {
diff --git a/server/pkg/ws/client_manager.go b/server/pkg/ws/client_manager.go
index 02eb1c78..77cccc90 100644
--- a/server/pkg/ws/client_manager.go
+++ b/server/pkg/ws/client_manager.go
@@ -2,7 +2,7 @@ package ws
 
 import (
 	"mayfly-go/pkg/logx"
-	"sync"
+	"mayfly-go/pkg/utils/collx"
 	"time"
 )
 
@@ -11,58 +11,44 @@ const heartbeatInterval = 25 * time.Second
 
 // UserClient 用户全部的连接
 type UserClient struct {
-	clients map[string]*Client // key->clientId, value->Client
-	mutex   sync.RWMutex
+	clients *collx.SM[string, *Client] // key->clientId, value->Client
 }
 
 func NewUserClient() *UserClient {
 	return &UserClient{
-		clients: make(map[string]*Client),
+		clients: collx.NewSM[string, *Client](),
 	}
 }
 
 // AllClients 获取全部的连接
 func (ucs *UserClient) AllClients() []*Client {
-	ucs.mutex.RLock()
-	defer ucs.mutex.RUnlock()
-	result := make([]*Client, 0, len(ucs.clients))
-	for _, client := range ucs.clients {
-		result = append(result, client)
-	}
-	return result
+	return ucs.clients.Values()
 }
 
 // GetByCid 获取指定客户端ID的客户端
 func (ucs *UserClient) GetByCid(clientId string) *Client {
-	ucs.mutex.RLock()
-	defer ucs.mutex.RUnlock()
-	return ucs.clients[clientId]
+	cli, _ :=  ucs.clients.Load(clientId)
+	return cli
 }
 
 // AddClient 添加客户端
 func (ucs *UserClient) AddClient(client *Client) {
-	ucs.mutex.Lock()
-	defer ucs.mutex.Unlock()
-	ucs.clients[client.ClientId] = client
+	ucs.clients.Store(client.ClientId, client)
 }
 
 // DeleteByCid 删除指定客户端ID的客户端
 func (ucs *UserClient) DeleteByCid(clientId string) {
-	ucs.mutex.Lock()
-	defer ucs.mutex.Unlock()
-	delete(ucs.clients, clientId)
+	ucs.clients.Delete(clientId)
 }
 
 // Count 返回客户端数量
 func (ucs *UserClient) Count() int {
-	ucs.mutex.RLock()
-	defer ucs.mutex.RUnlock()
-	return len(ucs.clients)
+	return ucs.clients.Len()
 }
 
 // 连接管理
 type ClientManager struct {
-	UserClientMap sync.Map // 全部的用户连接, key->userid, value->*UserClient
+	UserClientMap *collx.SM[UserId, *UserClient] // 全部的用户连接, key->userid, value->*UserClient
 
 	ConnectChan    chan *Client // 连接处理
 	DisConnectChan chan *Client // 断开连接处理
@@ -71,7 +57,7 @@ type ClientManager struct {
 
 func NewClientManager() (clientManager *ClientManager) {
 	return &ClientManager{
-		UserClientMap:  sync.Map{},
+		UserClientMap:  collx.NewSM[UserId, *UserClient](),
 		ConnectChan:    make(chan *Client, 10),
 		DisConnectChan: make(chan *Client, 10),
 		MsgChan:        make(chan *Msg, 100),
@@ -118,10 +104,8 @@ func (manager *ClientManager) CloseByUid(userId UserId) {
 // 获取所有的客户端
 func (manager *ClientManager) AllUserClient() map[UserId]*UserClient {
 	result := make(map[UserId]*UserClient)
-	manager.UserClientMap.Range(func(key, value any) bool {
-		userId := key.(UserId)
-		userClients := value.(*UserClient)
-		result[userId] = userClients
+	manager.UserClientMap.Range(func(userId UserId, userClient *UserClient) bool {
+		result[userId] = userClient
 		return true
 	})
 	return result
@@ -130,7 +114,7 @@ func (manager *ClientManager) AllUserClient() map[UserId]*UserClient {
 // 通过userId获取用户所有客户端信息
 func (manager *ClientManager) GetByUid(userId UserId) *UserClient {
 	if value, ok := manager.UserClientMap.Load(userId); ok {
-		return value.(*UserClient)
+		return value
 	}
 	return nil
 }
@@ -145,12 +129,7 @@ func (manager *ClientManager) GetByUidAndCid(uid UserId, clientId string) *Clien
 
 // 客户端数量
 func (manager *ClientManager) Count() int {
-	count := 0
-	manager.UserClientMap.Range(func(key, value any) bool {
-		count++
-		return true
-	})
-	return count
+	return manager.UserClientMap.Len()
 }
 
 // 发送json数据给指定用户
@@ -199,9 +178,7 @@ func (manager *ClientManager) HeartbeatTimer() {
 		for {
 			<-ticker.C
 			//发送心跳
-			manager.UserClientMap.Range(func(key, value any) bool {
-				userId := key.(UserId)
-				userClient := value.(*UserClient)
+			manager.UserClientMap.Range(func(userId UserId, userClient *UserClient) bool {
 				for _, cli := range userClient.AllClients() {
 					if cli == nil || cli.WsConn == nil {
 						continue
@@ -243,8 +220,7 @@ func (manager *ClientManager) doDisconnect(client *Client) {
 func (manager *ClientManager) addUserClient2Map(client *Client) {
 	// 先尝试加载现有的UserClients
 	if value, ok := manager.UserClientMap.Load(client.UserId); ok {
-		userClient := value.(*UserClient)
-		userClient.AddClient(client)
+		value.AddClient(client)
 	} else {
 		// 创建新的UserClients
 		userClient := NewUserClient()
@@ -255,10 +231,9 @@ func (manager *ClientManager) addUserClient2Map(client *Client) {
 
 func (manager *ClientManager) delUserClient4Map(client *Client) {
 	if value, ok := manager.UserClientMap.Load(client.UserId); ok {
-		userClient := value.(*UserClient)
-		userClient.DeleteByCid(client.ClientId)
+		value.DeleteByCid(client.ClientId)
 		// 如果用户所有客户端都关闭，则移除manager中的UserClientsMap值
-		if userClient.Count() == 0 {
+		if value.Count() == 0 {
 			manager.UserClientMap.Delete(client.UserId)
 		}
 	}