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

Question

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

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

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

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

像这样的断言 assertEquals(client.getHello(user).getMessage(), "Hello John Doe")

可以，但我不知道，就我而言，我应该如何放置客户端部分......

GreetingService 节俭服务的代码:

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;
}

if I use this instead, then helloClient variable is clearly not used, but of course I need it to answer the method call

When a GreetingsNode X, connected to the management Node, makes an RPC of that method, another registered GreetingsNode must answer with the message "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;
}

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