1、编写proto文档如下：

    1. syntax = "proto3";
    2. package proto;
    4. message RunMessage {
    5.     int64 time = 1; //unix时间
    6.     string id = 2; //爬虫名称
    7.     Status status = 3; //状态,
    8.     string remark = 4; //备注信息
    9. }
    11. service Trans {
    12.     //双向流 客户端发送状态，服务端发送指令
    13.     rpc SendStatus (stream RunMessage) returns (stream Order) {
    14.     }
    15.     //rpc Order(stream Order)returns(Status){}
    16. }
    17. message Order {
    18.     Status status = 1;
    19. }
    21. enum Status {
    22.     START = 0; //开始
    23.     RUNNING = 1; //执行中
    24.     END = 2; //结束
    25.     IDEAL = 3; //空闲中
    26. }

    2、执行命令： protoc —go_out=plugins=grpc:. .proto
    注意：明确    .proto文件的位置，要不会找不到*.proto文件，不能生成go的文件
    3、生成的代码

    1. // Code generated by protoc-gen-go. DO NOT EDIT.
    2. // source: message.proto
    4. package proto
    6. import (
    7.     context "context"
    8.     fmt "fmt"
    9.     proto "github.com/golang/protobuf/proto"
    10.     grpc "google.golang.org/grpc"
    11.     codes "google.golang.org/grpc/codes"
    12.     status "google.golang.org/grpc/status"
    13.     math "math"
    14. )
    16. // Reference imports to suppress errors if they are not otherwise used.
    17. var _ = proto.Marshal
    18. var _ = fmt.Errorf
    19. var _ = math.Inf
    21. // This is a compile-time assertion to ensure that this generated file
    22. // is compatible with the proto package it is being compiled against.
    23. // A compilation error at this line likely means your copy of the
    24. // proto package needs to be updated.
    25. const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package
    27. type Status int32
    29. const (
    30.     Status_START   Status = 0
    31.     Status_RUNNING Status = 1
    32.     Status_END     Status = 2
    33.     Status_IDEAL   Status = 3
    34. )
    36. var Status_name = map[int32]string{
    37.     0: "START",
    38.     1: "RUNNING",
    39.     2: "END",
    40.     3: "IDEAL",
    41. }
    43. var Status_value = map[string]int32{
    44.     "START":   0,
    45.     "RUNNING": 1,
    46.     "END":     2,
    47.     "IDEAL":   3,
    48. }
    50. func (x Status) String() string {
    51.     return proto.EnumName(Status_name, int32(x))
    52. }
    54. func (Status) EnumDescriptor() ([]byte, []int) {
    55.     return fileDescriptor_33c57e4bae7b9afd, []int{0}
    56. }
    58. type RunMessage struct {
    59.     Time                 int64    `protobuf:"varint,1,opt,name=time,proto3" json:"time,omitempty"`
    60.     Id                   string   `protobuf:"bytes,2,opt,name=id,proto3" json:"id,omitempty"`
    61.     Status               Status   `protobuf:"varint,3,opt,name=status,proto3,enum=proto.Status" json:"status,omitempty"`
    62.     Remark               string   `protobuf:"bytes,4,opt,name=remark,proto3" json:"remark,omitempty"`
    63.     XXX_NoUnkeyedLiteral struct{} `json:"-"`
    64.     XXX_unrecognized     []byte   `json:"-"`
    65.     XXX_sizecache        int32    `json:"-"`
    66. }
    68. func (m *RunMessage) Reset()         { *m = RunMessage{} }
    69. func (m *RunMessage) String() string { return proto.CompactTextString(m) }
    70. func (*RunMessage) ProtoMessage()    {}
    71. func (*RunMessage) Descriptor() ([]byte, []int) {
    72.     return fileDescriptor_33c57e4bae7b9afd, []int{0}
    73. }
    75. func (m *RunMessage) XXX_Unmarshal(b []byte) error {
    76.     return xxx_messageInfo_RunMessage.Unmarshal(m, b)
    77. }
    78. func (m *RunMessage) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
    79.     return xxx_messageInfo_RunMessage.Marshal(b, m, deterministic)
    80. }
    81. func (m *RunMessage) XXX_Merge(src proto.Message) {
    82.     xxx_messageInfo_RunMessage.Merge(m, src)
    83. }
    84. func (m *RunMessage) XXX_Size() int {
    85.     return xxx_messageInfo_RunMessage.Size(m)
    86. }
    87. func (m *RunMessage) XXX_DiscardUnknown() {
    88.     xxx_messageInfo_RunMessage.DiscardUnknown(m)
    89. }
    91. var xxx_messageInfo_RunMessage proto.InternalMessageInfo
    93. func (m *RunMessage) GetTime() int64 {
    94.     if m != nil {
    95.         return m.Time
    96.     }
    97.     return 0
    98. }
    100. func (m *RunMessage) GetId() string {
    101.     if m != nil {
    102.         return m.Id
    103.     }
    104.     return ""
    105. }
    107. func (m *RunMessage) GetStatus() Status {
    108.     if m != nil {
    109.         return m.Status
    110.     }
    111.     return Status_START
    112. }
    114. func (m *RunMessage) GetRemark() string {
    115.     if m != nil {
    116.         return m.Remark
    117.     }
    118.     return ""
    119. }
    121. type Order struct {
    122.     Status               Status   `protobuf:"varint,1,opt,name=status,proto3,enum=proto.Status" json:"status,omitempty"`
    123.     XXX_NoUnkeyedLiteral struct{} `json:"-"`
    124.     XXX_unrecognized     []byte   `json:"-"`
    125.     XXX_sizecache        int32    `json:"-"`
    126. }
    128. func (m *Order) Reset()         { *m = Order{} }
    129. func (m *Order) String() string { return proto.CompactTextString(m) }
    130. func (*Order) ProtoMessage()    {}
    131. func (*Order) Descriptor() ([]byte, []int) {
    132.     return fileDescriptor_33c57e4bae7b9afd, []int{1}
    133. }
    135. func (m *Order) XXX_Unmarshal(b []byte) error {
    136.     return xxx_messageInfo_Order.Unmarshal(m, b)
    137. }
    138. func (m *Order) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
    139.     return xxx_messageInfo_Order.Marshal(b, m, deterministic)
    140. }
    141. func (m *Order) XXX_Merge(src proto.Message) {
    142.     xxx_messageInfo_Order.Merge(m, src)
    143. }
    144. func (m *Order) XXX_Size() int {
    145.     return xxx_messageInfo_Order.Size(m)
    146. }
    147. func (m *Order) XXX_DiscardUnknown() {
    148.     xxx_messageInfo_Order.DiscardUnknown(m)
    149. }
    151. var xxx_messageInfo_Order proto.InternalMessageInfo
    153. func (m *Order) GetStatus() Status {
    154.     if m != nil {
    155.         return m.Status
    156.     }
    157.     return Status_START
    158. }
    160. func init() {
    161.     proto.RegisterEnum("proto.Status", Status_name, Status_value)
    162.     proto.RegisterType((*RunMessage)(nil), "proto.RunMessage")
    163.     proto.RegisterType((*Order)(nil), "proto.Order")
    164. }
    166. func init() { proto.RegisterFile("message.proto", fileDescriptor_33c57e4bae7b9afd) }
    168. var fileDescriptor_33c57e4bae7b9afd = []byte{
    169.     // 230 bytes of a gzipped FileDescriptorProto
    170.     0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x7c, 0x8f, 0x41, 0x4b, 0xc3, 0x30,
    171.     0x14, 0xc7, 0xf7, 0xda, 0xb5, 0x63, 0x4f, 0x37, 0xea, 0x3b, 0x48, 0xf0, 0x54, 0x0a, 0x42, 0xf0,
    172.     0x50, 0x64, 0xf3, 0xe8, 0x65, 0xb0, 0x21, 0x03, 0x8d, 0x90, 0xd6, 0x0f, 0x10, 0x49, 0x90, 0x22,
    173.     0x6d, 0x25, 0x49, 0xbf, 0xbf, 0x98, 0x06, 0xf4, 0xe4, 0x29, 0xc9, 0xef, 0x9f, 0xf7, 0x7b, 0xef,
    174.     0xe1, 0xa6, 0x37, 0xce, 0xa9, 0x0f, 0x53, 0x7f, 0xd9, 0xd1, 0x8f, 0x94, 0x85, 0xa3, 0x1a, 0x11,
    175.     0xe5, 0x34, 0xbc, 0xcc, 0x11, 0x11, 0x2e, 0x7d, 0xd7, 0x1b, 0x06, 0x25, 0xf0, 0x54, 0x86, 0x3b,
    176.     0x6d, 0x31, 0xe9, 0x34, 0x4b, 0x4a, 0xe0, 0x6b, 0x99, 0x74, 0x9a, 0x6e, 0x31, 0x77, 0x5e, 0xf9,
    177.     0xc9, 0xb1, 0xb4, 0x04, 0xbe, 0xdd, 0x6d, 0x66, 0x61, 0xdd, 0x04, 0x28, 0x63, 0x48, 0xd7, 0x98,
    178.     0x5b, 0xd3, 0x2b, 0xfb, 0xc9, 0x96, 0xa1, 0x34, 0xbe, 0xaa, 0x1a, 0xb3, 0x57, 0xab, 0x8d, 0xfd,
    179.     0xe3, 0x81, 0x7f, 0x3c, 0x77, 0x0f, 0x98, 0xcf, 0x84, 0xd6, 0x98, 0x35, 0xed, 0x41, 0xb6, 0xc5,
    180.     0x82, 0x2e, 0x70, 0x25, 0xdf, 0x84, 0x38, 0x8b, 0xa7, 0x02, 0x68, 0x85, 0xe9, 0x49, 0x1c, 0x8b,
    181.     0xe4, 0xe7, 0xc3, 0xf9, 0x78, 0x3a, 0x3c, 0x17, 0xe9, 0xee, 0x11, 0xb3, 0xd6, 0xaa, 0xc1, 0xd1,
    182.     0x1e, 0xb1, 0x31, 0x83, 0x8e, 0x8a, 0xab, 0xd8, 0xe3, 0x77, 0xe5, 0x9b, 0xcb, 0x88, 0xc2, 0x50,
    183.     0xd5, 0x82, 0xc3, 0x3d, 0xbc, 0xe7, 0x01, 0xed, 0xbf, 0x03, 0x00, 0x00, 0xff, 0xff, 0xd0, 0x4e,
    184.     0x12, 0xc0, 0x33, 0x01, 0x00, 0x00,
    185. }
    187. // Reference imports to suppress errors if they are not otherwise used.
    188. var _ context.Context
    189. var _ grpc.ClientConn
    191. // This is a compile-time assertion to ensure that this generated file
    192. // is compatible with the grpc package it is being compiled against.
    193. const _ = grpc.SupportPackageIsVersion4
    195. // TransClient is the client API for Trans service.
    196. //
    197. // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream.
    198. type TransClient interface {
    199.     //双向流 客户端发送状态，服务端发送指令
    200.     SendStatus(ctx context.Context, opts ...grpc.CallOption) (Trans_SendStatusClient, error)
    201. }
    203. type transClient struct {
    204.     cc *grpc.ClientConn
    205. }
    207. func NewTransClient(cc *grpc.ClientConn) TransClient {
    208.     return &transClient{cc}
    209. }
    211. func (c *transClient) SendStatus(ctx context.Context, opts ...grpc.CallOption) (Trans_SendStatusClient, error) {
    212.     stream, err := c.cc.NewStream(ctx, &_Trans_serviceDesc.Streams[0], "/proto.Trans/SendStatus", opts...)
    213.     if err != nil {
    214.         return nil, err
    215.     }
    216.     x := &transSendStatusClient{stream}
    217.     return x, nil
    218. }
    220. type Trans_SendStatusClient interface {
    221.     Send(*RunMessage) error
    222.     Recv() (*Order, error)
    223.     grpc.ClientStream
    224. }
    226. type transSendStatusClient struct {
    227.     grpc.ClientStream
    228. }
    230. func (x *transSendStatusClient) Send(m *RunMessage) error {
    231.     return x.ClientStream.SendMsg(m)
    232. }
    234. func (x *transSendStatusClient) Recv() (*Order, error) {
    235.     m := new(Order)
    236.     if err := x.ClientStream.RecvMsg(m); err != nil {
    237.         return nil, err
    238.     }
    239.     return m, nil
    240. }
    242. // TransServer is the server API for Trans service.
    243. type TransServer interface {
    244.     //双向流 客户端发送状态，服务端发送指令
    245.     SendStatus(Trans_SendStatusServer) error
    246. }
    248. // UnimplementedTransServer can be embedded to have forward compatible implementations.
    249. type UnimplementedTransServer struct {
    250. }
    252. func (*UnimplementedTransServer) SendStatus(srv Trans_SendStatusServer) error {
    253.     return status.Errorf(codes.Unimplemented, "method SendStatus not implemented")
    254. }
    256. func RegisterTransServer(s *grpc.Server, srv TransServer) {
    257.     s.RegisterService(&_Trans_serviceDesc, srv)
    258. }
    260. func _Trans_SendStatus_Handler(srv interface{}, stream grpc.ServerStream) error {
    261.     return srv.(TransServer).SendStatus(&transSendStatusServer{stream})
    262. }
    264. type Trans_SendStatusServer interface {
    265.     Send(*Order) error
    266.     Recv() (*RunMessage, error)
    267.     grpc.ServerStream
    268. }
    270. type transSendStatusServer struct {
    271.     grpc.ServerStream
    272. }
    274. func (x *transSendStatusServer) Send(m *Order) error {
    275.     return x.ServerStream.SendMsg(m)
    276. }
    278. func (x *transSendStatusServer) Recv() (*RunMessage, error) {
    279.     m := new(RunMessage)
    280.     if err := x.ServerStream.RecvMsg(m); err != nil {
    281.         return nil, err
    282.     }
    283.     return m, nil
    284. }
    286. var _Trans_serviceDesc = grpc.ServiceDesc{
    287.     ServiceName: "proto.Trans",
    288.     HandlerType: (*TransServer)(nil),
    289.     Methods:     []grpc.MethodDesc{},
    290.     Streams: []grpc.StreamDesc{
    291.         {
    292.             StreamName:    "SendStatus",
    293.             Handler:       _Trans_SendStatus_Handler,
    294.             ServerStreams: true,
    295.             ClientStreams: true,
    296.         },
    297.     },
    298.     Metadata: "message.proto",
    299. }

    3、服务端实现grpc的接口

    1. package service
    3. import (
    4.     "crawler/log"
    5.     "crawler/proto"
    6.     "io"
    7. )
    9. //实现grpc的接口
    10. type Service struct {
    11. }
    13. func (s *Service) SendStatus(stream proto.Trans_SendStatusServer) error {
    14.     ctx := stream.Context()
    15.     log.Debug("进入监听函数")
    16.     for {
    17.         log.Debug("循环监听请求")
    18.         select {
    19.         case <-ctx.Done():
    20.             log.Info("收到客户端的context发出终止信号")
    21.             return ctx.Err()
    22.         default:
    23.             //接受来自客户端发来的消息
    24.             in,err:=stream.Recv()
    25.             if err!=nil{
    26.                 if err==io.EOF{
    27.                     log.Debug("客户端发送的数据流结束,%s",err.Error())
    28.                 }else{
    29.                     log.Error("接受客户端信息出错,%s",err.Error())
    30.           //推出连接
    31.                     return err
    32.                 }
    33.             }else{
    34.                 log.Debug("来自客户端的信息,%s",in.GetId())
    35.                 stream.Send(&proto.Order{Status:proto.Status_END})
    37.             }
    40.         }
    41.     }
    43.     return nil
    44. }

    4、客户端实现

    1. package main
    3. import (
    4.     ctx "context"
    5.     "crawler/log"
    6.     "crawler/proto"
    7.     "google.golang.org/grpc"
    8.     "gopkg.in/urfave/cli.v2"
    9.     "os"
    10.     "time"
    11. )
    13. var client proto.TransClient
    14. var close = make(chan int)
    16. func main() {
    17.     //设置日志
    18.     log.SetConfig(log.DEBUG_L, 3, os.Stderr)
    19.     app := cli.App{
    20.         Name:    "爬虫客户端，针对金联创的数据",
    21.         Version: "1.0",
    22.         Usage:   "该客户端需要和管理端配合，由管理端调度和管理",
    23.         Authors: []*cli.Author{{Name: "lhn", Email: "550124023@qq.com"}},
    24.         Flags: []cli.Flag{
    25.             &cli.StringFlag{
    26.                 Name:    "server",
    27.                 Aliases: []string{"s"},
    28.                 Usage:   "管理端的地址，使用grpc通讯,例如：127.0.0.1:8555",
    29.                 EnvVars: []string{"SERVER", "S"},
    30.                 Value:   "127.0.0.1:8555",
    31.             },
    32.         },
    33.         Commands: []*cli.Command{
    34.             &cli.Command{
    35.                 Name: "send",
    36.                 Action: func(context *cli.Context) error {
    37.                     log.Info("输入参数：%s", context.Args().First())
    38.                     _ctx, cancel := ctx.WithTimeout(ctx.Background(), 10*time.Second)
    39.                     defer cancel()
    40.                     stream, err := client.SendStatus(_ctx)
    41.                     if err != nil {
    42.                         log.Error("%s", err.Error())
    43.                         return err
    44.                     }
    45.                     stream.Send(&proto.RunMessage{Status: proto.Status_END, Id: "fff"})
    46.                     return nil
    47.                 },
    48.             },
    49.         },
    50.         Action: func(context *cli.Context) error {
    51.             log.Debug("测试")
    52.             var opts = []grpc.DialOption{grpc.WithInsecure()}
    53.             log.Info("服务器地址:%s", context.String("server"))
    54.             conn, err := grpc.Dial(context.String("server"), opts...)
    55.             if err != nil {
    56.                 log.Fatal("连接服务器失败")
    57.             }
    58.             defer conn.Close()
    59.             client = proto.NewTransClient(conn)
    60.             _ctx := ctx.Background()//ctx.WithTimeout(ctx.Background(), 10*time.Hour)
    61.             //defer cancel()
    62.             stream, err := client.SendStatus(_ctx)
    63.             if err != nil {
    64.                 log.Error("发送消息错误%s", err.Error())
    65.                 return err
    66.             }
    67.             go func() {
    68.                 //必须要开一个runtime来接受消息，要处服务端发送过来没回应的话就会卡住，除非是客户端主动断开，消息才会被消化掉
    69.                 for {
    70.                     msg, err := stream.Recv()
    71.                     if err != nil {
    72.                         log.Error("接受发生错误,%s", err.Error())
    73.                     } else {
    74.                         log.Debug("接受到消息,%s", msg.Status)
    75.                     }
    76.                 }
    77.             }()
    78.             for {
    79.                 log.Debug("发送消息")
    80.                 //循环发送消息到服务端
    81.                 err = stream.Send(&proto.RunMessage{Status: proto.Status_END, Id: "fff"})
    82.                 if err != nil {
    83.                     log.Error("发送消息错误:%s", err.Error())
    84.                 }
    85.                 time.Sleep(1 * time.Millisecond)
    86.             }
    87.             return nil
    88.         },
    89.     }
    91.     app.Run(os.Args)
    93. }