分布式环境中 Apache Thrift RPC 的 Java 方法实现

Question

我的项目的简要说明： 我正在编写一个名为“GreetingsNode”的 java 类，它在有一个“managementNode”的分布式环境中工作，它就像服务存储库一样，接收和存储其他节点和调度的信息（主机端口号和提供的服务）注册的服务提供的方法的 RPC。如果一个节点可以对 RPC 进行应答，则打开一个 thrift socket，并在调用节点和应答节点之间建立连接，应答节点返回结果。

我使用 Apache thrift 作为 IDL 和 RPC 框架。

现在的问题。 我的 GreetingsNodeHandler 类实现了一个简单的 thrift 接口，其中包含一个方法“getHello(user)”（user 是一个包含节点名称的结构，它是 GreetingsNode 类的构造函数的参数）。 当连接到管理节点的 GreetingsNode X 对该方法进行 RPC 时，另一个已注册的 GreetingsNode 必须以消息“hello X”进行响应。

我没有正确理解如何实现返回结果的处理程序部分，因此我不明白我应该如何编写应该检查方法实现是否正常工作的 junit 测试。

类似的断言 assertEquals(client.getHello(user).getMessage(), "Hello John Doe")

会工作，但我不明白，在我的情况下，我应该如何放置客户端部分......

GreetingService thrift 服务代码：

struct Message {
    1: string message   
}  

struct User {
    1: string name
}

service GreetingsService {
    Message getHello(1: User user)
}

必须实现 GreetingsService 方法 getHello() 的 GreetingsServiceHandler 代码

public class GreetingsServiceHandler implements GreetingsService.Iface {

private static Random random = new Random(10);
private ManagementService.Client managementClient;
private GreetingsService.Client helloClient;

@Override
public Message getHello(User user) throws TException {
    Message answer = null;
    // class ServiceProvider is generated by thrift, part of ManagementService thrift service 
    ServiceProvider provider = null; 
    List<ServiceProvider>providers = managementClient.getProvidersForService(user.name);

    if (providers.isEmpty())
        throw new NoProviderAvailableException(); //separate file contains Exception
    else {
        provider = providers.get(random.nextInt(providers.size()));
        //connection between nodes is established here
        TTransport helloTransport = new TSocket(provider.getHostName(), provider.getPort());
        TProtocol helloProtocol = new TBinaryProtocol(helloTransport);
        helloClient = new GreetingsService.Client(helloProtocol);
        helloTransport.open();

        // here lies my problem
        answer = helloClient.getHello(user);
        //if I use this instead, then helloClient variable is clearly not used, but of course I need it to answer the method call
        answer = answer.setMessage("Ciao " + user.getName() + ", welcome among us!");

    }
    return answer;

}

和 GreetingsNode 代码如下：

public class GreetingsNode implements NodeIface {

private ThriftServer helloServer;
private ManagementService.Client managementClient;
private NodeManifest nodeManifest;
private User user;
private String name;

public GreetingsNode(NodeManifest nodeManifest, String name) {
    this.nodeManifest = nodeManifest;
    this.helloServer = new ThriftServer(GreetingsServiceHandler.class);
    this.name = name;
}

@Override
public void turnOn() throws TException {

    helloServer.start();

    TSocket helloServerTransport = new TSocket("localhost", Constants.SERVER_PORT);
    TBinaryProtocol helloServerProtocol = new TBinaryProtocol(helloServerTransport);
    managementClient = new ManagementService.Client(helloServerProtocol);
    this.setUser(new User(name));
    helloServerTransport.open();

    helloServer = new ThriftServer(GreetingsServiceHandler.class);
    //portNegotiator is a class described in a separate file, that handles the registration of other nodes to the managementNode. NodeManifest is a file generated by thrift, part of managementService thrift file, describing a struct that contains hostname and port number of nodes.
    PortNegotiator negotiator = new PortNegotiator(managementClient);
    negotiator.negotiate(nodeManifest, helloServer);

}

@Override
public void turnOff() {
    helloServer.stop();
}

public User getUser() {
    return user;
}

public void setUser(User user) {
    this.user = user;
}

Answer 1

处理程序中的基本方法 impl 非常简单，应该执行以下操作（免责声明：未测试）：

@Override
public Message getHello(User user) throws TException {
    Message answer = new Message();
    answer = answer.setMessage("Ciao " + user.getName() + ", welcome among us!");
    return answer;
}

如果我改用这个，那么显然没有使用 helloClient 变量，但是我当然需要它来回答方法调用

当连接到管理节点的 GreetingsNode X 对该方法进行 RPC 时，另一个已注册的 GreetingsNode 必须以消息“hello X”进行响应。

如果这意味着我们想要一个像 Client => ServerA => Server B 这样的调用序列，那么这也是可能的，只需要稍作修改。从我们上面的基本示例开始，我们相应地增强了代码：

private Message callTheOtherNode(User user) {
  // class ServiceProvider is generated by Thrift, 
  // part of ManagementService Thrift service 
  ServiceProvider provider = null; 
  List<ServiceProvider>providers = managementClient.getProvidersForService(user.name);

  if (providers.isEmpty())
      throw new NoProviderAvailableException(); //separate file contains Exception

  provider = providers.get(random.nextInt(providers.size()));
  //connection between nodes is established here
  TTransport helloTransport = new TSocket(provider.getHostName(), provider.getPort());
  TProtocol helloProtocol = new TBinaryProtocol(helloTransport);
  helloClient = new GreetingsService.Client(helloProtocol);
  helloTransport.open();
  return helloClient.getHello(user);
}

@Override
public Message getHello(User user) throws TException {
    Message answer = callTheOtherNode(user);
    return answer;
}

当然，被调用的“其他节点”需要对请求进行实际处理，而不是简单地将其再次转发到另一个节点。