逻辑层实现：gRPC 服务与消息处理

网关只负责「把字节从连接上搬进搬出」，真正决定消息能不能发、发给谁、落哪张表的逻辑全在逻辑层。第 03 章的 send_msg 帧到网关后，会被转成一次 gRPC（Google Remote Procedure Call）调用；逻辑层是无状态服务：任何一台网关都能调它，它从不持有连接，因此可以按 CPU 直接扩副本，消息的可靠性与顺序性由它的幂等（idempotent）落库与投递编排保证。

本章目标

把 logic/ 从空目录跑成一个能被 grpcurl 调通的 gRPC 服务（监听 logic.grpc_addr，默认 0.0.0.0:9090），落地 SendMessage、PullOffline、AckMessage 三个方法，并说清每个方法里哪一步会失败、失败后由谁兜底。

涉及的文件

入口 cmd/logic/main.go（gRPC 端口 + 第 04 章的 HTTP 鉴权口 8081）；logic/server.go（三个 RPC 实现）；logic/validate.go（入参校验，集中在 RPC 边界）；logic/interceptor.go（trace_id 透传、panic 兜底、慢调用日志）；service/message.go 与 service/group.go（业务编排）；store/（MySQL 与 Redis 访问）；proto/im.proto（生成物落到 pkg/pb）。

proto：先把接口钉死

syntax = "proto3";
package im.v1;
option go_package = "github.com/example/im-go/pkg/pb;pb";

service LogicService {
  rpc SendMessage(SendMessageReq) returns (SendMessageResp);
  rpc PullOffline(PullOfflineReq) returns (PullOfflineResp);
  rpc AckMessage(AckMessageReq) returns (AckMessageResp);
}

message SendMessageReq {
  string client_msg_id = 1; // 客户端生成，重发时复用，幂等键（第 10 章）
  uint64 from_uid      = 2; // 网关只填会话里的 uid，绝不采信报文里的 from
  uint64 to_uid        = 3; // 单聊收件人
  uint64 group_id      = 4; // 群聊群 ID
  int32  conv_type     = 5; // 1=单聊 2=群聊
  bytes  payload       = 6; // 第 03 章 Envelope.Body 原样透传，逻辑层不解析
  string device_id     = 7; // 上行来源设备，多端去重用
}

SendMessageResp 只回三个字段：msg_id（服务端权威 ID）、seq（会话内递增序号）、sent_at（落库毫秒时间戳），网关把它们填进 send_msg 的响应信封回给客户端。PullOfflineReq 带 uid、device_id、cursors（[{conv_id, seq}] 的设备位点）与 limit（第 09 章展开），AckMessageReq 带 msg_id、uid、conv_id（第 10 章展开）。

进程入口：注册服务与优雅停止

func main() {
	confPath := flag.String("conf", "configs/config.yaml", "配置文件路径") // 与第 01 章一致
	flag.Parse()
	cfg := mustConfig(*confPath)          // config.Load 失败即打日志退出，下同
	logger := logx.Init(cfg.Env, "logic") // 全项目统一 slog JSON 输出
	st := mustStore(cfg)                  // store.Open：MySQL + Redis 两个连接池
	defer st.Close()
	lis, err := net.Listen("tcp", cfg.Logic.GRPCAddr) // 只监听内网地址，绝不暴露公网
	if err != nil {
		logger.Error("listen failed", "grpc_addr", cfg.Logic.GRPCAddr, "err", err)
		os.Exit(1)
	}
	srv := grpc.NewServer(grpc.ChainUnaryInterceptor(logic.RecoverInterceptor, logic.TraceInterceptor))
	pb.RegisterLogicServiceServer(srv, logic.NewServer(st, cfg))

	go func() {
		logger.Info("logic serving", "grpc_addr", cfg.Logic.GRPCAddr)
		logger.Error("serve", "err", srv.Serve(lis)) // Serve 返回即监听失效，进程应退出
	}()

	// 收到信号先 GracefulStop（等在途请求做完）再退出，发布时才不会掐断网关调用
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()
	<-ctx.Done()
	srv.GracefulStop()
}

SendMessage：网关上行的唯一入口

// SendMessage 是上行消息的唯一入口：先做参数校验，再交给业务服务落库与投递，
// 最后把权威的 msg_id / seq 回给网关，由网关填进响应信封。
func (s *Server) SendMessage(ctx context.Context, req *pb.SendMessageReq) (*pb.SendMessageResp, error) {
	if err := validateSend(req); err != nil {
		return nil, status.Error(codes.InvalidArgument, err.Error())
	}

	res, err := s.msg.Send(ctx, service.SendInput{
		ClientMsgID: req.GetClientMsgId(), FromUID: req.GetFromUid(),
		ToUID: req.GetToUid(), GroupID: req.GetGroupId(),
		ConvType: int(req.GetConvType()), Payload: req.GetPayload(),
		DeviceID: req.GetDeviceId(),
	})
	if err != nil {
		// 业务错误翻译成状态码，网关按码决定「不重试」还是「可重试」
		switch {
		case errors.Is(err, service.ErrDuplicate):
			return nil, status.Error(codes.AlreadyExists, "duplicate client_msg_id")
		case errors.Is(err, service.ErrNotMember):
			return nil, status.Error(codes.PermissionDenied, "sender not in group")
		default:
			slog.Error("send message", "uid", req.GetFromUid(), "client_msg_id", req.GetClientMsgId(), "err", err)
			return nil, status.Error(codes.Internal, "send failed")
		}
	}
	return &pb.SendMessageResp{MsgId: res.MsgID, Seq: res.Seq, SentAt: res.SentAt}, nil
}

边界校验的规则很简单，但缺一条都会把脏数据写进库：from_uid/client_msg_id 必填，payload 长度必须落在 (0, 32KB]（与第 03 章的 MaxFrameSize 对齐），单聊必须有 to_uid、群聊必须有 group_id。PullOffline 按设备位点增量补齐，limit 要钳到 200 以内——第 09 章的 OfflineStore.Pull 已经这么做了，RPC 边界再钳一次防越权；AckMessage 只做一件事——把消息状态从「已落库」改成「已送达」，第一次成功后才给发送方推回执。

验证方式

cd im-go && protoc --go_out=. --go-grpc_out=. proto/im.proto && go build -o bin/logic ./cmd/logic
./bin/logic -conf configs/config.yaml
# 期望日志：level=INFO msg="logic serving" service=logic grpc_addr=0.0.0.0:9090

# 单聊上行：payload 是 bytes 用 base64（aGVsbG8= 即 hello）；uint64 在 protojson 里是字符串
grpcurl -plaintext -d '{"client_msg_id":"c-1","from_uid":"1001","to_uid":"1002","conv_type":1,
  "payload":"aGVsbG8=","device_id":"ios-8a2c"}' 127.0.0.1:9090 im.v1.LogicService/SendMessage
# 期望输出：{"msgId":"70001","seq":"1","sentAt":"1735689600123"}
# 重发同一个 client_msg_id 应返回相同 msgId/seq；缺 client_msg_id 则报 InvalidArgument
grpcurl -plaintext -d '{"from_uid":"1001"}' 127.0.0.1:9090 im.v1.LogicService/SendMessage
# 期望输出：ERROR: Code: InvalidArgument Message: client_msg_id required

常见坑

坑	现象	正确做法
逻辑层缓存了在线表	扩副本后投递忽好忽坏	会话与路由只放 Redis，逻辑层保持无状态
handler 里 `go func()` 后直接返回	落库报 `context canceled`	同步做完，或换独立的带超时上下文
错误码一律 `Internal`	网关对不可重试的错误反复重试	`InvalidArgument` 不重试，`Internal` 才重试

小结：逻辑层是「无状态 + 幂等」的 gRPC 服务——SendMessage 负责校验、落库与投递编排，PullOffline 按设备位点补齐，AckMessage 把送达状态变成回执；业务错误要在 RPC 边界翻译成网关能据以决策的状态码，拦截器负责 trace_id 透传与 panic 兜底。下一章解决它投递时最依赖的一块数据：在线路由表。

笔记加载中…

标签：#IM #Go #项目实战

返回知识库