8.5 服务间通信模式（¶

// 服务端：user-service（提供用户查询接口）
package main

import (
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "strconv"
    "time"
)

// User 资源定义
type User struct {
    ID   int    `json:"id"`
    Name string `json:"name"`
    Age  int    `json:"age"`
}

// 处理用户查询请求（GET /api/user?id=1）
func getUserHandler(w http.ResponseWriter, r *http.Request) {
    // 1. 校验请求方法
    if r.Method != http.MethodGet {
        w.WriteHeader(http.StatusMethodNotAllowed)
        fmt.Fprintf(w, "只支持GET请求")
        return
    }

    // 2. 获取请求参数
    idStr := r.URL.Query().Get("id")
    if idStr == "" {
        w.WriteHeader(http.StatusBadRequest)
        fmt.Fprintf(w, "缺少id参数")
        return
    }
    id, err := strconv.Atoi(idStr)
    if err != nil {
        w.WriteHeader(http.StatusBadRequest)
        fmt.Fprintf(w, "id必须是数字")
        return
    }

    // 3. 模拟数据库查询（实际项目中替换为SQL查询）
    user := User{ID: id, Name: "张三", Age: 28}

    // 4. 设置响应头（JSON格式）
    w.Header().Set("Content-Type", "application/json")
    w.WriteHeader(http.StatusOK)

    // 5. 序列化JSON并返回
    if err := json.NewEncoder(w).Encode(user); err != nil {
        log.Printf("序列化用户失败：%v", err)
        w.WriteHeader(http.StatusInternalServerError)
        return
    }
}

func main() {
    // 注册路由
    http.HandleFunc("/api/user", getUserHandler)

    // 启动服务（带超时控制）
    server := &http.Server{
        Addr:         ":8080",
        ReadTimeout:  5 * time.Second,  // 读超时（避免客户端一直发数据）
        WriteTimeout: 10 * time.Second, // 写超时（避免服务端处理过久）
    }

    log.Println("HTTP服务启动：http://localhost:8080")
    if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
        log.Fatalf("服务启动失败：%v", err)
    }
}

// 客户端：order-service（调用user-service查询用户）
package main

import (
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "time"
)

type User struct {
    ID   int    `json:"id"`
    Name string `json:"name"`
    Age  int    `json:"age"`
}

func main() {
    // 1. 构建请求（带超时客户端）
    client := &http.Client{
        Timeout: 3 * time.Second, // 客户端超时（避免服务端无响应导致阻塞）
    }
    req, err := http.NewRequest(http.MethodGet, "http://localhost:8080/api/user?id=1", nil)
    if err != nil {
        log.Fatalf("构建请求失败：%v", err)
    }

    // 2. 发送请求
    resp, err := client.Do(req)
    if err != nil {
        log.Fatalf("请求失败：%v", err)
    }
    defer resp.Body.Close() // 必须关闭Body，避免资源泄漏

    // 3. 校验响应状态码
    if resp.StatusCode != http.StatusOK {
        log.Fatalf("响应状态码异常：%d", resp.StatusCode)
    }

    // 4. 解析JSON响应
    var user User
    if err := json.NewDecoder(resp.Body).Decode(&user); err != nil {
        log.Fatalf("解析响应失败：%v", err)
    }

    // 5. 处理结果
    fmt.Printf("查询到用户：ID=%d, Name=%s, Age=%d\n", user.ID, user.Name, user.Age)
}

// server/main.go
package main

import (
    "context"
    "log"
    "net"
    "time"

    "google.golang.org/grpc"
    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/status"

    "your-project-path/userpb" // 替换为你的实际路径
)

// userServer 实现UserService接口
type userServer struct {
    userpb.UnimplementedUserServiceServer // 必须嵌入（兼容proto更新）
}

// GetUser 实现Unary RPC接口
func (s *userServer) GetUser(ctx context.Context, req *userpb.GetUserRequest) (*userpb.GetUserResponse, error) {
    // 1. 校验参数
    if req.Id <= 0 {
        return nil, status.Errorf(codes.InvalidArgument, "id必须大于0")
    }

    // 2. 模拟数据库查询（带超时控制，响应ctx的取消信号）
    select {
    case <-ctx.Done():
        log.Printf("请求被取消：%v", ctx.Err())
        return nil, status.Errorf(codes.DeadlineExceeded, "请求超时")
    case <-time.After(100 * time.Millisecond): // 模拟处理耗时
        // 3. 构造响应
        user := &userpb.User{
            Id:   req.Id,
            Name: "张三",
            Age:  28,
        }
        return &userpb.GetUserResponse{User: user}, nil
    }
}

// ListUsers 实现服务端流式RPC接口
func (s *userServer) ListUsers(req *userpb.ListUsersRequest, stream userpb.UserService_ListUsersServer) error {
    // 1. 校验参数
    if req.Page <= 0 {
        req.Page = 1
    }
    if req.PageSize <= 0 || req.PageSize > 100 {
        req.PageSize = 10
    }

    // 2. 模拟分页查询（流式返回3条数据）
    start := (req.Page - 1) * req.PageSize
    for i := 0; i < int(req.PageSize); i++ {
        user := &userpb.User{
            Id:   start + int32(i+1),
            Name: fmt.Sprintf("用户%d", start+i+1),
            Age:  20 + int32(i%10),
        }

        // 3. 流式发送响应（需检查ctx是否取消）
        if err := stream.Send(&userpb.ListUsersResponse{User: user}); err != nil {
            return err
        }
        time.Sleep(200 * time.Millisecond) // 模拟分批返回
    }

    return nil
}

func main() {
    // 1. 创建gRPC服务器（可添加拦截器，如日志、认证）
    s := grpc.NewServer(
        grpc.MaxRecvMsgSize(1024*1024), // 最大接收消息大小（1MB）
        grpc.MaxSendMsgSize(1024*1024), // 最大发送消息大小（1MB）
    )

    // 2. 注册服务
    userpb.RegisterUserServiceServer(s, &userServer{})

    // 3. 监听端口
    lis, err := net.Listen("tcp", ":50051")
    if err != nil {
        log.Fatalf("监听失败：%v", err)
    }
    defer lis.Close()

    log.Println("gRPC服务启动：localhost:50051")
    if err := s.Serve(lis); err != nil {
        log.Fatalf("服务启动失败：%v", err)
    }
}

// client/main.go
package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure" // 开发环境用（无TLS）

    "your-project-path/userpb" // 替换为你的实际路径
)

func main() {
    // 1. 连接gRPC服务（开发环境禁用TLS，生产环境需配置TLS）
    conn, err := grpc.Dial(
        "localhost:50051",
        grpc.WithTransportCredentials(insecure.NewCredentials()), // 无TLS
        grpc.WithBlock(), // 等待连接成功
        grpc.WithTimeout(3*time.Second), // 连接超时
    )
    if err != nil {
        log.Fatalf("连接失败：%v", err)
    }
    defer conn.Close() // 退出前关闭连接

    // 2. 创建客户端
    client := userpb.NewUserServiceClient(conn)

    // 3. 调用Unary RPC（GetUser）
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel() // 确保ctx被取消，释放资源

    getResp, err := client.GetUser(ctx, &userpb.GetUserRequest{Id: 1})
    if err != nil {
        // 解析gRPC错误码
        if stat, ok := status.FromError(err); ok {
            log.Fatalf("调用GetUser失败：%s（错误码：%d）", stat.Message(), stat.Code())
        }
        log.Fatalf("调用GetUser失败：%v", err)
    }
    fmt.Printf("Unary RPC结果：ID=%d, Name=%s, Age=%d\n", 
        getResp.User.Id, getResp.User.Name, getResp.User.Age)

    // 4. 调用服务端流式RPC（ListUsers）
    stream, err := client.ListUsers(ctx, &userpb.ListUsersRequest{Page: 1, PageSize: 3})
    if err != nil {
        log.Fatalf("调用ListUsers失败：%v", err)
    }

    fmt.Println("\n服务端流式RPC结果：")
    for {
        // 循环接收流式响应
        streamResp, err := stream.Recv()
        if err == io.EOF { // 流结束
            break
        }
        if err != nil {
            log.Fatalf("接收流数据失败：%v", err)
        }
        fmt.Printf("ID=%d, Name=%s, Age=%d\n", 
            streamResp.User.Id, streamResp.User.Name, streamResp.User.Age)
    }
}

// producer/main.go
package main

import (
    "log"
    "time"

    "github.com/streadway/amqp"
)

// 处理RabbitMQ错误
func failOnError(err error, msg string) {
    if err != nil {
        log.Fatalf("%s: %s", msg, err)
    }
}

func main() {
    // 1. 连接RabbitMQ
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    failOnError(err, "连接RabbitMQ失败")
    defer conn.Close()

    // 2. 创建通道（RabbitMQ推荐用通道而非直接用连接）
    ch, err := conn.Channel()
    failOnError(err, "创建通道失败")
    defer ch.Close()

    // 3. 声明交换机（Topic模式，持久化）
    exchangeName := "order.exchange"
    err = ch.ExchangeDeclare(
        exchangeName, // 交换机名
        "topic",      // 类型（topic支持通配符路由）
        true,         // 是否持久化（交换机重启后不丢失）
        false,        // 是否自动删除（无队列绑定时删除）
        false,        // 是否内部交换机（不允许外部生产者发送）
        false,        // 是否等待响应
        nil,          // 额外参数
    )
    failOnError(err, "声明交换机失败")

    // 4. 构造消息（订单数据）
    orderID := "ORDER123456"
    messageBody := "{\"order_id\":\"" + orderID + "\",\"user_id\":1001,\"amount\":99.9}"
    msg := amqp.Publishing{
        DeliveryMode: amqp.Persistent, // 消息持久化（重启后不丢失）
        ContentType:  "application/json",
        Body:         []byte(messageBody),
        Timestamp:    time.Now(),
    }

    // 5. 发送消息（指定路由键，用于消费者过滤）
    routingKey := "order.created" // 路由键：订单创建事件
    err = ch.Publish(
        exchangeName, // 交换机名
        routingKey,   // 路由键
        false,        // 是否强制发送（无队列绑定时返回错误）
        false,        // 是否立即发送（无消费者时返回错误）
        msg,          // 消息内容
    )
    failOnError(err, "发送消息失败")

    log.Printf("已发送消息：order_id=%s, routing_key=%s", orderID, routingKey)
}

// consumer/main.go
package main

import (
    "log"
    "time"

    "github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
    if err != nil {
        log.Fatalf("%s: %s", msg, err)
    }
}

func main() {
    // 1. 连接RabbitMQ
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    failOnError(err, "连接RabbitMQ失败")
    defer conn.Close()

    // 2. 创建通道
    ch, err := conn.Channel()
    failOnError(err, "创建通道失败")
    defer ch.Close()

    // 3. 声明交换机（与生产者一致）
    exchangeName := "order.exchange"
    err = ch.ExchangeDeclare(
        exchangeName,
        "topic",
        true,
        false,
        false,
        false,
        nil,
    )
    failOnError(err, "声明交换机失败")

    // 4. 声明队列（持久化，用于存储消息）
    queueName := "inventory.queue"
    q, err := ch.QueueDeclare(
        queueName, // 队列名
        true,      // 持久化
        false,     // 自动删除
        false,     // 独占队列（仅当前连接可用）
        false,     // 等待响应
        nil,
    )
    failOnError(err, "声明队列失败")

    // 5. 绑定队列到交换机（指定路由键，只接收order.created事件）
    routingKey := "order.created"
    err = ch.QueueBind(
        q.Name,       // 队列名
        routingKey,   // 路由键（匹配生产者的发送键）
        exchangeName, // 交换机名
        false,
        nil,
    )
    failOnError(err, "绑定队列失败")

    // 6. 消费消息（关闭自动ACK，手动确认）
    msgs, err := ch.Consume(
        q.Name,  // 队列名
        "",      // 消费者标签（用于标识消费者）
        false,   // 是否自动ACK（false=手动确认，确保处理完再删除）
        false,   // 是否独占消费
        false,   // 是否禁止本地消费者接收自己发送的消息
        false,   // 是否等待响应
        nil,
    )
    failOnError(err, "注册消费者失败")

    // 7. 处理消息（用通道接收消息，避免阻塞）
    forever := make(chan bool)
    go func() {
        for d := range msgs {
            log.Printf("收到消息：order_id=%s, body=%s", 
                extractOrderID(string(d.Body)), string(d.Body))

            // 模拟业务处理（如扣减库存）
            time.Sleep(1 * time.Second)

            // 手动ACK：处理完消息后，通知MQ删除消息
            err := d.Ack(false) // false=只确认当前消息，true=确认所有未确认消息
            if err != nil {
                log.Printf("ACK失败：%v", err)
                // 处理失败：可将消息发送到死信队列，避免重复消费
                return
            }
            log.Printf("消息处理完成：order_id=%s", extractOrderID(string(d.Body)))
        }
    }()

    log.Printf("消费者启动：等待接收消息（路由键：%s）", routingKey)
    <-forever // 阻塞主进程
}

// 提取订单ID（简化实现，实际项目用JSON解析）
func extractOrderID(body string) string {
    // 实际项目中用encoding/json解析body
    if len(body) > 0 {
        return body[14:20] // 截取ORDER123456中的123456（仅示例）
    }
    return "unknown"
}

// 基于Redis实现幂等性（订单ID作为唯一键）
func processOrder(ctx context.Context, redisClient *redis.Client, orderID string) error {
    // 1. 尝试设置Redis键（NX=不存在才设置，EX=过期时间）
    key := "order:processed:" + orderID
    ok, err := redisClient.SetNX(ctx, key, "1", 24*time.Hour).Result()
    if err != nil {
        return err
    }

    // 2. 键已存在→消息已处理，直接返回
    if !ok {
        log.Printf("订单已处理：order_id=%s", orderID)
        return nil
    }

    // 3. 键不存在→处理消息（扣减库存）
    log.Printf("处理订单：order_id=%s", orderID)
    // 此处添加库存扣减逻辑...

    return nil
}

creds, err := credentials.NewServerTLSFromFile("server.crt", "server.key")
if err != nil {
    log.Fatalf("加载TLS证书失败：%v", err)
}
s := grpc.NewServer(grpc.Creds(creds))

creds, err := credentials.NewClientTLSFromFile("server.crt", "localhost")
if err != nil {
    log.Fatalf("加载TLS证书失败：%v", err)
}
conn, err := grpc.Dial("localhost:50051", grpc.WithTransportCredentials(creds))

// 安装监控依赖
go get github.com/grpc-ecosystem/go-grpc-prometheus

// 服务端添加监控拦截器
import "github.com/grpc-ecosystem/go-grpc-prometheus"

func main() {
    s := grpc.NewServer(
        grpc.UnaryInterceptor(grpc_prometheus.UnaryServerInterceptor),
        grpc.StreamInterceptor(grpc_prometheus.StreamServerInterceptor),
    )
    // 注册监控指标
    grpc_prometheus.Register(s)
    // 启动HTTP监控服务（暴露/metrics端点）
    go func() {
        http.Handle("/metrics", promhttp.Handler())
        log.Fatal(http.ListenAndServe(":9090", nil))
    }()
    // ... 其他代码
}