Stream data persistence Pub/Sub using NATS Java client

NATS Server itself does not support persistence; in this example, NATS JetStream is used

[TOC]

producers

:one: creates a connection

This part is no different from the native NATS

//DurablePub.java
Options options = new Options.Builder()
    .server("nats://localhost:4222")
    .reconnectWait(Duration.ofSeconds(1))
    .build();
Connection nc = Nats.connect(options);
Copy the code

:two: configure and create flows

NATS Server: NATS Server: NATS Server

//DurablePub.java
private static final String STREAM = "CONTRACT";
private static final String SUBJECTS = "CONTRACT.*";
Copy the code

//DurablePub.java
//Configure and create Stream
StreamConfiguration streamConfiguration = StreamConfiguration.builder()
    .name(STREAM)
    .subjects(SUBJECTS)
    .retentionPolicy(RetentionPolicy.WorkQueue)
    .replicas(1)
    .discardPolicy(DiscardPolicy.Old)
    .duplicateWindow(Duration.ofSeconds(30))
    .build();

StreamInfo streamInfo = nc.jetStreamManagement().addStream(streamConfiguration);
Copy the code

The attributes I configured here are explained as follows:

• name: The name of the flow
• subjectsSubjects in flow
• retentionPolicy: Retention policy for messages. The default value isLimitsPolicyThat is, the retention of messages is determined by various constraints
• replicas: Number of copies of the message in the cluster. The maximum value is 5
• discardPolicy: Discard policy. The default value isDiscardOldWhen the message store reaches its limit, the old message is deleted
• duplicateWindow: You are advised to keep the time window for message deduplication as small as possible

The full attributes are shown in the following table (quoted from the official website):

Retention Strategy

The MaxMsgs, MaxMsgs, and MaxAge mentioned above are used to restrict messages and are also attributes that can only be used by the LimitsPolicy policy.

Under the WorkQueuePolicy, messages are deleted as soon as a consumer receives an acknowledgement. InterestPolicy, on the other hand, deletes stories immediately if no consumers are online.

Note that under WorkQueuePolicy and InterestPolicy, the MaxMsgs, MaxMsgs, and MaxAge attributes are still valid and exist as preconditions.

Acknowledgement Models

There are three confirmation modes for consumers:

:three: sends the message

Once the flow is configured and created, messages can be sent

//Pub
for (int i = 0; i < 10; i++) {
  nc.jetStream().publish("CONTRACT.EFFECT", generateData().getBytes(StandardCharsets.UTF_8));
}
Copy the code

//Generate Data
private static String generateData(a) {
  Contract contract = new Contract(UUID.randomUUID().toString(), "EFFECT".new Date());
  GsonBuilder builder = new GsonBuilder();
  Gson gson = builder.create();
  return gson.toJson(contract);
}

 private static class Contract {

    public String contractId;

    public String status;

    public Date signDate;

    public Contract(String contractId, String status, Date signDate) {
      this.contractId = contractId;
      this.status = status;
      this.signDate = signDate; }}Copy the code

:four: Closes the connection

nc.flush(Duration.ZERO);
nc.close();
Copy the code

consumers

The connection between the consumer and the producer will be skipped

:one: Configure and create consumer:

private static final String STREAM = "CONTRACT";
private static final String SUBJECTS = "CONTRACT.EFFECT";
private static final String CONSUMER = "consumer-1";
Copy the code

//Configure and create consumer
ConsumerConfiguration configuration = ConsumerConfiguration.builder()
    //Durable Consumer Name
    .durable(CONSUMER)
    .filterSubject(SUBJECTS)
    .replayPolicy(ReplayPolicy.Instant)
    //This requires every message to be specifically acknowledged, it's the only supported option for pull-based Consumers
    .ackPolicy(AckPolicy.Explicit)
    .ackWait(Duration.ofSeconds(30))
    .deliverPolicy(DeliverPolicy.All)
    .maxDeliver(20)
    .rateLimit(100)
    .maxAckPending(20000)
    .build();

ConsumerInfo consumerInfo = nc.jetStreamManagement()
    .addOrUpdateConsumer(STREAM, configuration);
Copy the code

:two: creates a pull configuration (PushSubscribeOptions, which is similar)

PullSubscribeOptions pullSubscribeOptions = PullSubscribeOptions
          .builder()
          .configuration(configuration)
          .build();
Copy the code

:three: Creates the subscription, reads it in batches, prints the message content, and confirms it manually

JetStreamSubscription jetStreamSubscription = js.subscribe(SUBJECTS, pullSubscribeOptions);

Iterator<Message> iter = jetStreamSubscription.iterate(10, Duration.ofMillis(1000));
while (iter.hasNext()) {
  Message message = iter.next();
  System.out.printf("Message Received : %s\n".new String(message.getData(), StandardCharsets.UTF_8));
  message.ack();
}
Copy the code

The list of consumer configuration properties is as follows:

Consumer Starting Position

When configuring consumers to decide where to start spending, NATS supports the following deliverpolicies:

conclusion

NATS is much easier to use than message-oriented middleware such as RabbitMQ and Kafka, but the Java client doesn’t feel very user-friendly and the official Spring client has been shelved, which is a bit of a hassle and requires some extra work to simplify use.