refactor(worker): 改进随机数生成并优化任务处理结构

使用crypto/rand替代math/rand生成更安全的随机盐值
重构Worker和Task结构体字段顺序，提高可读性
提取selectWorker方法复用worker选择逻辑
统一使用ServiceUnavailableMsg常量

Author: go-enols

.github/workflows/release.yml

@@ -42,6 +42,11 @@ jobs:
           npm install go-server-sdk@${{ steps.get_version.outputs.VERSION }}
           ```
           
+          **Rust SDK:**
+          ```bash
+          cargo add go-server-sdk@${{ steps.get_version.outputs.VERSION }}
+          ```
+          
           **Go SDK:**
           ```bash
           go get github.com/go-enols/go-server@${{ steps.get_version.outputs.VERSION }}
@@ -55,6 +60,7 @@ jobs:
           
           - [PyPI Package](https://pypi.org/project/go-server-sdk/${{ steps.get_version.outputs.VERSION }}/)
           - [NPM Package](https://www.npmjs.com/package/go-server-sdk/v/${{ steps.get_version.outputs.VERSION }})
+          - [Crates.io Package](https://crates.io/crates/go-server-sdk/${{ steps.get_version.outputs.VERSION }})
           - [Documentation](https://github.com/go-enols/go-server/blob/main/README.md)
         draft: false
         prerelease: false
@@ -147,6 +153,12 @@ jobs:
         node-version: '18'
         registry-url: 'https://registry.npmjs.org'
 
+    - name: Set up Rust
+      uses: actions-rs/toolchain@v1
+      with:
+        toolchain: stable
+        override: true
+    
     - name: Install Python build dependencies
       run: |
         python -m pip install --upgrade pip
@@ -167,4 +179,12 @@ jobs:
         npm ci
         npm publish --access public
       env:
-        NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }}
+        NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }}
+    
+    - name: Login to crates.io
+      working-directory: ./rust-sdk
+      run: cargo login ${{ secrets.RUST_TOKEN }}
+    
+    - name: Publish Rust package
+      working-directory: ./rust-sdk
+      run: cargo publish

.gitignore

@@ -29,4 +29,5 @@ dist/
 go_server_sdk.egg-info/
 pyproject.toml
 
-target
+target
+*.ps1

go-sdk/worker/call.go

@@ -2,12 +2,12 @@
 package worker
 
 import (
+	"crypto/rand"
 	"encoding/json"
 	"errors"
 	"fmt"
-	"math/rand"
+	"math/big"
 	"reflect"
-	"time"
 
 	"github.com/go-enols/go-server/go-sdk/scheduler"
 )
@@ -31,7 +31,11 @@ func Call(host, method string, params, out interface{}, encryptionKey ...string)
 	// 如果提供了加密密钥，使用加密方式调用
 	if len(encryptionKey) > 0 && encryptionKey[0] != "" {
 		// 生成随机盐值
-		salt := rand.New(rand.NewSource(time.Now().UnixNano())).Intn(999999) + 100000 // 生成6位随机数
+		saltBig, err := rand.Int(rand.Reader, big.NewInt(900000))
+		if err != nil {
+			return fmt.Errorf("生成随机盐值失败: %v", err)
+		}
+		salt := int(saltBig.Int64()) + 100000 // 生成6位随机数
 		res, err := client.ExecuteEncrypted(method, encryptionKey[0], salt, params)
 		if err != nil {
 			return err

scheduler.go

@@ -29,6 +29,7 @@ const (
 	TaskExecutionFailedMsg = "Task execution failed"
 	WorkerTimeoutMsg       = "Worker timeout"
 	WorkerDisconnectedMsg  = "Worker disconnected"
+	ServiceUnavailableMsg  = "服务不可用"
 )
 
 //go:embed index.html
@@ -189,14 +190,38 @@ func (s *Scheduler) cleanupWorkerTasks(worker *Worker, errorMsg string) {
 	s.mu.Unlock()
 }
 
+// selectWorker 选择可用的Worker来执行指定方法
+func (s *Scheduler) selectWorker(method string) *Worker {
+	var selectedWorker *Worker
+	minCount := math.MaxInt
+
+	s.workers.Range(func(key, value interface{}) bool {
+		worker := value.(*Worker)
+		lastPingNano := atomic.LoadInt64(&worker.LastPing)
+		lastPing := time.Unix(0, lastPingNano)
+		if time.Since(lastPing) > 60*time.Second {
+			return true
+		}
+
+		for _, workerMethod := range worker.Methods {
+			if workerMethod.Name == method {
+				workerCount := atomic.LoadInt64(&worker.Count)
+				if workerCount < int64(minCount) {
+					selectedWorker = worker
+					minCount = int(workerCount)
+				}
+				break
+			}
+		}
+		return true
+	})
+
+	return selectedWorker
+}
+
 func (s *Scheduler) handleWorkerMessages(worker *Worker, conn *websocket.Conn) {
 	for {
-		var msg struct {
-			Type   string          `json:"type"`
-			TaskID string          `json:"taskId"`
-			Result json.RawMessage `json:"result"`
-			Error  string          `json:"error"`
-		}
+		var msg TaskResultMessage
 		if err := conn.ReadJSON(&msg); err != nil {
 			s.cleanupWorkerTasks(worker, WorkerDisconnectedMsg)
 			log.Printf("Worker %s disconnected: %v", worker.ID, err)
@@ -207,17 +232,12 @@ func (s *Scheduler) handleWorkerMessages(worker *Worker, conn *websocket.Conn) {
 		case "pong":
 			atomic.StoreInt64(&worker.LastPing, time.Now().UnixNano())
 		case "result":
-			s.processTaskResult(msg)
+			s.processTaskResult(&msg)
 		}
 	}
 }
 
-func (s *Scheduler) processTaskResult(msg struct {
-	Type   string          `json:"type"`
-	TaskID string          `json:"taskId"`
-	Result json.RawMessage `json:"result"`
-	Error  string          `json:"error"`
-}) {
+func (s *Scheduler) processTaskResult(msg *TaskResultMessage) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
@@ -273,33 +293,11 @@ func (s *Scheduler) handleEncryptedExecute(w http.ResponseWriter, r *http.Reques
 	s.encryptedTasks[encryptedTask.ID] = encryptedTask
 
 	// 选择可用的Worker
-	var selectedWorker *Worker
-	minCount := math.MaxInt
-
-	s.workers.Range(func(key, value interface{}) bool {
-		worker := value.(*Worker)
-		lastPingNano := atomic.LoadInt64(&worker.LastPing)
-		lastPing := time.Unix(0, lastPingNano)
-		if time.Since(lastPing) > 60*time.Second {
-			return true
-		}
-
-		for _, method := range worker.Methods {
-			if method.Name == req.Method {
-				workerCount := atomic.LoadInt64(&worker.Count)
-				if workerCount < int64(minCount) {
-					selectedWorker = worker
-					minCount = int(workerCount)
-				}
-				break
-			}
-		}
-		return true
-	})
+	selectedWorker := s.selectWorker(req.Method)
 
 	if selectedWorker == nil {
 		encryptedTask.Status = TaskStatusError
-		encryptedTask.Result = "服务不可用"
+		encryptedTask.Result = ServiceUnavailableMsg
 		s.mu.Unlock()
 	} else {
 		atomic.AddInt64(&selectedWorker.Count, 1)
@@ -363,36 +361,11 @@ func (s *Scheduler) handleExecute(w http.ResponseWriter, r *http.Request) {
 	s.tasks.Store(task.ID, task)
 
 	// 在同一个锁内完成worker选择和状态更新，避免竞态条件
-	var selectedWorker *Worker
-	minCount := math.MaxInt
-
-	// 查找可用Worker并验证其连接状态
-	s.workers.Range(func(key, value interface{}) bool {
-		worker := value.(*Worker)
-		// 使用原子操作检查worker连接是否仍然有效
-		lastPingNano := atomic.LoadInt64(&worker.LastPing)
-		lastPing := time.Unix(0, lastPingNano)
-		if time.Since(lastPing) > 60*time.Second {
-			return true // 跳过可能已断线的worker
-		}
-
-		for _, method := range worker.Methods {
-			if method.Name == req.Method {
-				// 使用原子操作读取计数
-				workerCount := atomic.LoadInt64(&worker.Count)
-				if workerCount < int64(minCount) {
-					selectedWorker = worker
-					minCount = int(workerCount)
-				}
-				break // 找到匹配方法就跳出内层循环
-			}
-		}
-		return true
-	})
+	selectedWorker := s.selectWorker(req.Method)
 
 	if selectedWorker == nil {
 		task.Status = TaskStatusError
-		task.Result = "服务不可用"
+		task.Result = ServiceUnavailableMsg
 		s.mu.Unlock()
 	} else {
 		// 使用原子操作增加计数

types.go

@@ -17,27 +17,27 @@ type Task struct {
 	Method  string
 	Params  json.RawMessage
 	Result  any
-	Status  TaskStatus // "pending", "processing", "done", "error"
 	Worker  *Worker
 	Created time.Time
+	Status  TaskStatus // "pending", "processing", "done", "error"
 }
 
 type Worker struct {
-	ID       string
 	Conn     *websocket.Conn
 	Methods  []MethodInfo
+	ConnMu   sync.Mutex // 保护websocket连接的并发写入
 	LastPing int64      // 使用原子操作，存储Unix纳秒时间戳
 	Count    int64      // 使用原子操作
-	ConnMu   sync.Mutex // 保护websocket连接的并发写入
+	ID       string
 	IP       string     // Worker的IP地址
 	Group    string     // Worker的分组
 }
 
 type WorkerInfo struct {
-	ID       string       `json:"id"`
 	Methods  []MethodInfo `json:"methods"`
 	LastPing time.Time    `json:"lastPing"`
 	Count    int64        `json:"count"`
+	ID       string       `json:"id"`
 	IP       string       `json:"ip"`
 	Group    string       `json:"group"`
 }
@@ -57,13 +57,21 @@ type EncryptedRequest struct {
 
 // EncryptedTask 端到端加密任务结构(调度器只做中转)
 type EncryptedTask struct {
-	ID      string
-	Key     string // 用户加盐后的密钥(原封不动传递)
-	Method  string // 方法名(明文，用于路由)
-	Params  string // 加密后的参数(原封不动传递)
-	Crypto  string // 加盐数据(原封不动传递)
-	Result  any
-	Status  TaskStatus
-	Worker  *Worker
-	Created time.Time
+	Result  interface{} `json:"result"`
+	Worker  *Worker     `json:"-"`
+	Created time.Time   `json:"created"`
+	ID      string      `json:"id"`
+	Key     string      `json:"key"`
+	Method  string      `json:"method"`
+	Params  string      `json:"params"`
+	Crypto  string      `json:"crypto"`
+	Status  TaskStatus  `json:"status"`
+}
+
+// TaskResultMessage represents the message structure for task results
+type TaskResultMessage struct {
+	Type   string          `json:"type"`
+	TaskID string          `json:"taskId"`
+	Error  string          `json:"error"`
+	Result json.RawMessage `json:"result"`
 }