Java中的Nats请求/回复

Question

我以前有卡夫卡的知识，我一直在玩Nats.io，这似乎是一个非常可靠的消息选择。

特别是，我对文档充分的Request/Reply机制很感兴趣，但我在使用Jnats驱动程序在Java中正确实现它时遇到了困难。

这是我的连接器：

// Single server nats connection
@PostConstruct
public void connect() throws ExternalServiceUnavailableException {
Options options = new Options.Builder()
.server(connectionString)
.maxReconnects(20)
.reconnectWait(Duration.ofSeconds(5))
.connectionTimeout(Duration.ofSeconds(5))
.connectionListener((conn, type) -> {
if (type == ConnectionListener.Events.CONNECTED) {
LOG.info("Connected to Nats Server");
} else if (type == ConnectionListener.Events.RECONNECTED) {
LOG.info("Reconnected to Nats Server");
} else if (type == ConnectionListener.Events.DISCONNECTED) {
LOG.error("Disconnected to Nats Server, reconnect attempt in seconds");
} else if (type == ConnectionListener.Events.CLOSED) {
LOG.info("Closed connection with Nats Server");
}
})
.build();

try {
connection = Nats.connect(options);
} catch (Exception e) {
LOG.error("Unable to connect to Nats Server");
throw new ExternalServiceUnavailableException(ExternalServiceUnavailableException.Service.NATS);
}
}

这是请求方法(出于测试目的，等待时间非常长(：

public Optional<String> asyncRequest(String topic, String message) throws ExternalServiceUnavailableException {
Future<Message> reply = natsConnector.getConnection().request(topic, message.getBytes());
try {
Message msg = reply.get(10L, TimeUnit.SECONDS);
LOG.info(new String(msg.getData()));
return Optional.of(new String(msg.getData(), StandardCharsets.UTF_8));
} catch (InterruptedException | ExecutionException | TimeoutException e) {
e.printStackTrace();
LOG.error("Unable to retrieve response for the sent request: " + message);
throw new ExternalServiceUnavailableException(ExternalServiceUnavailableException.Service.NATS);
}
}

这是具有回复机制的响应处理程序：

@PostConstruct
private void init() {
Dispatcher dispatcher = natsConnector.getConnection().createDispatcher(message -> {
});
Subscription assetsInfo = dispatcher.subscribe("assets-info", message -> {
JSONObject requestMessage = new JSONObject(new String(message.getData(), StandardCharsets.UTF_8));
if (requestMessage.getString("requestType").equals("stock-status")) {
if (requestMessage.getString("of").equals("all")) {
JSONObject response = assetQuery.retrieveYesterdayStockStatus();
LOG.info("response ready");
natsOperation.publishEvent("assets-info", response);
LOG.info("message sent");
}
}
});
}

我的两个独立服务通过一个码头化的Nats.io进行通信，我可以通过Nats Go客户端正确检查两个服务是否发送了关于同一主题的消息。

不幸的是，"请求者"在调用asyncRequest函数时，即使在reply.get(...)中有很高的等待，也不能很好地处理回复。

当我尝试在调试模式下评估reply对象时，它中没有任何数据，并显示一个TimeoutException。

在msg.getData()，程序崩溃。

你们有什么提示吗？谢谢

Answer 1

您应该更改"replyer"代码，以便从原始消息发布到replyTo主题。

@PostConstruct
private void init() {
Dispatcher dispatcher = natsConnector.getConnection().createDispatcher(message -> {
});
Subscription assetsInfo = dispatcher.subscribe("assets-info", message -> {
JSONObject requestMessage = new JSONObject(new String(message.getData(), StandardCharsets.UTF_8));
if (requestMessage.getString("requestType").equals("stock-status")) {
if (requestMessage.getString("of").equals("all")) {
JSONObject response = assetQuery.retrieveYesterdayStockStatus();
LOG.info("response ready");
//See Change Here
natsOperation.publish(message.getReplyTo(), response);
LOG.info("message sent");
}
}
});
}

请求-回复机制在生成的replyTo主题上寻找单个响应。

请参阅https://docs.nats.io/nats-concepts/reqreply