二、RPC 协议设计与实现

package mrpc

import (
	"encoding/binary"
	"encoding/json"
	"fmt"
	"io"
)

// ============================================================
// 第一部分：协议数据结构定义
// ============================================================

// RequestHeader 是请求的协议头
// 包含描述协议本身的数据和描述这次请求的数据
type RequestHeader struct {
	// --- 描述协议本身的数据 ---
	Version      uint8  // 协议版本号：用于后续协议升级，当前版本为 1
	Serializer   uint8  // 序列化协议标记：1=JSON, 2=Protobuf, 3=Gob
	Compressor   uint8  // 压缩算法标记：0=不压缩, 1=gzip, 2=snappy
	MessageID    uint64 // 消息ID：唯一标识一个请求，用于多路复用

	// --- 描述这次请求的数据 ---
	Service      string            // 服务名：服务端据此查找目标服务
	Method       string            // 方法名：服务端据此查找目标方法
	Meta         map[string]string // 元数据：链路追踪ID、A/B测试标记、超时时间等
}

// Request 是完整的 RPC 请求
// 由协议头和协议体组成
type Request struct {
	Header  RequestHeader // 协议头：包含路由和元数据信息
	Body    []byte        // 协议体：序列化后的请求参数（已经是字节流）
}

// ResponseHeader 是响应的协议头
type ResponseHeader struct {
	// --- 描述协议本身的数据 ---
	Version    uint8  // 协议版本号
	Serializer uint8  // 序列化协议标记
	Compressor uint8  // 压缩算法标记
	MessageID  uint64 // 消息ID：用于匹配请求和响应

	// --- 描述这次响应的数据 ---
	Error string // 错误信息：空字符串表示成功
}

// Response 是完整的 RPC 响应
type Response struct {
	Header ResponseHeader // 协议头
	Body   []byte         // 协议体：序列化后的响应数据
}

// ============================================================
// 第二部分：Request 编解码
// ============================================================

// EncodeRequest 将 Request 编码成字节流
// 编码格式：[头部长度(4字节)][头部JSON][体长度(4字节)][体数据]
// 这里用长度前缀来解决 TCP 粘包问题
func EncodeRequest(req *Request) ([]byte, error) {
	// 第一步：将 Header 序列化为 JSON
	// Header 是结构化数据，用 JSON 编码方便扩展
	headerData, err := json.Marshal(req.Header)
	if err != nil {
		return nil, fmt.Errorf("编码请求头失败: %w", err)
	}

	// 第二步：组装最终的字节流
	// 格式：[头部长度(4字节)] + [头部JSON] + [体长度(4字节)] + [体数据]
	// 使用 4 字节的 uint32 来表示长度，最大支持约 4GB 的消息
	result := make([]byte, 0, 4+len(headerData)+4+len(req.Body))

	// 写入头部长度（4字节，大端序）
	headerLenBuf := make([]byte, 4)
	binary.BigEndian.PutUint32(headerLenBuf, uint32(len(headerData)))
	result = append(result, headerLenBuf...)

	// 写入头部数据
	result = append(result, headerData...)

	// 写入体长度（4字节，大端序）
	bodyLenBuf := make([]byte, 4)
	binary.BigEndian.PutUint32(bodyLenBuf, uint32(len(req.Body)))
	result = append(result, bodyLenBuf...)

	// 写入体数据
	result = append(result, req.Body...)

	return result, nil
}

// DecodeRequest 从字节流中解码出 Request
// 解码过程与编码过程正好相反：先读头部长度 -> 读头部 -> 读体长度 -> 读体
func DecodeRequest(data []byte) (*Request, error) {
	if len(data) < 8 {
		return nil, fmt.Errorf("数据太短，无法解码")
	}

	// 第一步：读取头部长度
	headerLen := binary.BigEndian.Uint32(data[:4])
	if len(data) < int(4+headerLen+4) {
		return nil, fmt.Errorf("数据不完整，头部长度不匹配")
	}

	// 第二步：读取并反序列化头部
	headerData := data[4 : 4+headerLen]
	var header RequestHeader
	if err := json.Unmarshal(headerData, &header); err != nil {
		return nil, fmt.Errorf("解码请求头失败: %w", err)
	}

	// 第三步：读取体长度
	bodyLen := binary.BigEndian.Uint32(data[4+headerLen : 4+headerLen+4])
	if len(data) < int(4+headerLen+4+bodyLen) {
		return nil, fmt.Errorf("数据不完整，体长度不匹配")
	}

	// 第四步：读取体数据
	bodyData := data[4+headerLen+4 : 4+headerLen+4+bodyLen]

	// 组装 Request
	req := &Request{
		Header: header,
		Body:   make([]byte, bodyLen),
	}
	copy(req.Body, bodyData)

	return req, nil
}

// ============================================================
// 第三部分：Response 编解码
// ============================================================

// EncodeResponse 将 Response 编码成字节流
// 编码格式与 Request 一致：[头部长度(4字节)][头部JSON][体长度(4字节)][体数据]
func EncodeResponse(resp *Response) ([]byte, error) {
	// 将 Header 序列化为 JSON
	headerData, err := json.Marshal(resp.Header)
	if err != nil {
		return nil, fmt.Errorf("编码响应头失败: %w", err)
	}

	// 组装字节流
	result := make([]byte, 0, 4+len(headerData)+4+len(resp.Body))

	// 写入头部长度
	headerLenBuf := make([]byte, 4)
	binary.BigEndian.PutUint32(headerLenBuf, uint32(len(headerData)))
	result = append(result, headerLenBuf...)

	// 写入头部数据
	result = append(result, headerData...)

	// 写入体长度
	bodyLenBuf := make([]byte, 4)
	binary.BigEndian.PutUint32(bodyLenBuf, uint32(len(resp.Body)))
	result = append(result, bodyLenBuf...)

	// 写入体数据
	result = append(result, resp.Body...)

	return result, nil
}

// DecodeResponse 从字节流中解码出 Response
// 解码过程与编码过程相反
func DecodeResponse(data []byte) (*Response, error) {
	if len(data) < 8 {
		return nil, fmt.Errorf("数据太短，无法解码")
	}

	// 读取头部长度
	headerLen := binary.BigEndian.Uint32(data[:4])
	if len(data) < int(4+headerLen+4) {
		return nil, fmt.Errorf("数据不完整")
	}

	// 反序列化头部
	headerData := data[4 : 4+headerLen]
	var header ResponseHeader
	if err := json.Unmarshal(headerData, &header); err != nil {
		return nil, fmt.Errorf("解码响应头失败: %w", err)
	}

	// 读取体长度和体数据
	bodyLen := binary.BigEndian.Uint32(data[4+headerLen : 4+headerLen+4])
	bodyData := data[4+headerLen+4 : 4+headerLen+4+bodyLen]

	resp := &Response{
		Header: header,
		Body:   make([]byte, bodyLen),
	}
	copy(resp.Body, bodyData)

	return resp, nil
}

// ============================================================
// 第四部分：网络读写辅助函数
// ============================================================

// WriteRequest 将 Request 写入网络连接
// 先编码成字节流，再写入 conn
func WriteRequest(w io.Writer, req *Request) error {
	data, err := EncodeRequest(req)
	if err != nil {
		return err
	}
	_, err = w.Write(data)
	return err
}

// ReadRequest 从网络连接读取 Request
// 先读取全部数据，再解码
func ReadRequest(r io.Reader) (*Request, error) {
	// 先读 4 字节的头部长度
	headerLenBuf := make([]byte, 4)
	if _, err := io.ReadFull(r, headerLenBuf); err != nil {
		return nil, err
	}
	headerLen := binary.BigEndian.Uint32(headerLenBuf)

	// 读取头部数据
	headerData := make([]byte, headerLen)
	if _, err := io.ReadFull(r, headerData); err != nil {
		return nil, err
	}

	// 读 4 字节的体长度
	bodyLenBuf := make([]byte, 4)
	if _, err := io.ReadFull(r, bodyLenBuf); err != nil {
		return nil, err
	}
	bodyLen := binary.BigEndian.Uint32(bodyLenBuf)

	// 读取体数据
	bodyData := make([]byte, bodyLen)
	if _, err := io.ReadFull(r, bodyData); err != nil {
		return nil, err
	}

	// 组装完整数据并解码
	fullData := make([]byte, 0, 4+len(headerData)+4+len(bodyData))
	fullData = append(fullData, headerLenBuf...)
	fullData = append(fullData, headerData...)
	fullData = append(fullData, bodyLenBuf...)
	fullData = append(fullData, bodyData...)

	return DecodeRequest(fullData)
}

// WriteResponse 将 Response 写入网络连接
func WriteResponse(w io.Writer, resp *Response) error {
	data, err := EncodeResponse(resp)
	if err != nil {
		return err
	}
	_, err = w.Write(data)
	return err
}

// ReadResponse 从网络连接读取 Response
func ReadResponse(r io.Reader) (*Response, error) {
	// 读头部长度
	headerLenBuf := make([]byte, 4)
	if _, err := io.ReadFull(r, headerLenBuf); err != nil {
		return nil, err
	}
	headerLen := binary.BigEndian.Uint32(headerLenBuf)

	// 读头部数据
	headerData := make([]byte, headerLen)
	if _, err := io.ReadFull(r, headerData); err != nil {
		return nil, err
	}

	// 读体长度
	bodyLenBuf := make([]byte, 4)
	if _, err := io.ReadFull(r, bodyLenBuf); err != nil {
		return nil, err
	}
	bodyLen := binary.BigEndian.Uint32(bodyLenBuf)

	// 读体数据
	bodyData := make([]byte, bodyLen)
	if _, err := io.ReadFull(r, bodyData); err != nil {
		return nil, err
	}

	// 组装并解码
	fullData := make([]byte, 0, 4+len(headerData)+4+len(bodyData))
	fullData = append(fullData, headerLenBuf...)
	fullData = append(fullData, headerData...)
	fullData = append(fullData, bodyLenBuf...)
	fullData = append(fullData, bodyData...)

	return DecodeResponse(fullData)
}

package mrpc

// ============================================================
// 序列化协议接口与实现
// ============================================================

// Serializer 序列化协议接口
// 不同的序列化协议（JSON、Protobuf、Gob）都实现这个接口
// 这样客户端和服务端可以通过同一个接口使用不同的序列化方式
type Serializer interface {
	// Encode 将数据序列化为字节流
	// 参数 data 是任意类型的结构体指针
	// 返回序列化后的字节切片
	Encode(data interface{}) ([]byte, error)

	// Decode 将字节流反序列化为数据
	// 参数 data 是目标结构体的指针（用于接收反序列化结果）
	// 参数 bytes 是序列化后的字节切片
	Decode(data interface{}, bytes []byte) error

	// Code 返回序列化协议的唯一标识码
	// 用于在协议头中标记使用的序列化方式
	// 客户端和服务端之间需要协商一致
	// 例如：1=JSON, 2=Protobuf, 3=Gob
	Code() uint8
}

import "encoding/json"

// JSONSerializer 使用 JSON 格式进行序列化
// JSON 是最通用的序列化格式，可读性好，但体积较大
// 适合开发调试阶段使用
type JSONSerializer struct{}

// Encode 将数据序列化为 JSON 字节流
func (s *JSONSerializer) Encode(data interface{}) ([]byte, error) {
	return json.Marshal(data)
}

// Decode 将 JSON 字节流反序列化为数据
func (s *JSONSerializer) Decode(data interface{}, bytes []byte) error {
	return json.Unmarshal(bytes, data)
}

// Code 返回 JSON 序列化协议的标识码
// 约定：1 代表 JSON
func (s *JSONSerializer) Code() uint8 {
	return 1
}

import "google.golang.org/protobuf/proto"

// ProtoSerializer 使用 Protobuf 格式进行序列化
// Protobuf 体积小、速度快，但需要预先定义 .proto 文件并生成代码
// 适合生产环境使用
type ProtoSerializer struct{}

// Encode 将数据序列化为 Protobuf 字节流
// 注意：data 必须是 proto.Message 接口的实现
func (s *ProtoSerializer) Encode(data interface{}) ([]byte, error) {
	// 类型断言：将 interface{} 转换为 proto.Message
	msg, ok := data.(proto.Message)
	if !ok {
		return nil, fmt.Errorf("数据未实现 proto.Message 接口")
	}
	return proto.Marshal(msg)
}

// Decode 将 Protobuf 字节流反序列化为数据
func (s *ProtoSerializer) Decode(data interface{}, bytes []byte) error {
	msg, ok := data.(proto.Message)
	if !ok {
		return fmt.Errorf("数据未实现 proto.Message 接口")
	}
	return proto.Unmarshal(bytes, msg)
}

// Code 返回 Protobuf 序列化协议的标识码
// 约定：2 代表 Protobuf
func (s *ProtoSerializer) Code() uint8 {
	return 2
}

import "sync"

// ============================================================
// 序列化协议注册表
// ============================================================

// SerializerRegistry 序列化协议注册表
// 根据 Code 查找对应的 Serializer 实例
// 客户端和服务端都需要注册自己支持的序列化协议
type SerializerRegistry struct {
	mu      sync.RWMutex                  // 读写锁，保护并发访问
	serializers map[uint8]Serializer       // 存储已注册的序列化器
}

// NewSerializerRegistry 创建一个新的序列化协议注册表
func NewSerializerRegistry() *SerializerRegistry {
	return &SerializerRegistry{
		serializers: make(map[uint8]Serializer),
	}
}

// Register 注册一个序列化协议
// 参数 serializer 是实现了 Serializer 接口的实例
func (r *SerializerRegistry) Register(serializer Serializer) {
	r.mu.Lock()
	defer r.mu.Unlock()
	// 以 Code 为 key 存储序列化器
	r.serializers[serializer.Code()] = serializer
}

// Get 根据 Code 获取对应的序列化器
func (r *SerializerRegistry) Get(code uint8) (Serializer, error) {
	r.mu.RLock()
	defer r.mu.RUnlock()
	s, ok := r.serializers[code]
	if !ok {
		return nil, fmt.Errorf("未找到序列化协议: code=%d", code)
	}
	return s, nil
}

// ============================================================
// 改造后的客户端
// ============================================================

// ClientV2 是支持多序列化协议的 RPC 客户端
type ClientV2 struct {
	conn         net.Conn                  // TCP 连接
	serializer   Serializer                // 默认使用的序列化器
	registry     *SerializerRegistry       // 序列化协议注册表
	messageID    uint64                    // 消息ID计数器（用于生成唯一ID）
	mu           sync.Mutex                // 保护 messageID 的并发访问
}

// NewClientV2 创建新的客户端
// 参数 serializer 指定默认使用的序列化协议
func NewClientV2(addr string, serializer Serializer) (*ClientV2, error) {
	conn, err := net.DialTimeout("tcp", addr, 5*time.Second)
	if err != nil {
		return nil, fmt.Errorf("连接服务器失败: %w", err)
	}

	// 创建序列化注册表并注册默认序列化器
	registry := NewSerializerRegistry()
	registry.Register(serializer)

	return &ClientV2{
		conn:       conn,
		serializer: serializer,
		registry:   registry,
	}, nil
}

// CallV2 发起 RPC 调用（支持多序列化协议）
// 参数：
//   - ctx: 上下文，用于超时控制
//   - service: 服务名
//   - method: 方法名
//   - args: 请求参数
func (c *ClientV2) CallV2(ctx context.Context, service, method string, args interface{}) (*Response, error) {
	// 第一步：使用序列化器将参数编码为字节流
	body, err := c.serializer.Encode(args)
	if err != nil {
		return nil, fmt.Errorf("序列化请求参数失败: %w", err)
	}

	// 第二步：构造请求
	// 生成唯一的消息ID
	c.mu.Lock()
	c.messageID++
	msgID := c.messageID
	c.mu.Unlock()

	// 从 context 中提取需要传递的元数据
	meta := extractMetaFromContext(ctx)

	req := &Request{
		Header: RequestHeader{
			Version:    1,                      // 协议版本
			Serializer: c.serializer.Code(),    // 序列化协议标识
			Compressor: 0,                      // 暂不压缩
			MessageID:  msgID,                  // 消息ID
			Service:    service,                // 服务名
			Method:     method,                 // 方法名
			Meta:       meta,                   // 元数据
		},
		Body: body, // 序列化后的请求参数
	}

	// 第三步：发送请求
	if err := WriteRequest(c.conn, req); err != nil {
		return nil, fmt.Errorf("发送请求失败: %w", err)
	}

	// 第四步：读取响应
	// 使用 select + channel 实现超时控制
	respCh := make(chan *Response, 1)
	errCh := make(chan error, 1)

	go func() {
		resp, err := ReadResponse(c.conn)
		if err != nil {
			errCh <- err
			return
		}
		respCh <- resp
	}()

	// 等待响应或超时
	select {
	case resp := <-respCh:
		// 检查响应头中的错误信息
		if resp.Header.Error != "" {
			return resp, fmt.Errorf("服务端错误: %s", resp.Header.Error)
		}
		return resp, nil
	case err := <-errCh:
		return nil, fmt.Errorf("读取响应失败: %w", err)
	case <-ctx.Done():
		// 超时或被取消
		return nil, ctx.Err()
	}
}

// extractMetaFromContext 从 context 中提取需要传递的元数据
// 例如：trace ID、超时时间戳、单向调用标记等
func extractMetaFromContext(ctx context.Context) map[string]string {
	meta := make(map[string]string)

	// 提取超时时间戳（如果有）
	if deadline, ok := ctx.Deadline(); ok {
		// 将截止时间转为 Unix 毫秒时间戳传递
		meta["deadline"] = strconv.FormatInt(deadline.UnixMilli(), 10)
	}

	// 提取单向调用标记（如果有）
	if v := ctx.Value("one-way"); v != nil {
		meta["one-way"] = "true"
	}

	return meta
}

// ============================================================
// 改造后的服务端
// ============================================================

// ServerV2 是支持多序列化协议的 RPC 服务端
type ServerV2 struct {
	addr        string                      // 监听地址
	mu          sync.RWMutex                // 保护 serviceMap
	serviceMap  map[string]reflect.Value    // 服务注册表
	registry    *SerializerRegistry         // 序列化协议注册表
}

// NewServerV2 创建新的服务端
func NewServerV2(addr string) *ServerV2 {
	registry := NewSerializerRegistry()
	// 注册默认的序列化协议
	registry.Register(&JSONSerializer{})
	registry.Register(&ProtoSerializer{})

	return &ServerV2{
		addr:       addr,
		serviceMap: make(map[string]reflect.Value),
		registry:   registry,
	}
}

// handleConnV2 处理连接（支持多序列化协议）
func (s *ServerV2) handleConnV2(conn net.Conn) {
	defer conn.Close()

	for {
		// 读取请求
		req, err := ReadRequest(conn)
		if err != nil {
			if errors.Is(err, io.EOF) {
				return
			}
			return
		}

		// 根据请求头中的序列化协议标识，找到对应的序列化器
		serializer, err := s.registry.Get(req.Header.Serializer)
		if err != nil {
			// 找不到序列化器，返回错误响应
			s.sendError(conn, req.Header.MessageID, err.Error())
			continue
		}

		// 根据服务名查找服务
		s.mu.RLock()
		svcValue, ok := s.serviceMap[req.Header.Service]
		s.mu.RUnlock()
		if !ok {
			s.sendError(conn, req.Header.MessageID, fmt.Sprintf("服务 %s 不存在", req.Header.Service))
			continue
		}

		// 根据方法名查找方法
		svcType := svcValue.Type()
		method, ok := svcType.MethodByName(req.Header.Method)
		if !ok {
			s.sendError(conn, req.Header.MessageID, fmt.Sprintf("方法 %s 不存在", req.Header.Method))
			continue
		}

		// 使用序列化器将 Body 反序列化为方法参数
		// 约定：方法第二个参数是请求结构体指针
		argType := method.Type.In(2) // In(0)=接收者, In(1)=context, In(2)=请求参数
		argValue := reflect.New(argType.Elem()) // 创建请求结构体的指针
		if err := serializer.Decode(argValue.Interface(), req.Body); err != nil {
			s.sendError(conn, req.Header.MessageID, fmt.Sprintf("反序列化参数失败: %v", err))
			continue
		}

		// 根据元数据重建 context
		// 如果元数据中包含 deadline，则设置超时
		ctx := context.Background()
		ctx = rebuildContext(ctx, req.Header.Meta)

		// 反射调用方法
		in := []reflect.Value{
			svcValue,                        // 接收者
			reflect.ValueOf(ctx),            // context.Context
			argValue,                        // 请求参数
		}
		out := method.Func.Call(in)

		// 处理返回值
		var resp Response
		if len(out) >= 2 && !out[1].IsNil() {
			// 有错误返回
			errMsg := out[1].Interface().(error).Error()
			resp.Header = ResponseHeader{
				Version:    1,
				Serializer: req.Header.Serializer,
				MessageID:  req.Header.MessageID,
				Error:      errMsg,
			}
		} else {
			// 成功，序列化响应数据
			respData, err := serializer.Encode(out[0].Interface())
			if err != nil {
				s.sendError(conn, req.Header.MessageID, fmt.Sprintf("序列化响应失败: %v", err))
				continue
			}
			resp.Header = ResponseHeader{
				Version:    1,
				Serializer: req.Header.Serializer,
				MessageID:  req.Header.MessageID,
			}
			resp.Body = respData
		}

		// 写回响应
		if err := WriteResponse(conn, &resp); err != nil {
			return
		}
	}
}

// sendError 发送错误响应
func (s *ServerV2) sendError(conn net.Conn, msgID uint64, errMsg string) {
	resp := &Response{
		Header: ResponseHeader{
			Version:   1,
			MessageID: msgID,
			Error:     errMsg,
		},
	}
	WriteResponse(conn, resp)
}

// rebuildContext 根据元数据重建 context
// 如果元数据中包含 deadline，则创建带超时的 context
func rebuildContext(ctx context.Context, meta map[string]string) context.Context {
	if deadlineStr, ok := meta["deadline"]; ok {
		// 解析时间戳
		deadlineMs, err := strconv.ParseInt(deadlineStr, 10, 64)
		if err == nil {
			deadline := time.UnixMilli(deadlineMs)
			// 用剩余时间创建带超时的 context
			// 如果 deadline 已经过去，timeout 会是负数，context 会立即超时
			timeout := time.Until(deadline)
			if timeout > 0 {
				ctx, _ = context.WithDeadline(ctx, deadline)
			}
		}
	}
	return ctx
}

func main() {
	// --- 服务端初始化 ---
	server := NewServerV2(":9091")
	// 注册默认序列化协议（JSON 和 Proto 已在 NewServerV2 中注册）

	// 注册服务
	server.Register(&UserService{})

	go server.Start()

	// --- 客户端初始化 ---
	// 创建 JSON 序列化器
	jsonSerializer := &JSONSerializer{}

	// 创建客户端，指定使用 JSON 序列化
	client, err := NewClientV2("127.0.0.1:9091", jsonSerializer)
	if err != nil {
		fmt.Printf("连接失败: %v\n", err)
		return
	}

	// 发起调用
	ctx := context.Background()
	resp, err := client.CallV2(ctx, "UserService", "GetById", &GetByIdRequest{Id: 123})
	if err != nil {
		fmt.Printf("调用失败: %v\n", err)
		return
	}

	// 反序列化响应
	var user GetByIdResponse
	json.Unmarshal(resp.Body, &user)
	fmt.Printf("查询结果: %+v\n", user)
}

// ============================================================
// 单向调用支持
// ============================================================

// WithOneWay 在 context 中设置单向调用标记
// 用户通过这个函数创建带标记的 context，传给 RPC 客户端
// 客户端检测到标记后，不会等待响应
// 服务端检测到标记后，不会发回响应
func WithOneWay(ctx context.Context) context.Context {
	// 使用 context.WithValue 在 context 中携带单向调用标记
	// key 使用自定义类型，避免与其他包的 key 冲突
	return context.WithValue(ctx, oneWayKey{}, "true")
}

// oneWayKey 是单向调用标记的 context key 类型
// 使用自定义类型而不是字符串，避免 key 冲突
type oneWayKey struct{}

// isOneWay 检查 context 是否标记为单向调用
func isOneWay(ctx context.Context) bool {
	v := ctx.Value(oneWayKey{})
	return v != nil
}

// CallOneWay 发起单向调用
// 与普通调用的区别：
// 1. 在元数据中带上 one-way 标记
// 2. 发送请求后不等待响应
func (c *ClientV2) CallOneWay(ctx context.Context, service, method string, args interface{}) error {
	// 在 context 中设置单向调用标记
	ctx = WithOneWay(ctx)

	// 序列化请求参数
	body, err := c.serializer.Encode(args)
	if err != nil {
		return fmt.Errorf("序列化请求参数失败: %w", err)
	}

	// 生成消息ID
	c.mu.Lock()
	c.messageID++
	msgID := c.messageID
	c.mu.Unlock()

	// 提取元数据（包含 one-way 标记）
	meta := extractMetaFromContext(ctx)

	// 构造请求
	req := &Request{
		Header: RequestHeader{
			Version:    1,
			Serializer: c.serializer.Code(),
			MessageID:  msgID,
			Service:    service,
			Method:     method,
			Meta:       meta,
		},
		Body: body,
	}

	// 发送请求后直接返回，不等待响应
	// 这样客户端资源（goroutine、内存等）可以立即释放
	return WriteRequest(c.conn, req)
}

// handleConnV3 处理连接（支持单向调用）
func (s *ServerV2) handleConnV3(conn net.Conn) {
	defer conn.Close()

	for {
		req, err := ReadRequest(conn)
		if err != nil {
			if errors.Is(err, io.EOF) {
				return
			}
			return
		}

		// 检查是否为单向调用
		isOneWayCall := req.Header.Meta["one-way"] == "true"

		// 处理请求（查找服务、反序列化、反射调用等逻辑与之前相同）
		resp := s.processRequest(req)

		// 如果是单向调用，不发回响应，直接处理下一个请求
		// 这样服务端的连接和 goroutine 资源可以更快释放
		if isOneWayCall {
			continue
		}

		// 非单向调用，写回响应
		if err := WriteResponse(conn, resp); err != nil {
			return
		}
	}
}

// CallWithTimeout 带超时控制的 RPC 调用
// 使用 select + channel 同时监听响应和超时
func (c *ClientV2) CallWithTimeout(ctx context.Context, service, method string, args interface{}) (*Response, error) {
	// 序列化请求参数
	body, err := c.serializer.Encode(args)
	if err != nil {
		return nil, fmt.Errorf("序列化失败: %w", err)
	}

	// 生成消息ID
	c.mu.Lock()
	c.messageID++
	msgID := c.messageID
	c.mu.Unlock()

	// 提取元数据（包含超时时间戳）
	meta := extractMetaFromContext(ctx)

	// 构造请求
	req := &Request{
		Header: RequestHeader{
			Version:    1,
			Serializer: c.serializer.Code(),
			MessageID:  msgID,
			Service:    service,
			Method:     method,
			Meta:       meta,
		},
		Body: body,
	}

	// 发送请求
	if err := WriteRequest(c.conn, req); err != nil {
		return nil, fmt.Errorf("发送请求失败: %w", err)
	}

	// 使用 select + channel 监听响应和超时
	// 这是 Go 中最经典的超时控制写法
	respCh := make(chan *Response, 1) // 带缓冲的 channel，防止 goroutine 泄漏
	errCh := make(chan error, 1)

	// 启动 goroutine 读取响应
	go func() {
		resp, err := ReadResponse(c.conn)
		if err != nil {
			errCh <- err
			return
		}
		respCh <- resp
	}()

	// select 会同时监听多个 channel
	// 哪个 channel 先有数据，就执行哪个 case
	select {
	case resp := <-respCh:
		// 正常收到响应
		if resp.Header.Error != "" {
			return resp, fmt.Errorf("服务端错误: %s", resp.Header.Error)
		}
		return resp, nil

	case err := <-errCh:
		// 读取响应出错
		return nil, fmt.Errorf("读取响应失败: %w", err)

	case <-ctx.Done():
		// context 超时或被取消
		// 注意：超时之后并没有中断正在执行的调用
		// 只是 RPC 客户端丢掉了后面的响应而已
		// 每次都要新创建一个 channel，性能损耗比较大
		// 但这种实现很简单也很清晰
		return nil, ctx.Err()
	}
}

// ============================================================
// 超时时间跨端传递实现
// ============================================================

// 设置超时时间戳到元数据
// 在前面的 extractMetaFromContext 函数中已经实现：
// 如果 context 有 deadline，会将 deadline 转为 Unix 毫秒时间戳
// 存入 meta["deadline"]
//
// 服务端在 rebuildContext 函数中会读取这个时间戳，
// 用它创建新的带超时的 context

// 客户端设置 meta 的过程（已在 extractMetaFromContext 中实现）：
func extractMetaFromContext(ctx context.Context) map[string]string {
	meta := make(map[string]string)

	// 如果 context 设置了 deadline，将其转为时间戳传递
	if deadline, ok := ctx.Deadline(); ok {
		// 将截止时间转为 Unix 毫秒时间戳
		// 使用时间戳方案，服务端可以直接用这个时间戳重建 context
		meta["deadline"] = strconv.FormatInt(deadline.UnixMilli(), 10)
	}

	// 提取单向调用标记
	if v := ctx.Value(oneWayKey{}); v != nil {
		meta["one-way"] = "true"
	}

	return meta
}

// 服务端重建 context 的过程（已在 rebuildContext 中实现）：
func rebuildContext(ctx context.Context, meta map[string]string) context.Context {
	if deadlineStr, ok := meta["deadline"]; ok {
		// 解析时间戳
		deadlineMs, err := strconv.ParseInt(deadlineStr, 10, 64)
		if err == nil {
			deadline := time.UnixMilli(deadlineMs)
			// 用这个 deadline 创建新的 context
			// 如果 deadline 已经过去，context 会立即超时
			ctx, _ = context.WithDeadline(ctx, deadline)
		}
	}
	return ctx
}

func main() {
	// 创建客户端
	client, _ := NewClientV2("127.0.0.1:9091", &JSONSerializer{})
	defer client.Close()

	// 创建带 3 秒超时的 context
	// 这 3 秒是整个调用链的超时时间
	// 如果服务端再调用其他服务，剩余超时时间会继续传递
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel() // 确保资源释放

	// 发起调用，传入带超时的 ctx
	resp, err := client.CallWithTimeout(ctx, "UserService", "GetById", &GetByIdRequest{Id: 123})
	if err != nil {
		if errors.Is(err, context.DeadlineExceeded) {
			fmt.Println("调用超时")
			return
		}
		fmt.Printf("调用失败: %v\n", err)
		return
	}

	// 处理响应
	var user GetByIdResponse
	json.Unmarshal(resp.Body, &user)
	fmt.Printf("查询结果: %+v\n", user)
}