项目结构更新

2025-09-15 11:09:22 +08:00
parent 2ab7614ea0
commit 8ccf028ae4
25 changed files with 608 additions and 17 deletions
--- a/scripts/dtmScripts/dtm.go
+++ b/scripts/dtmScripts/dtm.go
@@ -0,0 +1,177 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"math/rand"
+	"net/http"
+	"time"
+
+	"github.com/dtm-labs/client/dtmcli"
+	"github.com/gin-gonic/gin"
+	"go.etcd.io/etcd/client/v3"
+)
+
+// ServiceInstance 服务实例结构
+type ServiceInstance struct {
+	IP        string            `json:"ip"`
+	Port      int               `json:"port"`
+	Service   string            `json:"service"`
+	Timestamp string            `json:"timestamp"`
+	Status    string            `json:"status"`
+	Metadata  map[string]string `json:"metadata"`
+}
+
+// EmailRequest 邮件发送请求
+type EmailRequest struct {
+	To      string `json:"to"`
+	Subject string `json:"subject"`
+	Content string `json:"content"`
+}
+
+// LoginRequest 用户登录请求
+type LoginRequest struct {
+	UserID   string `json:"user_id"`
+	Password string `json:"password"` // 实际项目中应传递加密后的凭证
+}
+
+// 初始化etcd客户端
+func initEtcdClient(endpoints []string) (*clientv3.Client, error) {
+	return clientv3.New(clientv3.Config{
+		Endpoints:   endpoints,
+		DialTimeout: 5 * time.Second,
+	})
+}
+
+// 从etcd发现指定服务的实例
+func discoverService(ctx context.Context, etcdClient *clientv3.Client, serviceName string) ([]ServiceInstance, error) {
+	// 服务在etcd中的路径（与注册时保持一致）
+	path := fmt.Sprintf("/microservices/%s/instances/", serviceName)
+	resp, err := etcdClient.Get(ctx, path, clientv3.WithPrefix())
+	if err != nil {
+		return nil, fmt.Errorf("查询etcd失败: %v", err)
+	}
+
+	var instances []ServiceInstance
+	for _, kv := range resp.Kvs {
+		var instance ServiceInstance
+		if err := json.Unmarshal(kv.Value, &instance); err != nil {
+			log.Printf("解析服务实例失败，忽略此实例: %v", err)
+			continue
+		}
+		if instance.Status == "running" {
+			instances = append(instances, instance)
+		}
+	}
+
+	if len(instances) == 0 {
+		return nil, fmt.Errorf("未发现可用的%s服务实例", serviceName)
+	}
+	return instances, nil
+}
+
+// 从可用实例中选择一个（随机负载均衡）
+func selectServiceInstance(instances []ServiceInstance) (string, error) {
+	if len(instances) == 0 {
+		return "", fmt.Errorf("没有可用的服务实例")
+	}
+	rand.Seed(time.Now().UnixNano())
+	selected := instances[rand.Intn(len(instances))]
+	return fmt.Sprintf("http://%s:%d", selected.IP, selected.Port), nil
+}
+
+// 创建包含登录和邮件发送的分布式事务
+func createCompositeTransaction(c *gin.Context, etcdClient *clientv3.Client) {
+	// DTM服务器地址
+	dtmServer := "http://localhost:36789/api/dtmsvr"
+	// 本地业务服务地址
+	ctx := context.Background()
+
+	// 1. 发现并选择user-login服务实例
+	loginInstances, err := discoverService(ctx, etcdClient, "user-login")
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("发现user-login服务失败: %v", err)})
+		return
+	}
+	loginServiceAddr, err := selectServiceInstance(loginInstances)
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("选择login服务实例失败: %v", err)})
+		return
+	}
+	log.Printf("已选择user-login服务实例: %s", loginServiceAddr)
+
+	emailInstances, err := discoverService(ctx, etcdClient, "email")
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("发现email服务失败: %v", err)})
+		return
+	}
+	emailServiceAddr, err := selectServiceInstance(emailInstances)
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("选择email服务实例失败: %v", err)})
+		return
+	}
+	log.Printf("已选择email服务实例: %s", emailServiceAddr)
+
+	// 3. 创建SAGA分布式事务（包含两个步骤）
+	saga := dtmcli.NewSaga(dtmServer, dtmcli.MustGenGid(dtmServer)).
+		// 第一步：调用user-login服务进行登录验证
+		Add(
+			loginServiceAddr+"/login",            // 正向接口：用户登录
+			loginServiceAddr+"/login/compensate", // 补偿接口：登录状态回滚
+			LoginRequest{
+				UserID:   "user123",
+				Password: "encrypted_token_xxx", // 实际项目中使用加密凭证
+			},
+		).
+		// 第二步：调用email服务发送登录通知邮件（依赖第一步成功）
+		Add(
+			emailServiceAddr+"/send",            // 正向接口：发送邮件
+			emailServiceAddr+"/send/compensate", // 补偿接口：邮件发送回滚
+			EmailRequest{
+				To:      "user@example.com",
+				Subject: "登录成功通知",
+				Content: fmt.Sprintf("您的账号于%s成功登录系统", time.Now().Format("2006-01-02 15:04:05")),
+			},
+		)
+
+	// 4. 提交事务
+	if err := saga.Submit(); err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("事务提交失败: %v", err)})
+		return
+	}
+
+	c.JSON(http.StatusOK, gin.H{
+		"success":       true,
+		"message":       "分布式事务已提交",
+		"transactionId": saga.Gid,
+		"steps": []map[string]string{
+			{"service": "user-login", "address": loginServiceAddr},
+			{"service": "email", "address": emailServiceAddr},
+		},
+	})
+}
+
+func main() {
+	// 初始化etcd客户端
+	etcdClient, err := initEtcdClient([]string{"http://127.0.0.1:2379"})
+	if err != nil {
+		log.Fatalf("初始化etcd客户端失败: %v", err)
+	}
+	defer etcdClient.Close()
+
+	// 初始化Gin引擎
+	r := gin.Default()
+
+	// 注册复合事务接口
+	r.POST("/userAction", func(c *gin.Context) {
+		createCompositeTransaction(c, etcdClient)
+	})
+
+	// 启动业务服务
+	log.Println("业务服务启动，监听端口 8081")
+	if err := r.Run(":8081"); err != nil {
+		log.Fatalf("服务启动失败: %v", err)
+	}
+}
--- a/scripts/goroutine/connectPool/goroutine/readgo.go
+++ b/scripts/goroutine/connectPool/goroutine/readgo.go
@@ -0,0 +1,35 @@
+package goroutine
+
+import (
+	"database/sql"
+	"fmt"
+	"math/rand"
+	"time"
+)
+
+func Dataread(db *sql.DB, donetitle chan struct{}) {
+	// 启动一个长事务（整个循环在一个事务内，而非每次查询一个事务）
+	tx, err := db.Begin()
+	if err != nil {
+		panic(err)
+	}
+	defer tx.Rollback() // 确保退出时回滚（仅为测试）
+
+	// 随机种子（保持不变）
+	rand.Seed(time.Now().UnixNano())
+
+	// 只执行两次查询，方便观察结果
+	for i := 0; i < 2; i++ {
+		select {
+		case <-time.After(2 * time.Second): // 第一次查询后等待2秒，给Datawrite插入时间
+			var count int
+			err = tx.QueryRow("SELECT count(*) FROM article WHERE age > 50  FOR UPDATE ").Scan(&count)
+			if err != nil {
+				panic(err)
+			}
+			fmt.Printf("第%d次查询 count: %d\n", i+1, count)
+		case <-donetitle:
+			return
+		}
+	}
+}
--- a/scripts/goroutine/connectPool/goroutine/updatego.go
+++ b/scripts/goroutine/connectPool/goroutine/updatego.go
@@ -0,0 +1,11 @@
+package goroutine
+
+import "database/sql"
+
+func DataUpdate(db *sql.DB) {
+	_, err := db.Begin()
+	if err != nil {
+		panic(err)
+	}
+
+}
--- a/scripts/goroutine/connectPool/goroutine/writego.go
+++ b/scripts/goroutine/connectPool/goroutine/writego.go
@@ -0,0 +1,45 @@
+package goroutine
+
+import (
+	"database/sql"
+	"fmt"
+	"github.com/google/uuid"
+	"math/rand"
+	"strconv"
+	"time"
+)
+
+func Datawrite(db *sql.DB, donetitle chan struct{}) {
+	timetick := time.Tick(1 * time.Second)
+	rand.Seed(time.Now().UnixNano())
+
+	for {
+		select {
+		case <-timetick:
+			uid, error := uuid.NewRandom()
+			if error != nil {
+				panic(error)
+			}
+			tx, err := db.Begin()
+			if err != nil {
+				panic(err)
+			}
+			fmt.Println("开始事务")
+			_, err = tx.Exec(`insert INTO article (uid,content,author,age) values (?,?,?,?)`, uid, "123", "用户"+strconv.Itoa(rand.Intn(1000)), rand.Intn(40)+60)
+			if err != nil {
+				if rbErr := tx.Rollback(); rbErr != nil {
+					panic("回滚失败: " + rbErr.Error())
+				}
+				panic("插入失败: " + err.Error()) // 明确错误类型
+			}
+			err = tx.Commit()
+			if err != nil {
+				panic(err)
+			}
+			fmt.Println("插入数据成功")
+		case <-donetitle:
+			return
+		}
+	}
+
+}
--- a/scripts/goroutine/connectPool/pool.go
+++ b/scripts/goroutine/connectPool/pool.go
@@ -0,0 +1,32 @@
+package connectPool
+
+import (
+	"database/sql"
+	"fmt"
+	_ "github.com/go-sql-driver/mysql"
+	goroutine2 "toutoukan/scripts/goroutine/connectPool/goroutine"
+)
+
+var db *sql.DB
+
+func ConnectPool(donetitle chan struct{}) {
+	dsn := "mayiming:Mydream5654my,@tcp(43.142.81.151:3306)/goLearn?charset-uft8mb4&parseTime=True"
+	db, _ = sql.Open("mysql", dsn)
+
+	err := db.Ping()
+	if err != nil {
+		panic(err)
+	}
+	fmt.Println("数据库连接成功")
+	for i := 0; i < 20; i++ {
+		go goroutine2.Datawrite(db, donetitle)
+	}
+	go goroutine2.Dataread(db, donetitle)
+}
+
+func DisconnectPool() {
+	err := db.Close()
+	if err != nil {
+		panic(err)
+	}
+}
--- a/scripts/goroutine/consumer.go
+++ b/scripts/goroutine/consumer.go
@@ -0,0 +1,27 @@
+package goroutine
+
+import (
+	"fmt"
+	"sync"
+	"toutoukan/model/routine"
+)
+
+func NewConsumer(ch chan routine.Task, wg *sync.WaitGroup, done chan struct{}, num int, data *routine.Data, rwmutex *sync.RWMutex) {
+	defer wg.Done()
+	for {
+		select {
+		case <-ch:
+			rwmutex.RLock()
+			fmt.Printf("消费者%d号处理消息，统计消息总数为：%d\n", num+1, data.Count)
+			fmt.Printf("消费者%d号：当前记录详情如下：\n", num+1)
+			for key, value := range data.Record {
+				fmt.Printf("  键：%d，值：%d\n", key, value)
+			}
+			rwmutex.RUnlock()
+		case <-done:
+			fmt.Printf("消费者%d号退出接收\n", num+1)
+			return
+		}
+	}
+
+}
--- a/scripts/goroutine/producer.go
+++ b/scripts/goroutine/producer.go
@@ -0,0 +1,44 @@
+package goroutine
+
+import (
+	"fmt"
+	"math/rand"
+	"strconv"
+	"sync"
+	"time"
+	"toutoukan/model/routine"
+)
+
+func NewProducer(ch chan routine.Task, wg *sync.WaitGroup, done chan struct{}, num int, mutex *sync.Mutex, data *routine.Data, rwmutex *sync.RWMutex) {
+	defer wg.Done()
+	timech := time.Tick(1 * time.Second)
+	rand.Seed(time.Now().UnixNano())
+	for {
+		select {
+		case <-timech:
+			fmt.Printf("生产者%d号上锁\n", num+1)
+			randnum := rand.Intn(100)
+			task := routine.Task{
+				Id:      strconv.Itoa(randnum),
+				Content: "访问数据库",
+			}
+			mutex.Lock()
+			data.Count += 1
+			mutex.Unlock()
+
+			rwmutex.Lock()
+			data.Record[randnum] += 1
+			rwmutex.Unlock()
+
+			fmt.Printf("生产者%d号注入编号为%s,内容为:%s\n", num+1, task.Id, task.Content)
+			ch <- task
+			fmt.Printf("生产者%d号释放锁\n", num+1)
+
+		case <-done:
+			fmt.Printf("生产者%d号退出任务\n", num+1)
+			return
+		}
+
+	}
+
+}
--- a/scripts/oneKUsers/mian.go
+++ b/scripts/oneKUsers/mian.go
@@ -0,0 +1,122 @@
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"sync"
+	"time"
+	"toutoukan/model/usermodel/userOrder"
+)
+
+// 请求体结构（Order字段对应目标商品标识）
+
+// 响应体结构（根据实际接口响应调整）
+type Response struct {
+	Success bool   `json:"success"`
+	Message string `json:"message"`
+	// 可根据实际接口返回补充字段，如订单号、库存状态等
+}
+
+func main() {
+	const totalBatches = 1           // 固定发送5批请求
+	const batchSize = 150            // 每批固定150个请求
+	const interval = 5 * time.Second // 每批请求的时间间隔（可按需调整）
+	var wg sync.WaitGroup
+
+	userID := 1 // 用户ID计数器，从1开始递增（确保每个请求用户ID唯一）
+
+	// 循环发送5批请求，每批对应不同的商品order（stock10000 ~ stock10004）
+	for batch := 1; batch <= totalBatches; batch++ {
+		// 计算当前批次的目标商品order：第1批→stock10000，第2批→stock10001...第5批→stock10004
+		targetOrder := fmt.Sprintf("stock:1000%d", batch-1)
+		fmt.Printf("=== 开始第 %d 批请求 ===\n", batch)
+		fmt.Printf("当前批次目标商品: %s | 请求数量: %d 个 | 开始时间: %v\n",
+			targetOrder, batchSize, time.Now().Format("15:04:05"))
+
+		// 并发发送当前批次的150个请求
+		for i := 0; i < batchSize; i++ {
+			wg.Add(1)
+			// 捕获当前循环的uid和order（避免goroutine闭包引用循环变量问题）
+			currentUID := userID
+			currentOrder := targetOrder
+			go func(uid int, order string) {
+				defer wg.Done()
+				// 发送单个POST请求
+				result, err := sendPostRequest(uid, order)
+				if err != nil {
+					// 请求失败：输出错误日志（包含用户ID和商品标识）
+					fmt.Printf("❌ 用户 %d (商品%s) 请求失败: %v\n", uid, order, err)
+					return
+				}
+				// 请求成功：输出响应结果（可按需调整日志粒度，避免刷屏）
+				fmt.Printf("✅ 用户 %d (商品%s) 响应: 成功=%v, 消息=%s\n",
+					uid, order, result.Success, result.Message)
+			}(currentUID, currentOrder)
+			userID++ // 每生成一个请求，用户ID递增1
+		}
+
+		// 等待当前批次所有请求完成（确保批次内请求全部处理后，再进入下一批）
+		wg.Wait()
+		fmt.Printf("=== 第 %d 批请求全部完成 | 完成时间: %v ===\n\n",
+			batch, time.Now().Format("15:04:05"))
+
+		// 非最后一批请求，等待指定间隔后再发送下一批（避免请求集中压测）
+		if batch < totalBatches {
+			fmt.Printf("等待 %v 后发送下一批请求...\n\n", interval)
+			time.Sleep(interval)
+		}
+	}
+
+	fmt.Println("🎉 所有5批请求已全部完成！")
+}
+
+// 发送POST请求（接收用户ID和商品order，构造请求体）
+func sendPostRequest(userID int, order string) (*Response, error) {
+	// 目标接口地址（根据实际部署调整）
+	targetURL := "http://localhost:9096/user/kill"
+	// 1. 构造请求体（绑定当前用户ID和目标商品order）
+	requestBody := userOrder.UserRequest{
+		UserID: userID,
+		Order:  order,
+	}
+
+	// 2. 将请求体序列化为JSON格式
+	jsonBody, err := json.Marshal(requestBody)
+	if err != nil {
+		return nil, fmt.Errorf("JSON序列化失败: %v", err)
+	}
+
+	// 3. 发送POST请求（设置Content-Type为application/json）
+	resp, err := http.Post(
+		targetURL,
+		"application/json",
+		bytes.NewBuffer(jsonBody),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("请求发送失败（网络/连接问题）: %v", err)
+	}
+	defer resp.Body.Close() // 确保响应体关闭，避免资源泄漏
+
+	// 4. 解析接口响应（无论状态码是否为200，均尝试解析，便于排查问题）
+	var response Response
+	decoder := json.NewDecoder(resp.Body)
+	if err := decoder.Decode(&response); err != nil {
+		return nil, fmt.Errorf("响应解析失败（格式不匹配）: %v", err)
+	}
+
+	// 5. 检查HTTP状态码（非200视为异常，返回详细信息）
+	if resp.StatusCode != http.StatusOK {
+		// 将响应内容转为格式化JSON，便于查看完整错误信息
+		responseDetail, _ := json.MarshalIndent(response, "", "  ")
+		return &response, fmt.Errorf(
+			"HTTP状态码异常: %d | 接口响应详情: %s",
+			resp.StatusCode,
+			responseDetail,
+		)
+	}
+
+	// 6. 请求成功，返回解析后的响应
+	return &response, nil
+}