四种通信模式：一元、服务端流、客户端流、双向流

一条 IM 链路上四种调用形态同时存在：发消息是「一问一答」，消息下发是「一次请求、持续推送」，已读位点上报是「多次上报、一次确认」，接入层与逻辑层之间的会话通道要「双方同时收发」。gRPC 用 stream 关键字区分这四种模式：写在参数位置 → 客户端流；写在返回值位置 → 服务端流；两边都有 → 双向流；都没写 → 一元（unary，第 4 章）。

语义：stream 写在哪一侧，模式就是哪一种

syntax = "proto3";
package im.message.v1;
option go_package = "example.com/im/gen/message/v1;messagev1";

service MessageService {
  rpc SendMsg (SendMsgReq) returns (SendMsgResp);              // 一元
  rpc Subscribe (SubscribeReq) returns (stream Message);        // 服务端流
  rpc ReportRead (stream ReadReport) returns (ReportReadResp);  // 客户端流
  rpc Chat (stream ChatFrame) returns (ChatFrame);              // 双向流
}

message SendMsgReq { int64 uid = 1; string text = 2; }
message SendMsgResp { string msg_id = 1; int64 seq = 2; }
message SubscribeReq { int64 uid = 1; }
message Message { int64 seq = 1; string text = 2; }
message ReadReport { int64 uid = 1; int64 seq = 2; }
message ReportReadResp { int32 count = 1; }
message ChatFrame { string body = 1; }

服务端：三个流式方法的签名

一元方法就是普通函数；三种流式方法的第二个参数是各自的流对象（MessageService_SubscribeServer 等），只有流对象上才有 Send/Recv。「多条」体现在流对象上，而不是多调用几次函数。

type messageServer struct {
	pb.UnimplementedMessageServiceServer // 未实现的方法统一返回 Unimplemented（第 4 章）
}

// 服务端流：Send 多次；返回 nil 表示正常结束，客户端 Recv 得到 io.EOF
func (s *messageServer) Subscribe(req *pb.SubscribeReq, stream pb.MessageService_SubscribeServer) error {
	for i := int64(0); i < 3; i++ {
		if err := stream.Send(&pb.Message{Seq: i, Text: "hi"}); err != nil {
			return err // Send 失败多半是 ctx 到期或对端已断开
		}
	}
	return nil
}

// 客户端流：Recv 到 io.EOF 后用 SendAndClose 回唯一的响应（只能调一次）
func (s *messageServer) ReportRead(stream pb.MessageService_ReportReadServer) error {
	var n int32
	for {
		if _, err := stream.Recv(); err != nil {
			if errors.Is(err, io.EOF) {
				return stream.SendAndClose(&pb.ReportReadResp{Count: n})
			}
			return err
		}
		n++
	}
}

// 双向流：读一帧回一帧；客户端 CloseSend 之后 Recv 返回 io.EOF
func (s *messageServer) Chat(stream pb.MessageService_ChatServer) error {
	for {
		frame, err := stream.Recv()
		if errors.Is(err, io.EOF) {
			return nil // 正常收尾；直接 return err 会被归成 Unknown
		}
		if err != nil {
			return err
		}
		if err := stream.Send(&pb.ChatFrame{Body: "echo:" + frame.GetBody()}); err != nil {
			return err
		}
	}
}

客户端：流式要先拿流对象

一元调用直接同步返回（第 4 章）；流式方法都是「先拿到流对象，再 Send/Recv」，三种模式共用同一个注册入口（pb.RegisterMessageServiceServer）。服务端流循环 Recv 直到 io.EOF；客户端流是多次 up.Send(...) 后 up.CloseAndRecv()；双向流必须把读写拆到两个 goroutine：

// 服务端流：持续 Recv，只有 io.EOF 才是正常结束（其余示例从略错误检查）
sub, _ := cli.Subscribe(ctx, &pb.SubscribeReq{Uid: 1})
for {
	msg, err := sub.Recv()
	if errors.Is(err, io.EOF) {
		break // 服务端 return nil 的正常收尾
	}
	if err != nil {
		return err
	}
	log.Println(msg.GetText())
}

up, _ := cli.ReportRead(ctx) // 客户端流：Send 多次 → CloseAndRecv 取回唯一响应
_ = up.Send(&pb.ReadReport{Uid: 1, Seq: 1})
_, _ = up.CloseAndRecv()

// 双向流：写侧 goroutine + 读侧循环，两个方向互不阻塞
bidi, _ := cli.Chat(ctx)
go func() {
	defer bidi.CloseSend() // 不 CloseSend，服务端永远等不到 io.EOF
	_ = bidi.Send(&pb.ChatFrame{Body: "ping"})
}()
for {
	frame, err := bidi.Recv()
	if errors.Is(err, io.EOF) {
		break
	}
	if err != nil {
		return err
	}
	log.Println(frame.GetBody())
}

四种模式对比

维度	一元	服务端流	客户端流	双向流
请求 / 响应	1 / 1	1 / N	N / 1	N / N
服务端结束	直接返回	返回 nil	SendAndClose	返回 nil
客户端收尾	无	Recv 到 io.EOF	CloseAndRecv	CloseSend + Recv 到 io.EOF

常见坑

• 流的隐藏成本：每个 RPC 都是独立的 HTTP/2 stream（连接管理与多路复用见第 11 章），而且 Send 受 HTTP/2 流控窗口约束（第 12 章），下游消费慢时 Send 会一直卡住，必须靠 ctx 兜底。
• 流的结束动作别写错：SendAndClose 只能调一次且调用后不能再 Recv；服务端处理器也不要 return err 传 io.EOF——它不是 gRPC 状态错误，会被转成 Unknown: EOF，判断后 return nil 才算正常结束。

小结：stream 写在参数位是客户端流、写在返回值位是服务端流、两边都有是双向流；服务端流返回 nil 结束，客户端流用 SendAndClose，双向流靠 CloseSend 加 io.EOF 收尾；流式省掉的是往返次数与等待，换来的是重试与错误处理更复杂，能用一元解决就别上流。