java - NATS 持久消息 Java 客户端

Question

有人有使用 NATS Streaming Server 与 Java 客户端结合使用的经验吗？具体来说，我不知道如何使用 Java 客户端检索订阅者离线时发送的消息。

我可以使用 Go client 来查看我可以发布消息，然后添加订阅以检索所有已发布的消息。这是在 NATS Streaming Getting Started文档，它的工作原理如广告所示。

Publish several messages. For each publication you should get a result.

$ cd $GOPATH/src/github.com/nats-io/go-nats-streaming/examples
go run stan-pub.go foo "msg one"
Published [foo] : 'msg one'
$ go run stan-pub.go foo "msg two"
Published [foo] : 'msg two'
$ go run stan-pub.go foo "msg three"
Published [foo] : 'msg three'

Run the subscriber client. Use the --all flag to receive all published messages.

$ go run stan-sub.go --all -c test-cluster -id myID foo
Connected to nats://localhost:4222 clusterID: [test-cluster] clientID: [myID]
subscribing with DeliverAllAvailable
Listening on [foo], clientID=[myID], qgroup=[] durable=[]
[#1] Received on [foo]: 'sequence:1 subject:"foo" data:"msg one" timestamp:1465962202884478817 '
[#2] Received on [foo]: 'sequence:2 subject:"foo" data:"msg two" timestamp:1465962208545003897 '
[#3] Received on [foo]: 'sequence:3 subject:"foo" data:"msg three" timestamp:1465962215567601196

我正在尝试使用NATS Java client来做到这一点。我无法弄清楚这是否只是我没有找到类似的方法调用，或者 Java 客户端中是否不存在该功能。

这是我尝试过的

    import io.nats.client.Connection;
import io.nats.client.ConnectionFactory;
import io.nats.client.Constants;
import io.nats.client.Message;
import io.nats.client.SyncSubscription;

import java.io.IOException;
import java.security.SecureRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class NatsTest2 {

  private static final SecureRandom random = new SecureRandom();

  public static void main(String... args) throws Exception {
    final ConnectionFactory factory = new ConnectionFactory(Constants.DEFAULT_URL);
    try (final Connection conn = factory.createConnection()) {
      // Simple Async Subscriber
      final String expectMessage = "Yum, cookies " + System.currentTimeMillis();
      works(conn, expectMessage);
      broken(conn, expectMessage);
    }
  }

  private static void works(Connection conn, String expectMessage) throws IOException, TimeoutException {
    final String queue = Long.toString(random.nextLong());
    System.out.print(queue + "=>");
    try (final SyncSubscription subscription = conn.subscribeSync(queue)) {
      conn.publish(queue, expectMessage.getBytes());
      subscribe(subscription);
    }
  }

  private static void broken(Connection conn, String expectMessage) throws IOException, TimeoutException {
    final String queue = Long.toString(random.nextLong());
    System.out.print(queue + "=>");
    conn.publish(queue, expectMessage.getBytes());
    try (final SyncSubscription subscription = conn.subscribeSync(queue)) {
      subscribe(subscription);
    }
  }

  private static void subscribe(SyncSubscription subscription) throws IOException, TimeoutException {
    final Message message = subscription.nextMessage(1, TimeUnit.SECONDS);
    System.out.println(new String(message.getData()));
  }
}

这给出了输出

-8522002637987832314=>Yum, cookies 1473462495040
-3024385525006291780=>Exception in thread "main" java.util.concurrent.TimeoutException: Channel timed out waiting for items

Answer 1

如果您使用nats-streaming-server，则需要使用java-nats-streaming客户。您正在寻找的功能(订阅历史消息)仅存在于该客户端中。

无论如何，这就是您看到对 jnats 客户端所做的操作的原因:

nats-streaming-server 当前嵌入了 NATS 服务器 (gnatsd)，因此允许常规 NATS 客户端使用标准 NATS 功能，这就是您所看到的。

在您的示例代码中，works() 恰好可以工作，因为您的订阅在您发布消息之前就已创建(换句话说，您的 try-with-resources block 保证订阅在发生任何其他事情之前就已经处于 Activity 状态)。因此，您实际上并没有收到过去发布的消息；而是收到了一条过去发布的消息。您收到一条订阅开始后发布的消息。

broken() 示例不起作用，因为它实际上在创建订阅之前发布消息，并且该消息是由于没有兴趣(尚未)而被服务器丢弃。