gRPC 发布订阅模式

尝试基于grpc和docker pubsub包，提供一个跨网络的发布和订阅系统

安装依赖： go get github.com/moby/moby/pkg/pubsub

首先通过proto定义一个发布和订阅服务接口：

syntax="proto3";
package pb;
option go_package="../pb";

message String{
  string value=1;
}

service PubsubService{
  rpc Publish (String) returns (String);
  rpc Subscribe (String) returns (stream String);
}

其中Publish是一个普通的grpc服务函数，Subscribe是一个单向流函数，接收一个我们自定义的String类型，返回的是一个String流。

下面来实现发布和订阅服务了

首先定义一个我们需要的发布订阅的结构体来实现发布订阅函数：

type PubsubService struct {
   pub *pubsub.Publisher
}

func NewPubsubService() *PubsubService {
   return &PubsubService{
      // 新建一个Publisher对象
      pub: pubsub.NewPublisher(100*time.Millisecond, 10),
   }
}

我们简单看看pubsub.Publisher这个结构体

其中有四个变量：读写锁、缓冲区大小、超时时间、为订阅者提供的函数过滤器，其中也提供了一些对应的方法，之后我们需要用到的时候再看

消息发布接口的实现：

// Publish 实现发布方法
func (p *PubsubService) Publish(ctx context.Context, arg *pb.String) (*pb.String, error) {
    // 发布消息
    p.pub.Publish(arg.GetValue())
    return &pb.String{}, nil
}

在此我们使用了pub.Publish()方法，该方法会将参数中的数据发送给当前订阅该发布者的所有订阅者

消息订阅接口的实现：

// Subscribe 实现订阅方法
func (p *PubsubService) Subscribe(arg *pb.String, stream pb.PubsubService_SubscribeServer) error {
    // SubscribeTopic 增加一个使用函数过滤器的订阅者
    // func(v interface{}) 定义函数过滤的规则
    // SubscribeTopic 返回一个chan interface{}
    ch := p.pub.SubscribeTopic(func(v interface{}) bool {
        // 接收数据是string，并且key是以arg为前缀的
        if key, ok := v.(string); ok {
            if strings.HasPrefix(key, arg.GetValue()) {
                return true
            }
        }
        return false
    })

    // 服务器遍历chan，并将其中信息发送给订阅客户端
    for v := range ch {
        if err := stream.Send(&pb.String{Value: v.(string)}); err != nil {
            return err
        }
    }
    return nil
}

我们模拟一个客户端，向服务器发布信息：

// 从客户端向服务器发布信息
func main() {
    conn, err := grpc.Dial("localhost:1234",
        grpc.WithTransportCredentials(insecure.NewCredentials()))
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()

    // 新建一个客户端
    client := pb.NewPubsubServiceClient(conn)

    // 客户端发布信息 golang :hello Go
    _, err = client.Publish(context.Background(), &pb.String{Value: "golang: hello Go"})
    if err != nil {
        log.Fatal(err)
    }

    // 客户端发布信息 docker: hello Docker
    _, err = client.Publish(context.Background(), &pb.String{Value: "docker: hello Docker"})
    if err != nil {
        log.Fatal(err)
    }
}

然后在另一个文件中，模仿订阅者去订阅这个服务，注意我们的函数过滤器规则，看看订阅者会收到什么信息？

func main() {
    conn, err := grpc.Dial("localhost:1234", grpc.WithTransportCredentials(insecure.NewCredentials()))
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()

    // 新建一个客户端
    client := pb.NewPubsubServiceClient(conn)
    
    // 订阅服务，传入参数是 golang: 
    // 会想过滤器函数，订阅者应该收到的信息为 golang: hello Go
    stream, err := client.Subscribe(context.Background(), &pb.String{Value: "golang: "})
    if err != nil {
        log.Fatal(err)
    }
    
    // 阻塞遍历流，输出结果
    for {
        reply, err := stream.Recv()
        if err != nil {
            if err == io.EOF {
                break
            }
            log.Fatal(err)
        }
        fmt.Println(reply.GetValue())
    }
}