gRPC 示例：实现发布订阅模式

func NewGreeterClient(cc *grpc.ClientConn) GreeterClient {
	return &greeterClient{cc}
}

func RegisterGreeterServer(s *grpc.Server, srv GreeterServer) {
	s.RegisterService(&_Greeter_serviceDesc, srv)
}

package main
import (
	"context"
	"log"
	"net"
	"google.golang.org/grpc"
	"../pb"
)
const port = ":50051"

type server struct { // 服务端接口实现，这里直接内嵌 protoc 默认生成的代码，并进行了代码重写（override）
	pb.UnimplementedGreeterServer
}
// 服务端方法实现，覆盖了 pb.UnimplementedGreeterServer 中定义的 SayHello 方法
func (s *server) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {
	log.Printf("Received: %v", in.GetName())
	return &pb.HelloReply{Message: "Hello " + in.GetName()}, nil
}

func main() {
	// 新建一个端口监听 listen
	listen, err := net.Listen("tcp", port)
	if err != nil {
		log.Fatalf("failed to listen: %v", err)
	}
	// 新建一个 gRPC server，初始 gRPC server 是无状态的
	gRPCServer := grpc.NewServer()
	// 调用 protoc 生成的 RegisterGreeterServer 方法，将我们对于 server 的实现，注册到 gRPC server 中去
	pb.RegisterGreeterServer(gRPCServer, &server{})
	// 启动 gRPC server
	if err := gRPCServer.Serve(listen); err != nil {
		log.Fatalf("failed to serve: %v", err)
	}
}

package main
import (
	"context"
	"log"
	"os"
	"time"
	"google.golang.org/grpc"
	"../pb"
)

const (
	address     = "localhost:50051"
	defaultName = "world"
)

func main() {
	// 建立一个与 gRPC 服务端的 tcp 连接
	conn, err := grpc.Dial(address, grpc.WithInsecure(), grpc.WithBlock())
	if err != nil {
		log.Fatalf("did not connect: %v", err)
	}
	defer conn.Close() 
	// 调用 protoc 生成的 NewGreeterClient 方法，传入 conn，新建一个 gRPC client
	c := pb.NewGreeterClient(conn)

	// Contact the server and print out its response.
	name := defaultName
	if len(os.Args) > 1 {
		name = os.Args[1]
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	// 调用 SayHello 方法
	r, err := c.SayHello(ctx, &pb.HelloRequest{Name: name})
	if err != nil {
		log.Fatalf("could not greet: %v", err)
	}
	log.Printf("Greeting: %s", r.GetMessage())
}

type PubsubServiceClient interface {
	Publish(ctx context.Context, in *StrMsg, opts ...grpc.CallOption) (*StrMsg, error)
	Subscribe(ctx context.Context, in *StrMsg, opts ...grpc.CallOption) (PubsubService_SubscribeClient, error)
}

type pubsubServiceClient struct {
	cc grpc.ClientConnInterface
}

func NewPubsubServiceClient(cc grpc.ClientConnInterface) PubsubServiceClient {
	return &pubsubServiceClient{cc}
}
// 默认生成的 Publish 方法
func (c *pubsubServiceClient) Publish(ctx context.Context, in *StrMsg, opts ...grpc.CallOption) (*StrMsg, error) {
	out := new(StrMsg)
	err := c.cc.Invoke(ctx, "/build_in_grpc.PubsubService/Publish", in, out, opts...)
	if err != nil {
		return nil, err
	}
	return out, nil
}
// 默认生成的 Subscribe 方法
func (c *pubsubServiceClient) Subscribe(ctx context.Context, in *StrMsg, opts ...grpc.CallOption) (PubsubService_SubscribeClient, error) {
	stream, err := c.cc.NewStream(ctx, &_PubsubService_serviceDesc.Streams[0], "/build_in_grpc.PubsubService/Subscribe", opts...)
	if err != nil {
		return nil, err
	}
	x := &pubsubServiceSubscribeClient{stream}
	if err := x.ClientStream.SendMsg(in); err != nil {
		return nil, err
	}
	if err := x.ClientStream.CloseSend(); err != nil {
		return nil, err
	}
	return x, nil
}

type PubsubService_SubscribeClient interface {
	Recv() (*StrMsg, error)
	grpc.ClientStream
}

type PubsubServiceServer interface {
	Publish(context.Context, *StrMsg) (*StrMsg, error)
	Subscribe(*StrMsg, PubsubService_SubscribeServer) error
}
func RegisterPubsubServiceServer(s *grpc.Server, srv PubsubServiceServer) {
	s.RegisterService(&_PubsubService_serviceDesc, srv)
}
type PubsubService_SubscribeServer interface {
	Send(*StrMsg) error
	grpc.ServerStream
}

package main

import (
	"context"
	"github.com/atmosphere/grpc/build_in_grpc"
	"github.com/moby/moby/pkg/pubsub"
	"google.golang.org/grpc"
	"log"
	"net"
	"strings"
	"time"
)
// gRPC server 的实现
type PubSubService struct {
	pub *pubsub.Publisher
}
func NewPubsubService() *PubSubService {
	return &PubSubService{
		pub: pubsub.NewPublisher(100* time.Millisecond, 10),
	}
}
// Publish 方法实现，发布者通过 Publish 发布消息
func (p *PubSubService) Publish(ctx context.Context, arg *build_in_grpc.StrMsg) (*build_in_grpc.StrMsg, error){
	// 直接调用 pub 库发布消息
	p.pub.Publish(arg.GetValue())
	return &build_in_grpc.StrMsg{}, nil
}
// Subscribe 具体实现，此方法是供订阅者调用的
func (p *PubSubService) Subscribe(arg *build_in_grpc.StrMsg, steam build_in_grpc.PubsubService_SubscribeServer) error {
	// 调用 Moby 的库添加一个订阅者，并构造了一个过滤器
	ch := p.pub.SubscribeTopic(func(v interface{}) bool {
		if key, ok := v.(string); ok {
			if strings.HasPrefix(key, arg.GetValue()) {
				return true
			}
		}
		return false
	})
	for v := range ch { // 阻塞在channel上，有消息就返回给客户端
		log.Println("send a message to client")
		if err := steam.Send(&build_in_grpc.StrMsg{Value: v.(string)}); err != nil {
			return err
		}
	}
	return nil
}

func main()  {
	// 新建一个 gRPC server
	grpcServer := grpc.NewServer()
	// 注册 server
	build_in_grpc.RegisterPubsubServiceServer(grpcServer, NewPubsubService())
	lis, err := net.Listen("tcp", ":1234")
	if err != nil {
		log.Fatal(err)
	}
	// 启动 server
	grpcServer.Serve(lis)
}

func main()  {
	// 新建与 server 的 tcp 连接
	conn, err := grpc.Dial("localhost:1234", grpc.WithInsecure())
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	// 通过 client 发起订阅
	client := build_in_grpc.NewPubsubServiceClient(conn)
	stream, err := client.Subscribe(context.Background(), &build_in_grpc.StrMsg{
		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())
	}
}