在vertx中跨顶点共享单例客户端的正确方法

Posted 2023-02-19

【中文标题】在vertx中跨顶点共享单例客户端的正确方法

【英文标题】：Correct way of sharing singleton clients across verticles in vetx

【发布时间】：2021-12-21 22:38:01

【问题描述】：

我有一个 vertx 应用程序，我在其中部署了多个 verticle A 实例 (HttpVerticle.java) 和多个 verticle B 实例 (AerospikeVerticle.java)。 Aerospike verticles 需要共享一个 AerospikeClient。 HttpVerticle 监听端口 8888 并使用事件总线调用 AerospikeVerticle。我的问题是：

使用 sharedData 是共享单例客户端实例的正确方法吗？还有其他推荐/更清洁的方法吗？我计划在应用程序中创建和共享更多这样的单例对象（cosmos db 客户端、meterRegistry 等）。我计划使用 sharedData.localMap 以类似的方式共享它们。 是否可以使用 vertx 的事件循环作为 aerospike 客户端的支持事件循环？这样 aerospike 客户端初始化不需要创建自己的新事件循环？目前看起来 aerospike get 调用的 onRecord 部分在 aerospike 的事件循环上运行。

public class SharedAerospikeClient implements Shareable 
  public final EventLoops aerospikeEventLoops;
  public final AerospikeClient client;

  public SharedAerospikeClient() 
    EventPolicy eventPolicy = new EventPolicy();
    aerospikeEventLoops = new NioEventLoops(eventPolicy,  2 * Runtime.getRuntime().availableProcessors());
    ClientPolicy clientPolicy = new ClientPolicy();
    clientPolicy.eventLoops = aerospikeEventLoops;
    client = new AerospikeClient(clientPolicy, "localhost", 3000);
  

Main.java

public class Main 
  public static void main(String[] args) 
    Vertx vertx = Vertx.vertx();
    LocalMap localMap = vertx.sharedData().getLocalMap("SHARED_OBJECTS");
    localMap.put("AEROSPIKE_CLIENT", new SharedAerospikeClient());
    vertx.deployVerticle("com.demo.HttpVerticle", new DeploymentOptions().setInstances(2 * 4));
    vertx.deployVerticle("com.demo.AerospikeVerticle", new DeploymentOptions().setInstances(2 * 4));
  

HttpVerticle.java

public class HttpVerticle extends AbstractVerticle 

  @Override
  public void start(Promise<Void> startPromise) throws Exception 
    vertx.createHttpServer().requestHandler(req -> 
      vertx.eventBus().request("read.aerospike", req.getParam("id"), ar -> 
        req.response()
          .putHeader("content-type", "text/plain")
          .end(ar.result().body().toString());
        System.out.println(Thread.currentThread().getName());
      );
    ).listen(8888, http -> 
      if (http.succeeded()) 
        startPromise.complete();
        System.out.println("HTTP server started on port 8888");
       else 
        startPromise.fail(http.cause());
      
    );
  

AerospikeVerticle.java

public class AerospikeVerticle extends AbstractVerticle 
  private SharedAerospikeClient sharedAerospikeClient;

  @Override
  public void start(Promise<Void> startPromise) throws Exception 
    EventBus eventBus = vertx.eventBus();
    sharedAerospikeClient = (SharedAerospikeClient) vertx.sharedData().getLocalMap("SHARED_OBJECTS").get("AEROSPIKE_CLIENT");
    MessageConsumer<String> consumer = eventBus.consumer("read.aerospike");
    consumer.handler(this::getRecord);
    System.out.println("Started aerospike verticle");
    startPromise.complete();
  

  public void getRecord(Message<String> message) 
    sharedAerospikeClient.client.get(
      sharedAerospikeClient.aerospikeEventLoops.next(),
      new RecordListener() 
        @Override
        public void onSuccess(Key key, Record record) 
            if (record != null) 
              String result = record.getString("value");
              message.reply(result);
             else 
              message.reply("not-found");
            
        

        @Override
        public void onFailure(AerospikeException exception) 
            message.reply("error");
        
      ,
      sharedAerospikeClient.client.queryPolicyDefault,
      new Key("myNamespace", "mySet", message.body())
    );
  

【参考方案1】：

我不知道 Aerospike 客户端。

关于verticles之间的共享对象，实际上共享数据映射就是为此目的而设计的。

但是，这样做更容易：

在您的主类或自定义启动器中创建共享客户端 提供客户端作为verticle构造函数的参数

Vertx 接口有一个deployVerticle(Supplier<Verticle>, DeploymentOptions) 方法，在这种情况下很方便：

MySharedClient client = initSharedClient();
vertx.deploy(() -> new SomeVerticle(client), deploymentOptions);

【讨论】：

谢谢，这似乎更方便了。