gRPC 多端口订阅发布模式 C++ 示例：中心化服务管理消息

以下是一个使用 gRPC 实现多个 IP 端口的订阅发布模式的 C++ 示例程序。该示例程序使用一个中心化的服务来管理消息的订阅和发布。

首先，我们需要定义一个 .proto 文件，用于描述服务和消息的结构。

syntax = 'proto3';

package pubsub;

service PubSubService {
  rpc Subscribe(SubscriptionRequest) returns (stream Message);
  rpc Publish(Message) returns (PublishResponse);
}

message SubscriptionRequest {
  repeated string topics = 1;
}

message Message {
  string topic = 1;
  string content = 2;
}

message PublishResponse {
  bool success = 1;
}

接下来，我们使用 protoc 工具生成 C++ 的服务和消息的代码。

$ protoc -I=./ --cpp_out=./ ./pubsub.proto
$ protoc -I=./ --grpc_out=./ --plugin=protoc-gen-grpc=/usr/local/bin/grpc_cpp_plugin ./pubsub.proto

然后，我们可以实现服务的逻辑。以下是一个简单的服务实现示例。

#include <iostream>
#include <string>
#include <unordered_map>
#include <grpcpp/grpcpp.h>
#include "pubsub.grpc.pb.h"

using grpc::Server;
using grpc::ServerBuilder;
using grpc::ServerContext;
using grpc::Status;
using pubsub::PubSubService;
using pubsub::SubscriptionRequest;
using pubsub::Message;
using pubsub::PublishResponse;

class PubSubServiceImpl final : public PubSubService::Service {
public:
    Status Subscribe(ServerContext* context, const SubscriptionRequest* request, grpc::ServerWriter<Message>* writer) override {
        std::cout << "New subscription request" << std::endl;
        
        // Store the writer for later use
        std::string client_id = context->peer();
        writers_[client_id] = writer;
        
        // Send a welcome message to the client
        Message welcome_message;
        welcome_message.set_topic("Welcome");
        welcome_message.set_content("You are now subscribed");
        writer->Write(welcome_message);
        
        // Keep the connection alive
        while (!context->IsCancelled()) {
            std::this_thread::sleep_for(std::chrono::seconds(1));
        }
        
        // Remove the writer when the client disconnects
        writers_.erase(client_id);
        
        return Status::OK;
    }
    
    Status Publish(ServerContext* context, const Message* request, PublishResponse* response) override {
        std::cout << "New message: " << request->content() << std::endl;
        
        // Forward the message to all subscribers
        for (const auto& pair : writers_) {
            auto writer = pair.second;
            
            Message message;
            message.set_topic(request->topic());
            message.set_content(request->content());
            writer->Write(message);
        }
        
        response->set_success(true);
        return Status::OK;
    }
    
private:
    std::unordered_map<std::string, grpc::ServerWriter<Message>*> writers_;
};

void RunServer() {
    std::string server_address("0.0.0.0:50051");
    PubSubServiceImpl service;
    
    ServerBuilder builder;
    builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
    builder.RegisterService(&service);
    
    std::unique_ptr<Server> server(builder.BuildAndStart());
    std::cout << "Server listening on " << server_address << std::endl;
    
    server->Wait();
}

int main() {
    RunServer();
    return 0;
}

最后，我们可以编译并运行该示例程序。

$ g++ -std=c++11 -I=./ ./pubsub.pb.cc ./pubsub.grpc.pb.cc main.cpp -lgrpc++ -lgrpc -lprotobuf -pthread -o pubsub_server
$ ./pubsub_server

现在，我们已经实现了一个使用 gRPC 实现多个 IP 端口的订阅发布模式的 C++ 示例程序。该示例程序使用一个中心化的服务来管理消息的订阅和发布。