NATS Concepts

Overview

What is NATS?
NATS is a connective technology that powers modern distributed systems. A connective technology is responsible for addressing, discovery and exchanging of messages that drive the common patterns in distributed systems; asking and answering questions, aka services/microservices, and making and processing statements, or stream processing.

Challenges faced by modern distributed systems
Modern distributed systems are defined by an ever increasing number of hyper-connected moving parts and the additional data they generate. They employ both services and streams to drive business value. They are also being defined by location independence and mobility, and not just for things we would typically recognize as front end technologies. Today’s systems and the backend processes, microservices and stream processing are being asked to be location independent and mobile as well, all while being secure.

These modern systems present challenges to technologies that have been used to connect mobile front ends to fairly static backends. These incumbent technologies typically manage addressing and discovery via hostname (DNS) or IP and port, utilize a 1:1 communication pattern, and have multiple different security patterns for authentication and authorization. Although not perfect, incumbent technologies have been good enough in many situations, but times are changing quickly. As microservices, functions, and stream processing are being asked to move to the edge, these technologies and the assumptions they make are being challenged.

What makes the NATS connective technology unique for these modern systems?

Effortless M:N connectivity: NATS manages addressing and discovery based on subjects and not hostname and ports. Defaulting to M:N communications, which is a superset of 1:1, meaning it can do 1:1 but can also do so much more. If you have a 1:1 system that is successful in development, ask how many other moving parts are required for production to work around the assumption of 1:1? Things like load balancers, log systems, and network security models, as well as proxies and sidecars. If your production system requires all of these things just to get around the fact that the connective technology being used, e.g. HTTP or gRPC, is 1:1, it’s time to give NATS.io a look.

Deploy anywhere: NATS can be deployed nearly anywhere; on bare metal, in a VM, as a container, inside K8S, on a device, or whichever environment you choose. NATS runs well within deployment frameworks or without.

Secure: Similarly, NATS is secure by default and makes no requirements on network perimeter security models. When you start considering mobilizing your backend microservices and stream processors, many times the biggest roadblock is security.

Scalable, Future-Proof Deployments

NATS infrastructure and clients communicate all topology changes in real-time. This means that NATS clients do not need to change when NATS deployments change. Having to change clients with deployments would be like having to reboot your phone every time your cell provider added or changed a cell tower. This sounds ridiculous of course, but think about how many systems today have their front ends tied so closely to the backend, that any change requires a complete front end reboot or at least a reconfiguration. NATS clients and applications need no such change when backend servers are added and removed and changed. Even DNS is only used to bootstrap first contact, after that, NATS handles endpoint locations transparently.

Hybrid Deployments

Another advantage to utilizing a NATS is that it allows a hybrid mix of SaaS/Utility computing with separately owned and operated systems. Meaning you can have a shared NATS service with core microservices, streams and stream processing be extended by groups or individuals who have a need to run their own NATS infrastructure. You are not forced to choose one or the other.

Adaptability

Today’s systems will fall short with new demands. As modern systems continue to evolve and utilize more components and process more data, supporting patterns beyond 1:1 communications, with addressing and discovery tied to DNS is critical. Foundational technologies like NATS promise the most return on investment. Incumbent technologies will not work as modern systems unify cloud, Edge, IoT and beyond. NATS does.

Use Cases

NATS can run anywhere, from large servers and cloud instances, through edge gateways and even IoT devices. Use cases for NATS include:

Compare NATS

This feature comparison is a summary of a few of the major components in several of the popular messaging technologies of today. This is by no means an exhaustive list and each technology should be investigated thoroughly to decide which will work best for your implementation.

In this comparison, we will be featuring NATS, Apache Kafka, RabbitMQ, Apache Pulsar, and gRPC.

Language and Platform Coverage

Project Client Languages and Platforms
NATS Core NATS: 48 known client types, 11 supported by maintainers, 18 contributed by the community. NATS Streaming: 7 client types supported by maintainers, 4 contributed by the community. NATS servers can be compiled on architectures supported by Golang. NATS provides binary distributions.
gRPC 13 client languages.
Kafka 18 client types supported across the community and by Confluent. Kafka servers can run on platforms supporting java; very wide support.
Pulsar 7 client languages, 5 third-party clients - tested on macOS and Linux.
Rabbit At least 10 client platforms that are maintainer-supported with over 50 community supported client types. Servers are supported on the following platforms: Linux, Windows NT.

Built-in Patterns

Project Supported Patterns
NATS Streams and Services through built-in publish/subscribe, request-reply, and load-balanced queue subscriber patterns. Dynamic request permissioning and request subject obfuscation is supported.
gRPC One service, which may have streaming semantics, per channel. Load Balancing for a service can be done either client-side or by using a proxy.
Kafka Streams through publish/subscribe. Load balancing can be achieved with consumer groups. Application code must correlate requests with replies over multiple topics for a service (request-reply) pattern.
Pulsar Streams through publish/subscribe. Multiple competing consumer patterns support load balancing. Application code must correlate requests with replies over multiple topics for a service (request-reply) pattern.
Rabbit Streams through publish/subscribe, and services with a direct reply-to feature. Load balancing can be achieved with a Work Queue. Applications must correlate requests with replies over multiple topics for a service (request-reply) pattern.

Delivery Guarantees

Project Quality of Service / Guarantees
NATS At most once, at least once, and exactly once is available in JetStream.
gRPC At most once.
Kafka At most once, at least once, and exactly once.
Pulsar At most once, at least once, and exactly once.
Rabbit At most once, at least once.

Multi-tenancy and Sharing

Project Multi-tenancy Support
NATS NATS supports true multi-tenancy and decentralized security through accounts and defining shared streams and services.
gRPC N/A
Kafka Multi-tenancy is not supported.
Pulsar Multi-tenancy is implemented through tenants; built-in data sharing across tenants is not supported. Each tenant can have its own authentication and authorization scheme.
Rabbit Multi-tenancy is supported with vhosts; data sharing is not supported.

AuthN

Project Authentication
NATS NATS supports TLS, NATS credentials, NKEYS (NATS ED25519 keys), username and password, or simple token.
gRPC TLS, ALT, Token, channel and call credentials, and a plug-in mechanism.
Kafka Supports Kerberos and TLS. Supports JAAS and an out-of-box authorizer implementation that uses ZooKeeper to store connection and subject.
Pulsar TLS Authentication, Athenz, Kerberos, JSON Web Token Authentication.
Rabbit TLS, SASL, username and password, and pluggable authorization.

AuthZ

Project Authorization
NATS Account limits including number of connections, message size, number of imports and exports. User-level publish and subscribe permissions, connection restrictions, CIDR address restrictions, and time of day restrictions.
gRPC Users can configure call credentials to authorize fine-grained individual calls on a service.
Kafka Supports JAAS, ACLs for a rich set of Kafka resources including topics, clusters, groups, and others.
Pulsar Permissions may be granted to specific roles for lists of operations such as produce and consume.
Rabbit ACLs dictate permissions for configure, write, and read operations on resources like exchanges, queues, transactions, and others. Authentication is pluggable.

Message Retention and Persistence

Project Message Retention and Persistence Support
NATS Supports memory and file persistence. Messages can be replayed by time, count, or sequence number, and durable subscriptions are supported. With NATS streaming, scripts can archive old log segments to cold storage.
gRPC N/A
Kafka Supports file-based persistence. Messages can be replayed by specifying an offset, and durable subscriptions are supported. Log compaction is supported as well as KSQL.
Pulsar Supports tiered storage including file, Amazon S3 or Google Cloud Storage (GCS). Pulsar can replay messages from a specific position and supports durable subscriptions. Pulsar SQL and topic compaction is supported, as well as Pulsar functions.
Rabbit Supports file-based persistence. Rabbit supported queue-based semantics (vs log), so no message replay is available.

High Availability and Fault Tolerance

Project HA and FT Support
NATS Core NATS supports full mesh clustering with self-healing features to provide high availability to clients. NATS streaming has warm failover backup servers with two modes (FT and full clustering). JetStream supports horizontal scalability with built-in mirroring.
gRPC N/A. gRPC relies on external resources for HA/FT.
Kafka Fully replicated cluster members are coordinated via Zookeeper.
Pulsar Pulsar supports clustered brokers with geo-replication.
Rabbit Clustering Support with full data replication via federation plugins. Clusters require low-latency networks where network partitions are rare.

Deployment

Project Supported Deployment Models
NATS The NATS network element (server) is a small static binary that can be deployed anywhere from large instances in the cloud to resource constrained devices like a Raspberry PI. NATS supports the Adaptive Edge architecture which allows for large, flexible deployments. Single servers, leaf nodes, clusters, and superclusters (cluster of clusters) can be combined in any fashion for an extremely flexible deployment amenable to cloud, on-premise, edge and IoT. Clients are unaware of topology and can connect to any NATS server in a deployment.
gRPC gRPC is point to point and does not have a server or broker to deploy or manage, but always requires additional pieces for production deployments.
Kafka Kafka supports clustering with mirroring to loosely coupled remote clusters. Clients are tied to partitions defined within clusters. Kafka servers require a JVM, eight cores, 64 GB to128 GB of RAM, two or more 8-TB SAS/SSD disks, and a 10-Gig NIC. (1)__
Pulsar Pulsar supports clustering and built-in geo-replication between clusters. Clients may connect to any cluster with an appropriately configured tenant and namespace. Pulsar requires a JVM and requires at least 6 Linux machines or VMs. 3 running ZooKeeper. 3 running a Pulsar broker and a BookKeeper bookie. (2)__
Rabbit Rabbit supports clusters and cross-cluster message propagation through a federation plugin. Clients are unaware of topology and may connect to any cluster. The server requires the Erlang VM and dependencies.

Monitoring

Project Monitoring Tooling
NATS NATS supports exporting monitoring data to Prometheus and has Grafana dashboards to monitor and configure alerts. There are also development monitoring tools such as nats-top. Robust side car deployment or a simple connect-and-view model with NATS surveyor is supported.
gRPC External components such as a service mesh are required to monitor gRPC.
Kafka Kafka has a number of management tools and consoles including Confluent Control Center, Kafka, Kafka Web Console, Kafka Offset Monitor.
Pulsar CLI tools, per-topic dashboards, and third-party tools.
Rabbit CLI tools, a plugin-based management system with dashboards and third-party tools.

Management

Project Management Tooling
NATS NATS separates operations from security. User and Account management in a deployment may be decentralized and managed through a CLI. Server (network element) configuration is separated from security with a command line and configuration file which can be reloaded with changes at runtime.
gRPC External components such as a service mesh are required to manage gRPC.
Kafka Kafka has a number of management tools and consoles including Confluent Control Center, Kafka, Kafka Web Console, Kafka Offset Monitor.
Pulsar CLI tools, per-topic dashboards, and third-party tools.
Rabbit CLI tools, a plugin-based management system with dashboards and third-party tools.

Integrations

Project Built-in and Third Party Integrations
NATS NATS supports WebSockets, a Kafka bridge, an IBM MQ Bridge, a Redis Connector, Apache Spark, Apache Flink, CoreOS, Elastic, Elasticsearch, Prometheus, Telegraf, Logrus, Fluent Bit, Fluentd, OpenFAAS, HTTP, and MQTT, and more.
gRPC There are a number of third party integrations including HTTP, JSON, Prometheus, Grift and others. (3)__
Kafka Kafka has a large number of integrations in its ecosystem, including stream processing (Storm, Samza, Flink), Hadoop, database (JDBC, Oracle Golden Gate), Search and Query (ElasticSearch, Hive), and a variety of logging and other integrations.
Pulsar Pulsar has many integrations, including ActiveMQ, Cassandra, Debezium, Flume, Elasticsearch, Kafka, Redis, and others.
Rabbit RabbitMQ has many plugins, including protocols (MQTT, STOMP), WebSockets, and various authorization and authentication plugins.

References

  1. https://docs.cloudera.com/HDPDocuments/HDF3/HDF-3.1.0/bk_planning-your-deployment/content/ch_hardware-sizing.html
  2. https://pulsar.apache.org/docs/4.0.x/deploy-bare-metal/
  3. https://github.com/grpc-ecosystem

What is NATS

Software applications and services need to exchange data. NATS is an infrastructure that allows such data exchange, segmented in the form of messages. We call this a "message oriented middleware".

With NATS, application developers can:

NATS Client Applications

Developers use one of the NATS client libraries in their application code to allow them to publish, subscribe, request and reply between instances of the application or between completely separate applications. Those applications are generally referred to as 'client applications' or sometimes just as 'clients' throughout this manual (since from the point of view of the NATS server, they are clients).

NATS Service Infrastructure

The NATS services are provided by one or more NATS server processes that are configured to interconnect with each other and provide a NATS service infrastructure. The NATS service infrastructure can scale from a single NATS server process running on an end device (the nats-server process is less than 20 MB in size!) all the way to a public global super-cluster of many clusters spanning all major cloud providers and all regions of the world such as Synadia's NGS.

Connecting NATS Client applications to the NATS servers

To connect a NATS client application with a NATS service, and then subscribe or publish messages to subjects, it only needs to be configured with:

  1. URL: A 'NATS URL'. This is a string (in a URL format) that specifies the IP address and port where the NATS server(s) can be reached, and what kind of connection to establish (plain TCP, TLS, or Websocket).
  2. Authentication (if needed): Authentication details for the application to identify itself with the NATS server(s). NATS supports multiple authentication schemes (username/password, decentralized JWT, token, TLS certificates and Nkey with challenge).

Simple messaging design

NATS makes it easy for applications to communicate by sending and receiving messages. These messages are addressed and identified by subject strings, and do not depend on network location.

Data is encoded and framed as a message and sent by a publisher. The message is received, decoded, and processed by one or more subscribers.

With this simple design, NATS lets programs share common message-handling code, isolate resources and interdependencies, and scale by easily handling an increase in message volume, whether those are service requests or stream data.

NATS Quality of service (QoS)

NATS offers multiple qualities of service, depending on whether the application uses just the Core NATS functionality or also leverages the added functionalities enabled by NATS JetStream (JetStream is built into nats-server but may not be enabled on all service infrastructures).

Walkthrough Setup

We have provided Walkthroughs for you to try NATS (and JetStream) on your own. In order to follow along with the walkthroughs, you could choose one of these options:

Installing the nats CLI Tool

Please refer to the installation section in the readme.

Installing the NATS server locally (if needed)

If you are going to run a server locally you need to first install it and start it.
Please refer to the nats server installation doc

Alternatively if you already know how to use NATS on a remote server, you only need to pass the server URL to nats using the -s option or preferably create a context using nats context add, to specify the server URL(s) and credentials file containing your user JWT.

Start the NATS server (if needed)

To start a simple demonstration server locally, simply run:

nats-server

(or nats-server -m 8222 if you want to enable the HTTP monitoring functionality)

When the server starts successfully, you will see the following messages:

[14524] 2021/10/25 22:53:53.525530 [INF] Starting nats-server
[14524] 2021/10/25 22:53:53.525640 [INF]   Version:  2.6.1
[14524] 2021/10/25 22:53:53.525643 [INF]   Git:      [not set]
[14524] 2021/10/25 22:53:53.525647 [INF]   Name:     NDAUZCA4GR3FPBX4IFLBS4VLAETC5Y4PJQCF6APTYXXUZ3KAPBYXLACC
[14524] 2021/10/25 22:53:53.525650 [INF]   ID:       NDAUZCA4GR3FPBX4IFLBS4VLAETC5Y4PJQCF6APTYXXUZ3KAPBYXLACC
[14524] 2021/10/25 22:53:53.526392 [INF] Starting http monitor on 0.0.0.0:8222
[14524] 2021/10/25 22:53:53.526445 [INF] Listening for client connections on 0.0.0.0:4222
[14524] 2021/10/25 22:53:53.526684 [INF] Server is ready

The NATS server listens for client connections on TCP Port 4222.

Subject-Based Messaging

NATS is a system for publishing and listening for messages on named communication channels we call Subjects. Fundamentally, NATS is an interest-based messaging system, where the listener has to subscribe to a subset of subjects.

In other middleware systems subjects may be called topics, channels, streams (Note that in NATS the term stream is used for a JetStream message storage).

What is a subject? At its simplest, a subject is just a string of characters that form a name the publisher and subscriber can use to find each other. More commonly subject hierarchies are used to scope messages into semantic namespaces.

提示

Please check the constraint and conventions on naming for subjects here.

Location transparency Through subject-based addressing, NATS provides location transparency across a (large) cloud of routed NATS servers.

Wildcards

NATS provides two wildcards that can take the place of one or more elements in a dot-separated subject. Publishers will always send a message to a fully specified subject, without the wildcard. While subscribers can use these wildcards to listen to multiple subjects with a single subscription.

Subject hierarchies

The . character is used to create a subject hierarchy. For example, a world clock application might define the following to logically group related subjects:

time.us
time.us.east
time.us.east.atlanta
time.eu.east
time.eu.east.warsaw

Subject usage best practices

There is no hard limit to subject size, but it is recommended to keep the maximum number of tokens in your subjects to a reasonable value. E.g. a maximum of 16 tokens and the subject length to less than 256 characters.

Number of subjects

NATS can manage 10s of millions of subjects efficiently, therefore, you can use fine-grained addressing for your business entities. Subjects are ephemeral resources, which will disappear when no longer subscribed to.

Still, subject subscriptions need to be cached by the server in memory. Consider when increasing your subscribed subject count to more than one million you will need more than 1GB of server memory and it will grow linearly from there.

Subject-based filtering and security

The message subject can be filtered with various means and through various configuration elements in your NATS server cluster. For example, but not limited to:

A well-designed subject hierarchy will make the job a lot easier for those tasks.

Naming things

提示

There are only two hard problems in computer science: cache invalidation, naming things, and off-by-one errors. -- Unknown author

A subject hierarchy is a powerful tool for addressing your application resources. Most NATS users therefore encode business semantics into the subject name. You are free to choose a structure fit for your purpose, but you should refrain from over-complicating your subject design at the start of the project.

Some guidelines:

factory1.tools.group42.unit17

service.deploy.server-acme.app123

Pragmatic:

orders.online.store123.order171711

Maybe not so useful:

orders.online.us.server42.ccpayment.premium.store123.electronics.deliver-dhl.order171711.create

Matching a single token

The first wildcard is * which will match a single token. For example, if an application wanted to listen for eastern time zones, they could subscribe to time.*.east, which would match time.us.east and time.eu.east. Note that * can not match a substring within a token time.New*.east.

Matching multiple tokens

The second wildcard is > which will match one or more tokens, and can only appear at the end of the subject. For example, time.us.> will match time.us.east and time.us.east.atlanta, while time.us.* would only match time.us.east since it can't match more than one token.

Monitoring and wire taps

Subject to your security configuration, wildcards can be used for monitoring by creating something called a wire tap. In the simplest case, you can create a subscriber for >. This application will receive all messages -- again, subject to security settings -- sent on your NATS cluster.

Mixing wildcards

The wildcard * can appear multiple times in the same subject. Both types can be used as well. For example, *.*.east.> will receive time.us.east.atlanta.

Characters allowed and recommended for subject names

For compatibility across clients and ease of maintaining configuration files, we recommend using alphanumeric characters, - (dash) and _ (underscore) ASCII characters for subject and other entity names created by the user.

UTF-8 (UTF8) characters are supported in subjects. Please use UTF-8 characters at your own risk. Using multilingual names for technical entities can create many issues for editing, configuration files, display, and cross-border collaboration.

The rules and recommendations here apply to ALL system names, subjects, streams, durables, buckets, keys (in key-value stores), as NATS will create API subjects that contain those names. NATS will enforce these constraints in most cases, but we recommend not relying on this.

Good names

time.us
time.us2.east1
time.new-york
time.SanFrancisco

Deprecated subject names

location.Malmö
$location.Stockholm
_Subjects_.mysubject

Forbidden stream names

all*data
<my_stream>
service.stream.1

Pedantic mode

By default, for the sake of efficiency, subject names are not verified during message publishing. In particular, when generating subjects programmatically, this will result in illegal subjects which cannot be subscribed to. E.g. subjects containing wildcards may be ignored.

To enable subject name verification, activate pedantic mode in the client connection options.

//Java
Options options = Options.builder()
    .server("nats://127.0.0.1:4222")
    .pedantic()
    .build();
Connection nc = Nats.connect(options)

Core NATS

Core NATS is the foundational functionality in a NATS system. It operates on a publish-subscribe model using subject/topic-based addressing. This model offers two significant advantages: location independence and a default many-to-many (M:N) communication pattern. These fundamental concepts enable powerful and innovative solutions for common development patterns, such as microservices, without requiring additional technologies like load balancers, API gateways, or DNS configuration.

NATS systems can be enhanced with JetStream, which adds persistence capabilities. While Core NATS provides best-effort, at-most-once message delivery, JetStream introduces at-least-once and exactly-once semantics.

Publish-Subscribe

NATS implements a publish-subscribe message distribution model for one-to-many communication. A publisher sends a message on a subject and any active subscriber listening on that subject receives the message. Subscribers can also register interest in wildcard subjects that work a bit like a regular expression (but only a bit). This one-to-many pattern is sometimes called a fan-out.

Messages

Messages are composed of:

  1. A subject.
  2. A payload in the form of a byte array.
  3. Any number of header fields.
  4. An optional 'reply' address field.

Messages have a maximum size (which is set in the server configuration with max_payload). The size is set to 1 MB by default, but can be increased up to 64 MB if needed (though we recommend keeping the max message size to something more reasonable like 8 MB).

Pub/Sub Walkthrough

NATS is a publish subscribe messaging system based on subjects. Subscribers listening on a subject receive messages published on that subject. If the subscriber is not actively listening on the subject, the message is not received. Subscribers can use the wildcard tokens such as * and > to match a single token or to match the tail of a subject.

NATS Pub/Sub Walkthrough

This simple walkthrough demonstrates some ways in which subscribers listen on subjects, and publishers send messages on specific subjects.

Walkthrough prerequisites

If you have not already done so, you need to install the nats CLI Tool and optionally the nats-server on your machine.

1. Create Subscriber 1

In a shell or command prompt session, start a client subscriber program.

nats sub <subject>

Here, <subject> is a subject to listen on. It helps to use unique and well thought-through subject strings because you need to ensure that messages reach the correct subscribers even when wildcards are used.

For example:

nats sub msg.test

You should see the message: Listening on [msg.test]

2. Create a Publisher and publish a message

In another shell or command prompt, create a NATS publisher and send a message.

nats pub <subject> <message>

Where <subject> is the subject name and <message> is the text to publish.

For example:

nats pub msg.test "NATS MESSAGE"

3. Verify message publication and receipt

You'll notice that the publisher sends the message and prints: Published [msg.test] : 'NATS MESSAGE'.

The subscriber receives the message and prints: [#1] Received on [msg.test]: 'NATS MESSAGE'.

If the receiver does not get the message, you'll need to check if you are using the same subject name for the publisher and the subscriber.

4. Try publishing another message

nats pub msg.test "NATS MESSAGE 2"

You'll notice that the subscriber receives the message.
Note that a message count is incremented each time your subscribing client receives a message on that subject.

5. Create Subscriber 2

In a new shell or command prompt, start a new NATS subscriber.

nats sub msg.test

6. Publish another message using the publisher client

nats pub msg.test "NATS MESSAGE 3"

Verify that both subscribing clients receive the message.

7. Create Subscriber 3

In a new shell or command prompt session, create a new subscriber that listens on a different subject.

nats sub msg.test.new

8. Publish another message

nats pub msg.test "NATS MESSAGE 4"

Subscriber 1 and Subscriber 2 receive the message, but Subscriber 3 does not. Why? Because Subscriber 3 is not listening on the message subject used by the publisher.

9. Alter Subscriber 3 to use a wildcard

Change the last subscriber to listen on msg.* and run it:

nats sub msg.*

Note: NATS supports the use of wildcard characters for message subscribers only. You cannot publish a message using a wildcard subject.

10. Publish another message

nats pub msg.test "NATS MESSAGE 5"

This time, all three subscribing clients should receive the message.

Do try out a few more variations of substrings and wildcards to test your understanding.

See Also

Publish-subscribe pattern with the NATS CLI

https://www.youtube.com/watch?v=jLTVhP08Tq0 Publish-subscribe pattern - NATS CLI

Request-Reply

Request-Reply is a common pattern in modern distributed systems. A request is sent, and the application either waits on the response with a certain timeout, or receives a response asynchronously.

The increased complexity of modern systems necessitates features like location transparency, scale-up and scale-down, observability (measuring a system's state based on the data it generates) and more. In order to implement this feature-set, various other technologies needed to incorporate additional components, sidecars (processes or services that support the primary application) and proxies. NATS on the other hand, implemented Request-Reply much more easily.

NATS makes Request-Reply simple and powerful

The pattern

Try NATS request-reply on your own, using a live server by walking through the request-reply walkthrough.

No responders

When a request is sent to a subject that has no subscribers, it can be convenient to know about it right away. For this use-case, a NATS client can opt-into no_responder messages. This requires a server and client that support headers. When enabled, a request sent to a subject with no subscribers will immediately receive a reply that has no body, and a 503 status.

Most clients will represent this case by raising or returning an error. For example:

m, err := nc.Request("foo", nil, time.Second);
# err == nats.ErrNoResponders

Request-Reply Walkthrough

NATS supports request-reply messaging. In this tutorial you explore how to exchange point-to-point messages using NATS.

Prerequisites

If you have not already done so, you need to install the nats CLI Tool and optionally, the nats-server on your machine.

Walkthrough

Start two terminal sessions. These will be used to run the NATS request and reply clients.

In one terminal, run the reply client listener

nats reply help.please 'OK, I CAN HELP!!!'

You should see the message: Listening on [help.please]

This means that the NATS receiver client is listening for request messages on the "help.please" subject. In NATS, the receiver is a subscriber.

In the other terminal, run the request client

nats request help.please 'I need help!'

The NATS requestor client makes a request by sending the message "I need help!" on the “help.please” subject.

The NATS receiver client receives the message, formulates the reply ("OK, I CAN HELP!!!"), and sends it to the inbox of the requester.

Queue Groups

When subscribers register themselves to receive messages from a publisher, the 1:N fan-out pattern of messaging ensures that any message sent by a publisher, reaches all subscribers that have registered. NATS provides an additional feature named "queue", which allows subscribers to register themselves as part of a queue. Subscribers that are part of a queue, form the "queue group".

How queue groups function

As an example, consider message delivery occurring in the 1:N pattern to all subscribers based on the subject name (delivery happens even to subscribers that are not part of a queue group). If a subscriber is registered based on a queue name, it will always receive messages it is subscribed to, based on the subject name. However, if more subscribers are added to the same queue name, they become a queue group, and only one randomly chosen subscriber of the queue group will consume a message each time a message is received by the queue group. Such distributed queues are a built-in load balancing feature that NATS provides.

Advantages

Queue group names follow the same naming rules as subjects. Foremost, they are case sensitive and cannot contain whitespace. Consider structuring queue groups hierarchically using a period .. Some server functionalities like queue permissions can use wildcard matching on them.

Queue subscribers are ideal for scaling services. Scale up is as simple as running another application, scale down is terminating the application with a signal that drains the in flight requests. This flexibility and lack of any configuration changes makes NATS an excellent service communication technology that can work with all platform technologies.

No responder

When a request is made to a service (request/reply) and the NATS Server knows there are no services available (since there are no client applications currently subscribing to the subject in a queue-group) the server will send a “no-responders” protocol message back to the requesting client which will break from blocking API calls. This allows applications to react immediately. This further enables building a highly responsive system at scale, even in the face of application failures and network partitions.

Stream as a queue

With JetStream a stream can also be used as a queue by setting the retention policy to WorkQueuePolicy and leveraging pull consumers to get easy horizontal scalability of the processing (or using an explicit ack push consumer with a queue group of subscribers).

Queuing geo-affinity

When connecting to a globally distributed NATS super-cluster, there is an automatic service geo-affinity due to the fact that a service request message will only be routed to another cluster (i.e. another region) if there are no listeners on the cluster available to handle the request locally.

Tutorial

Try NATS queue subscriptions on your own, using a live server by walking through the queueing walkthrough.

Queueing Walkthrough

NATS supports a form of load balancing using queue groups. Subscribers register a queue group name. A single subscriber in the group is randomly selected to receive the message.

Walkthrough prerequisites

If you have not already done so, you need to install the nats CLI tool and optionally the nats-server on your machine.

1. Start the first member of the queue group

The nats reply instances don't just subscribe to the subject but also automatically join a queue group ("NATS-RPLY-22" by default)

nats reply foo "service instance A Reply# {{Count}}"

2. Start a second member of the queue group

In a new window

nats reply foo "service instance B Reply# {{Count}}"

3. Start a third member of the queue group

In a new window

nats reply foo "service instance C Reply# {{Count}}"

4. Publish a NATS message

nats request foo "Simple request"

5. Verify message publication and receipt

You should see that only one of the my-queue group subscribers receives the message and replies to it, and you can also see which one of the available queue group subscribers processed the request from the reply message received (i.e. service instance A, B or C)

6. Publish another message

nats request foo "Another simple request"

You should see that a different queue group subscriber receives the message this time, chosen at random among the 3 queue group members.

You can also send any number of requests back-to-back. From the received messages, you'll see the distribution of those requests amongst the members of the queue group. For example: nats request foo --count 10 "Request {{Count}}"

7. Stop/start queue group members

You can at any time start yet another service instance, or stop one and see how the queue group automatically takes care of adding/removing those instances from the group.

See Also

Queue groups using the NATS CLI

https://youtu.be/jLTVhP08Tq0?t=101 Queue Groups NATS CLI

JetStream

NATS has a built-in persistence engine called JetStream which enables messages to be stored and replayed at a later time. Unlike NATS Core which requires you to have an active subscription to process messages as they happen, JetStream allows the NATS server to capture messages and replay them to consumers as needed. This functionality enables a different quality of service for your NATS messages, and enables fault-tolerant and high-availability configurations.

JetStream is built into nats-server. If you have a cluster of JetStream-enabled servers you can enable data replication and thus guard against failures and service disruptions.

JetStream was created to address the problems identified with streaming technology today - complexity, fragility, and a lack of scalability. Some technologies address these better than others, but no current streaming technology is truly multi-tenant, horizontally scalable, or supports multiple deployment models. No other technology that we are aware of can scale from edge to cloud using the same security context while having complete deployment observability for operations.

Additional capabilities enabled by JetStream

The JetStream persistence layer enables additional use cases typically not found in messaging systems. Being built on top of JetStream they inherit the core capabilities of JetStream, replication, security, routing limits, and mirroring.

Key/Value and File transfer are capabilities commonly found in in-memory databases or deployment tools. While NATS does not intend to compete with the feature set of such tools, it is our goal to provide the developer with reasonable complete set of data storage and replications features for use cases like micro service, edge deployments and server management.

Configuration

To configure a nats-server with JetStream refer to:

Examples

For runnable JetStream code examples, refer to NATS by Example.

Goals

JetStream was developed with the following goals in mind:

JetStream capabilities

Streaming: temporal decoupling between the publishers and subscribers

One of the tenets of basic publish/subscribe messaging is that there is a required temporal coupling between the publishers and the subscribers: subscribers only receive the messages that are published when they are actively connected to the messaging system (i.e. they do not receive messages that are published while they are not subscribing or not running or disconnected). The traditional way for messaging systems to provide temporal decoupling of the publishers and subscribers is through the 'durable subscriber' functionality or sometimes through 'queues', but neither one is perfect:

However, in many use cases, you do not need to 'consume exactly once' functionality but rather the ability to replay messages on demand, as many times as you want. This need has led to the popularity of some 'streaming' messaging platforms.

JetStream provides both the ability to consume messages as they are published (i.e. 'queueing') as well as the ability to replay messages on demand (i.e. 'streaming'). See retention policies below.

Replay policies

JetStream consumers support multiple replay policies, depending on whether the consuming application wants to receive either:

Retention policies and limits

JetStream enables new functionalities and higher qualities of service on top of the base 'Core NATS' functionality. However, practically speaking, streams can't always just keep growing 'forever' and therefore JetStream supports multiple retention policies as well as the ability to impose size limits on streams.

Limits

You can impose the following limits on a stream

You must also select a discard policy which specifies what should happen once the stream has reached one of its limits and a new message is published:

Retention policy

You can choose what kind of retention you want for each stream:

Note that regardless of the retention policy selected, the limits (and the discard policy) always apply.

Subject mapping transformations

JetStream also enables the ability to apply subject mapping transformations to messages as they are ingested into a stream.

Persistent and Consistent distributed storage

You can choose the durability as well as the resilience of the message storage according to your needs.

JetStream uses a NATS optimized RAFT distributed quorum algorithm to distribute the persistence service between NATS servers in a cluster while maintaining immediate consistency (as opposed to eventual consistency) even in the face of failures.

For writes (publications to a stream), the formal consistency model of NATS JetStream is Linearizable. On the read side (listening to or replaying messages from streams) the formal models don't really apply because JetStream does not support atomic batching of multiple operations together (so the only kind of 'transaction' is the persisting, replicating and voting of a single operation on the stream) but in essence, JetStream is serializable because messages are added to a stream in one global order (which you can control using compare and publish).

Do note, while we do guarantee immediate consistency when it comes to monotonic writes and monotonic reads. We don't guarantee read your writes at this time, as reads through direct get requests may be served by followers or mirrors. More consistent results can be achieved by sending get requests to the stream leader.

JetStream can also provide encryption at rest of the messages being stored.

In JetStream the configuration for storing messages is defined separately from how they are consumed. Storage is defined in a Stream and consuming messages is defined by multiple Consumers.

Stream replication factor

A stream's replication factor (R, often referred to as the number 'Replicas') determines how many places it is stored allowing you to tune to balance risk with resource usage and performance. A stream that is easily rebuilt or temporary might be memory-based with a R=1 and a stream that can tolerate some downtime might be file-based R-1.

Typical usage to operate in typical outages and balance performance would be a file-based stream with R=3. A highly resilient, but less performant and more expensive configuration is R=5, the replication factor limit.

Rather than defaulting to the maximum, we suggest selecting the best option based on the use case behind the stream. This optimizes resource usage to create a more resilient system at scale.

Mirroring and Sourcing between streams

JetStream also allows server administrators to easily mirror streams, for example between different JetStream domains in order to offer disaster recovery. You can also define a stream that 'sources' from one or more other streams.

Syncing data to disk

JetStream’s file-based streams persist messages to disk. However, while JetStream does flush file writes to the OS synchronously, under the default configuration it does not immediately fsync data to disk. The server uses a configurable sync_interval option, with a default value of 2 minutes, which controls how often the server will fsync its data. The data will be fsync-ed no later than this interval. This has important consequences for durability with respect to OS failures (meaning ungraceful exit of the Operating System such as a power outage, and not just ungraceful exit or killing of the nats-server process itself):

In a non-replicated setup, an OS failure may result in data loss. A client might publish a message and receive an acknowledgment, but the data may not yet be safely stored to disk. As a result, after an OS failure recovery, a server may have lost recently acknowledged messages.

In a replicated setup, a published message is acknowledged after it successfully replicated to at least a quorum of servers. However, replication alone is not enough to guarantee the strongest level of durability against multiple systemic failures.

Setting a lower sync_interval increases the frequency of disk writes, and reduces the window for potential data loss, but at the expense of performance. Additionally, setting sync_interval: always will make sure servers fsync after every message before it is acknowledged. This setting, combined with replication in different data centers or availability zones, provides the strongest durability guarantees but at the slowest performance.

The default settings have been chosen to balance performance and risk of data loss in what we consider to be a typical production deployment scenario across multiple availability zones.

For example, consider a stream with 3 replicas deployed across three separate availability zones. For the stream state to diverge across nodes would require that:

In the end, 2 out of 3 nodes will be available, the previous stream leader with the writes will be unavailable, one server will have lost some writes due to the OS failure, and one server will have never seen these writes due to the earlier partition. The last two servers could then form a majority and accept new writes, essentially losing some of the former writes.

Importantly this is a failure condition where stream state could diverge, but in a system that is deployed across multiple availability zones, it would require multiple faults to align precisely in the right way.

A potential mitigation to a failure of this kind is not automatically bringing back a server process that was OS-failed until it is known that a majority of the remaining servers have received the new writes, or by peer-removing the crashed server and admitting it as a new and wiped peer and allowing it to recover over the network from existing healthy nodes (although this could be expensive depending on the amount of data involved).

For use cases where minimizing loss is an absolute priority, sync_interval: always can of course still be configured, but note that this will have a server-wide performance impact that may affect throughput or latencies. For production environments, operators should evaluate whether the default is correct for their use case, target environment, costs, and performance requirements.

Alternatively, a hybrid approach can be used where existing clusters still function under their default sync_interval settings but a new cluster gets added that’s configured with sync_interval: always, and utilizes server tags. The placement of a stream can then be specified to have this stream store data on this higher durability cluster through the use of placement tags.

# Configure a cluster that's dedicated to always sync writes.
server_tags: ["sync:always"]

jetstream {
    sync_interval: always
}

Create a replicated stream that’s specifically placed in the cluster using sync_interval: always, to ensure the strongest durability only for stream writes that require this level of durability.

nats stream add --replicas 3 --tag sync:always

De-coupled flow control

JetStream provides decoupled flow control over streams, the flow control is not 'end to end' where the publisher(s) are limited to publish no faster than the slowest of all the consumers (i.e. the lowest common denominator) can receive but is instead happening individually between each client application (publishers or consumers) and the nats server.

When using the JetStream publish calls to publish to streams there is an acknowledgment mechanism between the publisher and the NATS server, and you have the choice of making synchronous or asynchronous (i.e. 'batched') JetStream publish calls.

On the subscriber side, the sending of messages from the NATS server to the client applications receiving or consuming messages from streams is also flow controlled.

Exactly once semantics

Because publications to streams using the JetStream publish calls are acknowledged by the server the base quality of service offered by streams is 'at least once', meaning that while reliable and normally duplicate free there are some specific failure scenarios that could result in a publishing application believing (wrongly) that a message was not published successfully and therefore publishing it again, and there are failure scenarios that could result in a client application's consumption acknowledgment getting lost and therefore in the message being re-sent to the consumer by the server. Those failure scenarios while being rare and even difficult to reproduce do exist and can result in perceived 'message duplication' at the application level.

Therefore, JetStream also offers an 'exactly once' quality of service. For the publishing side, it relies on the publishing application attaching a unique message or publication ID in a message header and on the server keeping track of those IDs for a configurable rolling period of time in order to detect the publisher publishing the same message twice. For the subscribers a double acknowledgment mechanism is used to avoid a message being erroneously re-sent to a subscriber by the server after some kinds of failures.

Consumers

JetStream consumers are 'views' on a stream, they are subscribed to (or pulled) by client applications to receive copies of (or to consume if the stream is set as a working queue) messages stored in the stream.

Fast push consumers

Client applications can choose to use fast un-acknowledged push (ordered) consumers to receive messages as fast as possible (for the selected replay policy) on a specified delivery subject or to an inbox. Those consumers are meant to be used to 'replay' rather than 'consume' the messages in a stream.

Horizontally scalable pull consumers with batching

Client applications can also use and share pull consumers that are demand-driven, support batching and must explicitly acknowledge message reception and processing which means that they can be used to consume (i.e. use the stream as a distributed queue) as well as process the messages in a stream.

Pull consumers can and are meant to be shared between applications (just like queue groups) in order to provide easy and transparent horizontal scalability of the processing or consumption of messages in a stream without having (for example) to worry about having to define partitions or worry about fault-tolerance.

Note: using pull consumers doesn't mean that you can't get updates (new messages published into the stream) 'pushed' in real-time to your application, as you can pass a (reasonable) timeout to the consumer's Fetch call and call it in a loop.

Consumer acknowledgments

While you can decide to use un-acknowledged consumers trading quality of service for the fastest possible delivery of messages, most processing is not idem-potent and requires higher qualities of service (such as the ability to automatically recover from various failure scenarios that could result in some messages not being processed or being processed more than once) and you will want to use acknowledged consumers. JetStream supports more than one kind of acknowledgment:

Key Value Store

The JetStream persistence layer enables the Key Value store: the ability to store, retrieve and delete value messages associated with a key into a bucket.

Watch and History

You can subscribe to changes in a Key Value on the bucket or individual key level with watch and optionally retrieve a history of the values (and deletions) that have happened on a particular key.

Atomic updates and locking

The Key Value store supports atomic create and update operations. This enables pessimistic locks (by creating a key and holding on to it) and optimistic locks (using CAS - compare and set).

Object Store

The Object Store is similar to the Key Value Store. The key being replaced by a file name and value being designed to store arbitrarily large objects (e.g. files, even if they are very large) rather than 'values' that are message-sized (i.e. limited to 1Mb by default). This is achieved by chunking messages.

Legacy

Note that JetStream completely replaces the STAN legacy NATS streaming layer.

Streams

Streams are message stores, each stream defines how messages are stored and what the limits (duration, size, interest) of the retention are. Streams consume normal NATS subjects, any message published on those subjects will be captured in the defined storage system. You can do a normal publish to the subject for unacknowledged delivery, though it’s better to use the JetStream publish calls instead as the JetStream server will reply with an acknowledgement that it was successfully stored.

Orders

The diagram above shows the concept of storing all ORDERS.* in the Stream even though there are many types of order related messages. We’ll show how you can selectively consume subsets of messages later. Relatively speaking the Stream is the most resource consuming component so being able to combine related data in this manner is important to consider.

Streams can consume many subjects. Here we have ORDERS.* but we could also consume SHIPPING.state into the same Stream should that make sense.

Stream limits and message retention

Streams support various retention policies which define when messages in the stream can be automatically deleted, such as when stream limits are hit (like max count, size or age of messages), or also more novel options that apply on top of the limits such as interest-based retention or work-queue semantics (see Retention Policy).

Upon reaching message limits, the server will automatically discard messages either by removing the oldest messages to make room for new ones (DiscardOld) or by refusing to store new messages (DiscardNew). For more details, see Discard Policy.

Streams support deduplication using a Nats-Msg-Id header and a sliding window within which to track duplicate messages. See the Message Deduplication section.

For examples on how to configure streams with your preferred NATS client, see NATS by Example.

Configuration

Below are the set of stream configuration options that can be defined. The Version column indicates the version of the server the option was introduced. The Editable column indicates the option can be edited after the stream created. See client-specific examples here.

Field Description Version Editable
Name Identifies the stream and has to be unique within JetStream account. Names cannot contain whitespace, ., *, >, path separators (forward or backwards slash), and non-printable characters. 2.2.0 No
Storage The storage type for stream data. 2.2.0 No
Subjects A list of subjects to bind. Wildcards are supported. Cannot be set for mirror streams. 2.2.0 Yes
Replicas How many replicas to keep for each message in a clustered JetStream, maximum 5. 2.2.0 Yes
MaxAge Maximum age of any message in the Stream, expressed in nanoseconds. 2.2.0 Yes
MaxBytes Maximum number of bytes stored in the stream. Adheres to Discard Policy, removing oldest or refusing new messages if the Stream exceeds this size. 2.2.0 Yes
MaxMsgs Maximum number of messages stored in the stream. Adheres to Discard Policy, removing oldest or refusing new messages if the Stream exceeds this number of messages. 2.2.0 Yes
MaxMsgSize The largest message that will be accepted by the Stream. The size of a message is a sum of payload and headers. 2.2.0 Yes
MaxConsumers Maximum number of Consumers that can be defined for a given Stream, -1 for unlimited. 2.2.0 No
NoAck Default false. Disables acknowledging messages that are received by the Stream. This is mandatory when archiving messages which have a reply subject set. E.g. requests in an Request/Reply communication. By default JetStream will acknowledge each message with an empty reply on the reply subject. 2.2.0 Yes
Retention Declares the retention policy for the stream. 2.2.0 No
Discard The behavior of discarding messages when any streams’ limits have been reached. 2.2.0 Yes
DuplicateWindow The window within which to track duplicate messages, expressed in nanoseconds. 2.2.0 Yes
Placement Used to declare where the stream should be placed via tags and/or an explicit cluster name. 2.2.0 Yes
Mirror If set, indicates this stream is a mirror of another stream. 2.2.0 Yes (since 2.12.0)
Sources If defined, declares one or more streams this stream will source messages from. 2.2.0 Yes
MaxMsgsPerSubject Limits maximum number of messages in the stream to retain per subject. 2.3.0 Yes
Description A verbose description of the stream. 2.3.3 Yes
Sealed Sealed streams do not allow messages to be deleted via limits or API, sealed streams can not be unsealed via configuration update. Can only be set on already created streams via the Update API. 2.6.2 Yes (once)
DenyDelete Restricts the ability to delete messages from a stream via the API. 2.6.2 No
DenyPurge Restricts the ability to purge messages from a stream via the API. 2.6.2 No
AllowRollup Allows the use of the Nats-Rollup header to replace all contents of a stream, or subject in a stream, with a single new message. 2.6.2 Yes
RePublish If set, messages stored to the stream will be immediately republished to the configured subject. 2.8.3 Yes
AllowDirect If true, and the stream has more than one replica, each replica will respond to direct get requests for individual messages, not only the leader. 2.9.0 Yes
MirrorDirect If true, and the stream is a mirror, the mirror will participate in a serving direct get requests for individual messages from origin stream. 2.9.0 Yes
DiscardNewPerSubject If true, applies discard new semantics on a per subject basis. Requires DiscardPolicy to be DiscardNew and the MaxMsgsPerSubject to be set. 2.9.0 Yes
Metadata A set of application-defined key-value pairs for associating metadata on the stream. 2.10.0 Yes
Compression If file-based and a compression algorithm is specified, the stream data will be compressed on disk. Valid options are nothing (empty string) or s2 for Snappy compression. 2.10.0 Yes
FirstSeq If specified, a new stream will be created with its initial sequence set to this value. 2.10.0 No
SubjectTransform Applies a subject transform (to matching messages) before storing the message. 2.10.0 Yes
ConsumerLimits Sets default limits for consumers created for a stream. Those can be overridden per consumer. 2.10.0 Yes
AllowMsgTTL If set, allows header initiated per-message TTLs, instead of relying solely on MaxAge. 2.11.0 No (can only enable)
SubjectDeleteMarkerTTL If set, a subject delete marker will be placed after the last message of a subject ages out. This defines the TTL of the delete marker that's left behind. 2.11.0 Yes
AllowAtomicPublish If set, allows atomically writing a batch of N messages into the stream. 2.12.0 Yes
AllowBatchPublish If set, allows writing a batch of N messages into the stream. 2.14.0 Yes
AllowMsgCounter If set, the stream will function as a counter stream, hosting distributed counter CRDTs. 2.12.0 No
AllowMsgSchedules If set, allows message scheduling in the stream. 2.12.0 No (can only enable)

StorageType

The storage types include:

Subjects

Note: a stream configured as a mirror cannot be configured with a set of subjects. A mirror implicitly sources a subset of the origin stream (optionally with a filter), but does not subscribe to additional subjects.

If no explicit subject is specified, the default subject will be the same name as the stream. Multiple subjects can be specified and edited over time. Note, if messages are stored by a stream on a subject that is subsequently removed from the stream config, consumers will still observe those messages if their subject filter overlaps.

RetentionPolicy

The retention options include:

注意

If the InterestPolicy or WorkQueuePolicy is chosen for a stream, note that any limits, if defined, will still be enforced. For example, given a work-queue stream, if MaxMsgs are set and the default discard policy of old, messages will be automatically deleted even if the consumer did not receive them.

提示

If the InterestPolicy is chosen for a stream, note that when a consumer is deleted, either manually or as part of a server cleanup when the consumer's InactiveThreshold is reached, and that consumer was the last one that marked interest in a set of subjects, the server may not immediately delete all messages that belong to that set of subjects, for performance reasons. Instead, deletion may be deferred until those messages reach the start of the stream.

提示

WorkQueuePolicy streams will only delete messages enforced by limits or when a message has been successfully Ack’d by its consumer. Messages that have attempted redelivery and have reached MaxDeliver attempts for the consumer will remain in the stream and must be manually deleted via the JetStream API.

DiscardPolicy

The discard behavior applies only for streams that have at least one limit defined. The options include:

Placement

Refers to the placement of the stream assets (data) within a NATS deployment, be it a single cluster or a supercluster. A given stream, including all replicas (not mirrors), are bound to a single cluster. So when creating or moving a stream, a cluster will be chosen to host the assets.

Without declaring explicit placement for a stream, by default, the stream will be created within the cluster that the client is connected to assuming it has sufficient storage available.

By declaring stream placement, where these assets are located can be controlled explicitly. This is generally useful to co-locate with the most active clients (publishers or consumers) or may be required for data sovereignty reasons.

Placement is supported in all client SDKs as well as the CLI. For example, adding a stream via the CLI to place a stream in a specific cluster looks like this:

nats stream add --cluster aws-us-east1-c1

For this to work, all servers in a given cluster must define the name field within the cluster server configuration block.

cluster {
  name: aws-us-east1-c1
  # etc..
}

If you have multiple clusters that form a supercluster, then each is required to have a different name.

Another placement option are tags. Each server can have its own set of tags, defined in configuration, typically describing properties of geography, hosting provider, sizing tiers, etc. In addition, tags are often used in conjunction with the jetstream.unique_tag config option to ensure that replicas must be placed on servers having different values for the tag.

For example, a server A, B, and C in the above cluster might all the same configuration except for the availability zone they are deployed to.

// Server A
server_tags: ["cloud:aws", "region:us-east1", "az:a"]

jetstream: {
  unique_tag: "az"
}

// Server B
server_tags: ["cloud:aws", "region:us-east1", "az:b"]

jetstream: {
  unique_tag: "az"
}

// Server C
server_tags: ["cloud:aws", "region:us-east1", "az:c"]

jetstream: {
  unique_tag: "az"
}

Now we can create a stream by using tags, for example indicating we want a stream in us-east1.

nats stream add --tag region:us-east1

If we had a second cluster in Google Cloud with the same region tag, the stream could be placed in either the AWS or GCP cluster. However, the unique_tag constraint ensures each replica will be placed in a different AZ in the cluster that was selected implicitly by the placement tags.

Although less common, note that both the cluster and tags can be used for placement. This would be used if a single cluster contains servers have different properties.

Sources and Mirrors

When a stream is configured with a source or mirror, it will automatically and asynchronously replicate messages from the origin stream. There are several options when declaring the configuration.

A source or mirror stream can have its own retention policy, replication, and storage type. Changes to the source or mirror, e.g. deleting messages or publishing, do not reflect on the origin stream.

提示

Sources is a generalization of the Mirror and allows for sourcing data from one or more streams concurrently. We suggest to use Sources in new configurations. If you require the target stream to act as a read-only replica:

Stream sources

A stream defining Sources is a generalized replication mechanism and allows for sourcing data from one or more streams concurrently as well as allowing direct write/publish by clients. Essentially the source streams and client writes are aggregated into a single interleaved stream. Subject transformation and filtering allow for powerful data distribution architectures.

Mirrors

A mirror can source its messages from exactly one stream and a clients can not directly write to the mirror. Although messages cannot be published to a mirror directly by clients, messages can be deleted on-demand (beyond the retention policy), and consumers have all capabilities available on regular streams.

For details see:

AllowRollup

If enabled, the AllowRollup stream option allows for a published message having a Nats-Rollup header indicating all prior messages should be purged. The scope of the purge is defined by the header value, either all or sub.

The Nats-Rollup: all header will purge all prior messages in the stream. Whereas the sub value will purge all prior messages for a given subject.

A common use case for rollup is for state snapshots, where the message being published has accumulated all the necessary state from the prior messages, relative to the stream or a particular subject.

RePublish

If enabled, the RePublish stream option will result in the server re-publishing messages received into a stream automatically and immediately after a successful write, to a distinct destination subject.

For high scale needs where, currently, a dedicated consumer may add too much overhead, clients can establish a core NATS subscription to the destination subject and receive messages that were appended to the stream in real-time.

The fields for configuring republish include:

For each message that is republished, a set of headers are automatically added.

Note: The reply-subject is being removed on republished messages, even if no-ack is set on the stream.

SubjectTransform

If configured, the SubjectTransform will perform a subject transform to matching subjects of messages received by the stream and transform the subject, before storing it in the stream. The transform configuration specifies a Source and Destination field, following the rules of subject transform.

Source and Mirror Streams

When a stream is configured with a source or mirror, it will automatically and asynchronously replicate messages from the origin stream.

source or mirror are designed to be robust and will recover from a loss of connection. They are suitable for geographic distribution over high latency and unreliable connections. E.g. even a leaf node starting and connecting intermittently every few days will still receive or send messages over the source/mirror link.
Another use case is when connecting streams cross-account.

There are several options available when declaring the configuration.

The stream using a source or mirror configuration can have its own retention policy, replication, and storage type.

提示

提示

Sources is a generalization of the Mirror and allows for sourcing data from one or more streams concurrently.
If you require the target stream to act as a read-only replica:

General behavior

Source specific

A stream defining Sources is a generalized replication mechanism and allows for sourcing data from one or more streams concurrently. A stream with sources can still act as a regular stream allowing direct write/publish by local clients to the stream. Essentially the source streams and local client writes are aggregated into a single interleaved stream.
Combined with subject transformation and filtering sourcing allows to design sophisticated data distribution architectures.

提示

Sourcing messages does not retain sequence numbers. But it retain the in stream sequence of messages . Between streams sourced to the same target, the sequence of messages is undefined.

Mirror specific

A mirror can source its messages from exactly one stream and a clients can not directly write to the mirror. Although messages cannot be published to a mirror directly by clients, messages can be deleted on-demand (beyond the retention policy), and consumers have all capabilities available on regular streams.

提示

Expected behavior in edge conditions

WorkQueue/Interest retention

提示

Stream sourcing and mirroring for WorkQueue or Interest streams are supported since version 2.14.

When using stream sourcing/mirroring on a WorkQueue or Interest stream, the consumer used is automatically 'upgraded' to a durable consumer. This consumer is visible to the user and can be monitored or paused to temporarily stop the sourcing from the sourced-side.

注意

On versions prior to 2.14, sourcing from a WorkQueue/Interest stream is not supported. JetStream does not forbid this configuration, but the behavior is undefined.

"Bring your own" sourcing consumer

Starting in version 2.14, you can "bring your own" sourcing consumer. This pre-created consumer can then be used for stream sourcing/mirroring, instead of the server automatically creating it based on the stream configuration. Useful for more flexible consumer configuration, as well as additional monitoring or operational capabilities, or generally when you're not in control of "the other side" and are required to do this from a security point of view.

The following options are used to "bring your own" sourcing consumer:

Other options like StartSeq, StartTime, FilterSubject, etc. are not supported when using this approach. Instead, these settings should be configured directly on the consumer.

This consumer is required to have AckFlowControl as its ack policy, meaning the sourced-side will only receive acknowledgements until after the sourcing-side has persisted the messages.

Example

Streams with source and mirror configurations are best managed through a client API. If you intend to create such a configuration from command line with NATS CLI you should use a JSON configuration.

nats stream add --config stream_with_sources.json

Example stream configuration with two sources

Minimal example

{
  "name": "SOURCE_TARGET",
  "subjects": [
    "foo1.ext.*",
    "foo2.ext.*"
  ],
  "discard": "old",
  "duplicate_window": 120000000000,
  "sources": [
    {
      "name": "SOURCE1_ORIGIN"
    }
  ],
  "deny_delete": false,
  "sealed": false,
  "max_msg_size": -1,
  "allow_rollup_hdrs": false,
  "max_bytes": -1,
  "storage": "file",
  "allow_direct": false,
  "max_age": 0,
  "max_consumers": -1,
  "max_msgs_per_subject": -1,
  "num_replicas": 1,
  "name": "SOURCE_TARGET",
  "deny_purge": false,
  "compression": "none",
  "max_msgs": -1,
  "retention": "limits",
  "mirror_direct": false
}

With additional options

{
  "name": "SOURCE_TARGET",
  "subjects": [
    "foo1.ext.*",
    "foo2.ext.*"
  ],
  "discard": "old",
  "duplicate_window": 120000000000,
  "sources": [
    {
      "name": "SOURCE1_ORIGIN",
      "filter_subject": "foo1.bar",
      "opt_start_seq": 42,
      "external": {
        "deliver": "",
        "api": "$JS.domainA.API"
      }
    },
    {
      "name": "SOURCE2_ORIGIN",
      "filter_subject": "foo2.bar"
    }
  ],
  "consumer_limits": {
    
  },
  "deny_delete": false,
  "sealed": false,
  "max_msg_size": -1,
  "allow_rollup_hdrs": false,
  "max_bytes": -1,
  "storage": "file",
  "allow_direct": false,
  "max_age": 0,
  "max_consumers": -1,
  "max_msgs_per_subject": -1,
  "num_replicas": 1,
  "name": "SOURCE_TARGET",
  "deny_purge": false,
  "compression": "none",
  "max_msgs": -1,
  "retention": "limits",
  "mirror_direct": false
}

Example stream configuration with mirror

Minimal example

{
  "name": "MIRROR_TARGET"
  "discard": "old",
  "mirror": {
    "name": "MIRROR_ORIGIN"
  },
  "deny_delete": false,
  "sealed": false,
  "max_msg_size": -1,
  "allow_rollup_hdrs": false,
  "max_bytes": -1,
  "storage": "file",
  "allow_direct": false,
  "max_age": 0,
  "max_consumers": -1,
  "max_msgs_per_subject": -1,
  "num_replicas": 1,
  "name": "MIRROR_TARGET",
  "deny_purge": false,
  "compression": "none",
  "max_msgs": -1,
  "retention": "limits",
  "mirror_direct": false
}

With additional options

{
  "name": "MIRROR_TARGET"
  "discard": "old",
  "mirror": {
    "opt_start_time": "2024-07-11T08:57:20.4441646Z",
    "external": {
      "deliver": "",
      "api": "$JS.domainB.API"
    },
    "name": "MIRROR_ORIGIN"
  },
  "consumer_limits": {
    
  },
  "deny_delete": false,
  "sealed": false,
  "max_msg_size": -1,
  "allow_rollup_hdrs": false,
  "max_bytes": -1,
  "storage": "file",
  "allow_direct": false,
  "max_age": 0,
  "max_consumers": -1,
  "max_msgs_per_subject": -1,
  "num_replicas": 1,
  "name": "MIRROR_TARGET",
  "deny_purge": false,
  "compression": "none",
  "max_msgs": -1,
  "retention": "limits",
  "mirror_direct": false
}

Consumers

A consumer is a stateful view of a stream. It acts as an interface for clients to consume a subset of messages stored in a stream and will keep track of which messages were delivered and acknowledged by clients.

Unlike Core NATS, which provides an at most once delivery guarantee, a consumer in JetStream can provide an at least once delivery guarantee.

While Streams are responsible for storing the published messages, the consumer is responsible for tracking the delivery and acknowledgments. This tracking ensures that if a message is not acknowledged (un-acked or 'nacked'), the consumer will automatically attempt to re-deliver it. JetStream consumers support various acknowledgment types and policies. If a message is not acknowledged within a user-specified number of delivery attempts, an advisory notification is emitted.

Dispatch type - Pull / Push

Consumers can be push-based where messages will be delivered to a specified subject or pull-based which allows clients to request batches of messages on demand. The choice of what kind of consumer to use depends on the use-case.

If there is a need to process messages in an application controlled manner and easily scale horizontally, you would use a 'pull consumer'. A simple client application that wants a replay of messages from a stream sequentially you would use an 'ordered push consumer'. An application that wants to benefit from load balancing or acknowledge messages individually will use a regular push consumer.

提示 We recommend pull consumers for new projects. In particular when scalability, detailed flow control or error handling are a concern.

Ordered Consumers

Ordered consumers are the convenient default type of push & pull consumers designed for applications that want to efficiently consume a stream for data inspection or analysis.

Persistence - Durable / Ephemeral

In addition to the choice of being push or pull, a consumer can also be ephemeral or durable. A consumer is considered durable when an explicit name is set on the Durable field when creating the consumer; otherwise it is ephemeral. Setting InactiveThreshold does not make a consumer durable. InactiveThreshold only controls automatic cleanup of an inactive consumer and applies to both durable and ephemeral consumers (prior to 2.9 it applied to ephemeral consumers only). Note also that the Name field is distinct from Durable: setting Name without Durable creates a named consumer that the server still treats as ephemeral.

Durables and ephemeral have the same message delivery semantics but an ephemeral consumer will not have persisted state or fault tolerance (server memory only) and will be automatically cleaned up (deleted) after a period of inactivity, when no subscriptions are bound to the consumer.

By default, consumers will have the same replication factor as the stream they consume, and will remain even when there are periods of inactivity (unless InactiveThreshold is set explicitly). Consumers can recover from server and client failure.

https://youtu.be/334XuMma1fk NATS JS Consumers - The ONE feature that makes NATS more powerful than Kafka, Pulsar, RabbitMQ, & redis

Configuration

Below are the set of consumer configuration options that can be defined. The Version column indicates the version of nats-server in which the option was introduced. The Editable column indicates the option can be edited after the consumer is created.

General

Field Description Version Editable
Durable If set, clients can have subscriptions bind to the consumer and resume until the consumer is explicitly deleted. A durable name cannot contain whitespace, ., *, >, path separators (forward or backward slash), or non-printable characters. 2.2.0 No
FilterSubject A subject that overlaps with the subjects bound to the stream to filter delivery to subscribers. Note: This cannot be used with the FilterSubjects field. 2.2.0 Yes
AckPolicy The requirement of client acknowledgments, either AckExplicit, AckNone, AckAll, or AckFlowControl (2.14). 2.2.0 No
AckWait The duration that the server will wait for an acknowledgment for any individual message once it has been delivered to a consumer. If an acknowledgment is not received in time, the message will be redelivered. This setting is only effective when BackOff is not configured. 2.2.0 Yes
DeliverPolicy The point in the stream from which to receive messages: DeliverAll, DeliverLast, DeliverNew, DeliverByStartSequence, DeliverByStartTime, or DeliverLastPerSubject. 2.2.0 No
OptStartSeq Used with the DeliverByStartSequence deliver policy. 2.2.0 No
OptStartTime Used with the DeliverByStartTime deliver policy. 2.2.0 No
Description A description of the consumer. This can be particularly useful for ephemeral consumers to indicate their purpose since a durable name cannot be provided. 2.3.3 Yes
InactiveThreshold Duration that instructs the server to clean up consumers inactive for that long. Prior to 2.9, this only applied to ephemeral consumers. 2.2.0 Yes
MaxAckPending Defines the maximum number of messages, without acknowledgment, that can be outstanding. Once this limit is reached, message delivery will be suspended. This limit applies across all of the consumer's bound subscriptions. A value of -1 means there can be any number of pending acknowledgments (i.e., no flow control). The default is 1000. 2.2.0 Yes
MaxDeliver The maximum number of times a specific message delivery will be attempted. Applies to any message that is re-sent due to acknowledgment policy (i.e., due to a negative acknowledgment or no acknowledgment sent by the client). The default is -1 (redeliver until acknowledged). Messages that have reached the maximum delivery count will stay in the stream. 2.2.0 Yes
Backoff A sequence of delays controlling the re-delivery of messages on acknowledgment timeout (but not on nak). The sequence length must be less than or equal to MaxDeliver. If backoff is not set, a timeout will result in immediate re-delivery. E.g., MaxDeliver=5 backoff=[5s, 30s, 300s, 3600s, 84000s] will re-deliver a message 5 times over one day. When MaxDeliver is larger than the backoff list, the last delay in the list will apply for the remaining deliveries. Note that backoff is NOT applied to naked messages. A nak will result in immediate re-delivery unless nakWithDelay is used to set the re-delivery delay explicitly. When BackOff is set, it overrides AckWait entirely. The first value in the BackOff determines the AckWait value. 2.7.1 Yes
ReplayPolicy If the policy is ReplayOriginal, the messages in the stream will be pushed to the client at the same rate they were originally received, simulating the original timing. If the policy is ReplayInstant (default), the messages will be pushed to the client as fast as possible while adhering to the acknowledgment policy, Max Ack Pending, and the client's ability to consume those messages. 2.2.0 No
Replicas Sets the number of replicas for the consumer's state. By default, when the value is set to zero, consumers inherit the number of replicas from the stream. 2.8.3 Yes
MemoryStorage If set, forces the consumer state to be kept in memory rather than inherit the storage type of the stream (default is file storage). This reduces I/O from acknowledgments, useful for ephemeral consumers. 2.8.3 No
SampleFrequency Sets the percentage of acknowledgments that should be sampled for observability, 0-100. This value is a string and allows both 30 and 30% as valid values. 2.2.0 Yes
Metadata A set of application-defined key-value pairs for associating metadata with the consumer. 2.10.0 Yes
FilterSubjects A set of subjects that overlap with the subjects bound to the stream to filter delivery to subscribers. Note: This cannot be used with the FilterSubject field. 2.10.0 Yes
HeadersOnly Delivers only the headers of messages in the stream, adding a Nats-Msg-Size header indicating the size of the removed payload. 2.6.2 Yes

AckPolicy

The policy choices include:

If an acknowledgment is required but not received within the AckWait window, the message will be redelivered.

Warning: The server may consider an acknowledgment arriving out of the window. For instance, in a queue situation, if a first process fails to acknowledge within the window and the message has been redelivered to another consumer, the acknowledgment from the first consumer will be considered.

DeliverPolicy

The policy choices include:

MaxAckPending

The MaxAckPending capability provides flow control and applies to both push and pull consumers. For push consumers, MaxAckPending is the only form of flow control. For pull consumers, client-driven message delivery creates implicit one-to-one flow control with subscribers.

For high throughput, set MaxAckPending to a high value. For applications with high latency due to external services, use a lower value and adjust AckWait to avoid re-deliveries.

FilterSubjects

A filter subject provides server-side filtering of messages before delivery to clients.

For example, a stream factory-events with subject factory-events.*.* can have a consumer factory-A with a filter factory-events.A.* to deliver only events for factory A.

A consumer can have a singular FilterSubject or plural FilterSubjects. Multiple filters can be applied, such as [factory-events.A.*, factory-events.B.*] or specific event types [factory-events.*.item_produced, factory-events.*.item_packaged].

Warning: For granular consumer permissions, a single filter uses $JS.API.CONSUMER.CREATE.{stream}.{consumer}.{filter} to restrict users to specific filters. Multiple filters use the general $JS.API.CONSUMER.DURABLE.CREATE.{stream}.{consumer}, which does not include the {filter} token. Use a different strategy for granular permissions.

Pull-specific

These options apply only to pull consumers. For configuration examples, see NATS by Example.

Field Description Version Editable
MaxWaiting The maximum number of waiting pull requests. 2.2.0 No
MaxRequestExpires The maximum duration a single pull request will wait for messages to be available to pull. 2.7.0 Yes
MaxRequestBatch The maximum batch size a single pull request can make. When set with MaxRequestMaxBytes, the batch size will be constrained by whichever limit is hit first. 2.7.0 Yes
MaxRequestMaxBytes The maximum total bytes that can be requested in a given batch. When set with MaxRequestBatch, the batch size will be constrained by whichever limit is hit first. 2.8.3 Yes

Push-specific

These options apply only to push consumers. For configuration examples, see NATS by Example.

Field Description Version Editable
DeliverSubject The subject to deliver messages to. Setting this field decides whether the consumer is push or pull-based. With a deliver subject, the server will push messages to clients subscribed to this subject. 2.2.0 No
DeliverGroup The queue group name used to distribute messages among subscribers. Analogous to a queue group in core NATS. 2.2.0 Yes
FlowControl Enables per-subscription flow control using a sliding-window protocol. This protocol relies on the server and client exchanging messages to regulate when and how many messages are pushed to the client. This one-to-one flow control mechanism works in tandem with the one-to-many flow control imposed by MaxAckPending across all subscriptions bound to a consumer. 2.2.0 Yes
IdleHeartbeat If set, the server will regularly send a status message to the client during inactivity, indicating that the JetStream service is up and running. The status message will have a code of 100 and no reply address. Note: This mechanism is handled transparently by supported clients. 2.2.0 Yes
RateLimit Throttles the delivery of messages to the consumer, in bits per second. 2.2.0 Yes

Example

Consider this architecture

Orders

While it is an incomplete architecture it does show a number of key points:

A new order arrives on ORDERS.received, gets sent to the NEW Consumer who, on success, will create a new message on ORDERS.processed. The ORDERS.processed message again enters the Stream where a DISPATCH Consumer receives it and once processed it will create an ORDERS.completed message which will again enter the Stream. These operations are all pull based meaning they are work queues and can scale horizontally. All require acknowledged delivery ensuring no order is missed.

All messages are delivered to a MONITOR Consumer without any acknowledgement and using Pub/Sub semantics - they are pushed to the monitor.

As messages are acknowledged to the NEW and DISPATCH Consumers, a percentage of them are Sampled and messages indicating redelivery counts, ack delays and more, are delivered to the monitoring system.

Example Configuration

Additional documentation introduces the nats utility and how you can use it to create, monitor, and manage streams and consumers, but for completeness and reference this is how you'd create the ORDERS scenario. We'll configure a 1 year retention for order related messages:

nats stream add ORDERS --subjects "ORDERS.*" --ack --max-msgs=-1 --max-bytes=-1 --max-age=1y --storage file --retention limits --max-msg-size=-1 --discard=old
nats consumer add ORDERS NEW --filter ORDERS.received --ack explicit --pull --deliver all --max-deliver=-1 --sample 100
nats consumer add ORDERS DISPATCH --filter ORDERS.processed --ack explicit --pull --deliver all --max-deliver=-1 --sample 100
nats consumer add ORDERS MONITOR --filter '' --ack none --target monitor.ORDERS --deliver last --replay instant

JetStream Walkthrough

The following is a small walkthrough on creating a stream and a consumer and interacting with the stream using the nats cli.

Prerequisite: enabling JetStream

If you are running a local nats-server stop it and restart it with JetStream enabled using nats-server -js (if that's not already done)

You can then check that JetStream is enabled by using

nats account info

Account Information

                           User: 
                        Account: $G
                        Expires: never
                      Client ID: 5
                      Client IP: 127.0.0.1
                            RTT: 128µs
              Headers Supported: true
                Maximum Payload: 1.0 MiB
                  Connected URL: nats://127.0.0.1:4222
              Connected Address: 127.0.0.1:4222
            Connected Server ID: NAMR7YBNZA3U2MXG2JH3FNGKBDVBG2QTMWVO6OT7XUSKRINKTRFBRZEC
       Connected Server Version: 2.11.0-dev
                 TLS Connection: no

JetStream Account Information:

Account Usage:

                        Storage: 0 B
                         Memory: 0 B
                        Streams: 0
                      Consumers: 0

Account Limits:

            Max Message Payload: 1.0 MiB

  Tier: Default:

      Configuration Requirements:

        Stream Requires Max Bytes Set: false
         Consumer Maximum Ack Pending: Unlimited

      Stream Resource Usage Limits:

                               Memory: 0 B of Unlimited
                    Memory Per Stream: Unlimited
                              Storage: 0 B of Unlimited
                   Storage Per Stream: Unlimited
                              Streams: 0 of Unlimited
                            Consumers: 0 of Unlimited

If you see the below then JetStream is not enabled

JetStream Account Information:

   JetStream is not supported in this account

1. Creating a stream

Let's start by creating a stream to capture and store the messages published on the subject "foo".

Enter nats stream add <Stream name> (in the examples below we will name the stream "my_stream"), then enter "foo" as the subject name and hit return to use the defaults for all the other stream attributes:

nats stream add my_stream

? Subjects foo
? Storage file
? Replication 1
? Retention Policy Limits
? Discard Policy Old
? Stream Messages Limit -1
? Per Subject Messages Limit -1
? Total Stream Size -1
? Message TTL -1
? Max Message Size -1
? Duplicate tracking time window 2m0s
? Allow message Roll-ups No
? Allow message deletion Yes
? Allow purging subjects or the entire stream Yes
Stream my_stream was created

Information for Stream my_stream created 2024-06-07 12:29:36

              Subjects: foo
              Replicas: 1
               Storage: File

Options:

             Retention: Limits
       Acknowledgments: true
        Discard Policy: Old
      Duplicate Window: 2m0s
            Direct Get: true
     Allows Msg Delete: true
          Allows Purge: true
        Allows Rollups: false

Limits:

      Maximum Messages: unlimited
   Maximum Per Subject: unlimited
         Maximum Bytes: unlimited
           Maximum Age: unlimited
  Maximum Message Size: unlimited
     Maximum Consumers: unlimited

State:

              Messages: 0
                 Bytes: 0 B
        First Sequence: 0
         Last Sequence: 0
      Active Consumers: 0

You can then check the information about the stream you just created:

nats stream info my_stream

Information for Stream my_stream created 2024-06-07 12:29:36

              Subjects: foo
              Replicas: 1
               Storage: File

Options:

             Retention: Limits
       Acknowledgments: true
        Discard Policy: Old
      Duplicate Window: 2m0s
            Direct Get: true
     Allows Msg Delete: true
          Allows Purge: true
        Allows Rollups: false

Limits:

      Maximum Messages: unlimited
   Maximum Per Subject: unlimited
         Maximum Bytes: unlimited
           Maximum Age: unlimited
  Maximum Message Size: unlimited
     Maximum Consumers: unlimited

State:

              Messages: 0
                 Bytes: 0 B
        First Sequence: 0
         Last Sequence: 0
      Active Consumers: 0

2. Publish some messages into the stream

Let's now start a publisher

nats pub foo --count=1000 --sleep 1s "publication #{{.Count}} @ {{.TimeStamp}}"

As messages are being published on the subject "foo" they are also captured and stored in the stream, you can check that by using nats stream info my_stream and even look at the messages themselves using nats stream view my_stream or nats stream get my_stream

3. Creating a consumer

Now at this point if you create a 'Core NATS' (i.e. non-streaming) subscriber to listen for messages on the subject 'foo', you will only receive the messages being published after the subscriber was started, this is normal and expected for the basic 'Core NATS' messaging. In order to receive a 'replay' of all the messages contained in the stream (including those that were published in the past) we will now create a 'consumer'

We can administratively create a consumer using the 'nats consumer add ' command, in this example we will name the consumer "pull_consumer", and we will leave the delivery subject to 'nothing' (i.e. just hit return at the prompt) because we are creating a 'pull consumer' and select all for the start policy, you can then just use the defaults and hit return for all the other prompts. The stream the consumer is created on should be the stream 'my_stream' we just created above.

nats consumer add

? Consumer name pull_consumer
? Delivery target (empty for Pull Consumers) 
? Start policy (all, new, last, subject, 1h, msg sequence) all
? Acknowledgment policy explicit
? Replay policy instant
? Filter Stream by subjects (blank for all) 
? Maximum Allowed Deliveries -1
? Maximum Acknowledgments Pending 0
? Deliver headers only without bodies No
? Add a Retry Backoff Policy No
? Select a Stream my_stream
Information for Consumer my_stream > pull_consumer created 2024-06-07T12:32:09-05:00

Configuration:

                    Name: pull_consumer
               Pull Mode: true
          Deliver Policy: All
              Ack Policy: Explicit
                Ack Wait: 30.00s
           Replay Policy: Instant
         Max Ack Pending: 1,000
       Max Waiting Pulls: 512

State:

  Last Delivered Message: Consumer sequence: 0 Stream sequence: 0
    Acknowledgment Floor: Consumer sequence: 0 Stream sequence: 0
        Outstanding Acks: 0 out of maximum 1,000
    Redelivered Messages: 0
    Unprocessed Messages: 74
           Waiting Pulls: 0 of maximum 512

You can check on the status of any consumer at any time using nats consumer info or view the messages in the stream using nats stream view my_stream or nats stream get my_stream, or even remove individual messages from the stream using nats stream rmm

3. Subscribing from the consumer

Now that the consumer has been created and since there are messages in the stream we can now start subscribing to the consumer:

nats consumer next my_stream pull_consumer --count 1000

This will print out all the messages in the stream starting with the first message (which was published in the past) and continuing with new messages as they are published until the count is reached.

Note that in this example we are creating a pull consumer with a 'durable' name, this means that the consumer can be shared between as many consuming processes as you want. For example instead of running a single nats consumer next with a count of 1000 messages you could have started two instances of nats consumer each with a message count of 500 and you would see the consumption of the messages from the consumer distributed between those instances of nats

Replaying the messages again

Once you have iterated over all the messages in the stream with the consumer, you can get them again by simply creating a new consumer or by deleting that consumer (nats consumer rm) and re-creating it (nats consumer add).

4. Cleaning up

You can clean up a stream (and release the resources associated with it (e.g. the messages stored in the stream)) using nats stream purge

You can also delete a stream (which will also automatically delete all of the consumers that may be defined on that stream) using nats stream rm

Key/Value Store

JetStream, the persistence layer of NATS, not only allows for the higher qualities of service and features associated with 'streaming', but it also enables some functionalities not found in messaging systems.

One such feature is the Key/Value store functionality, which allows client applications to create buckets and use them as immediately (as opposed to eventually) consistent, persistent associative arrays (or maps). Note that this is an abstraction on top of the Stream functionality. Buckets are materialized as Streams (with a name starting with KV_), everything you can do with a bucket you can do with a Stream, but you ultimately have more functionality and flexibility and control when using the Stream functionality directly.

Do note, while we do guarantee immediate consistency when it comes to monotonic writes and monotonic reads. We don't guarantee read your writes at this time, as reads through direct get requests may be served by followers or mirrors. More consistent results can be achieved by sending get requests to the underlying stream leader of the Key/Value store.

Managing a Key Value store

  1. Create a bucket, which corresponds to a stream in the underlying storage. Define KV/Stream limits as appropriate
  2. Use the operation below.

Map style operations

You can use KV buckets to perform the typical operations you would expect from an immediately consistent key/value store:

Atomic operations used for locking and concurrency control

Limiting size, TTL etc.

You can set limits for your buckets, such as:

Treating the Key Value store as a message stream

Finally, you can even do things that typically can not be done with a Key/Value Store:

Notes

A valid key can contain the following characters: a-z, A-Z, 0-9, _, -, ., = and /, i.e. it can be a dot-separated list of tokens (which means that you can then use wildcards to match hierarchies of keys when watching a bucket). The value can be any byte array.

Key/Value Store Walkthrough

The Key/Value Store is a JetStream feature, so we need to verify it is enabled by

nats account info

which may return

JetStream Account Information:

   JetStream is not supported in this account

In this case, you should enable JetStream.

Prerequisite: enabling JetStream

If you are running a local nats-server stop it and restart it with JetStream enabled using nats-server -js (if that's not already done)

You can then check that JetStream is enabled by using

nats account info

Connection Information:

               Client ID: 6
               Client IP: 127.0.0.1
                     RTT: 64.996µs
       Headers Supported: true
         Maximum Payload: 1.0 MiB
           Connected URL: nats://127.0.0.1:4222
       Connected Address: 127.0.0.1:4222
     Connected Server ID: ND2XVDA4Q363JOIFKJTPZW3ZKZCANH7NJI4EJMFSSPTRXDBFG4M4C34K

JetStream Account Information:

           Memory: 0 B of Unlimited
          Storage: 0 B of Unlimited
          Streams: 0 of Unlimited
        Consumers: 0 of Unlimited

Creating a KV bucket

A 'KV bucket' is like a stream; you need to create it before using it, as in nats kv add <KV Bucket Name>:

nats kv add my-kv

my_kv Key-Value Store Status

         Bucket Name: my-kv
         History Kept: 1
        Values Stored: 0
           Compressed: false
   Backing Store Kind: JetStream
          Bucket Size: 0 B
  Maximum Bucket Size: unlimited
   Maximum Value Size: unlimited
          Maximum Age: unlimited
     JetStream Stream: KV_my-kv
              Storage: File

Storing a value

Now that we have a bucket, we can assign, or 'put', a value to a specific key:

nats kv put my-kv Key1 Value1

which should return the key's value Value1

Getting a value

We can fetch, or 'get', the value for a key "Key1":

nats kv get my-kv Key1

my-kv > Key1 created @ 12 Oct 21 20:08 UTC

Value1

Deleting a value

You can always delete a key and its value by using

nats kv del my-kv Key1

It is harmless to delete a non-existent key (check this!!).

Atomic operations

K/V Stores can also be used in concurrent design patterns, such as semaphores, by using atomic 'create' and 'update' operations.

E.g. a client wanting exclusive use of a file can lock it by creating a key, whose value is the file name, with create and deleting this key after completing use of that file. A client can increase the resilience against failure by using a timeout for the bucket containing this key. The client can use update with a revision number to keep the bucket alive.

Updates can also be used for more fine-grained concurrency control, sometimes known as optimistic locking, where multiple clients can try a task, but only one can successfully complete it.

Create (aka exclusive locking)

Create a lock/semaphore with the create operation.

nats kv create my-sem Semaphore1 Value1

Only one create can succeed. First come, first serve. All concurrent attempts will result in an error until the key is deleted

nats kv create my-sem Semaphore1 Value1
nats: error: nats: wrong last sequence: 1: key exists

Update with CAS (aka optimistic locking)

We can also atomically update, sometimes known as a CAS (compare and swap) operation, a key with an additional parameter revision

nats kv update my-sem Semaphore1 Value2 13

A second attempt with the same revision 13, will fail

nats kv update my-sem Semaphore1 Value2 13
nats: error: nats: wrong last sequence: 14

Watching a K/V Store

An unusual functionality of a K/V Store is being able to 'watch' a bucket, or a specific key in that bucket, and receive real-time updates to changes in the store.

For the example above, run nats kv watch my-kv. This will start a watcher on the bucket we have just created earlier. By default, the KV bucket has a history size of one, and so it only remembers the last change. In our case, the watcher should see a delete of the value associated with the key "Key1":

nats kv watch my-kv

[2021-10-12 13:15:03] DEL my-kv > Key1

If we now concurrently change the value of 'my-kv' by

nats kv put my-kv Key1 Value2

The watcher will see that change:

[2021-10-12 13:25:14] PUT my-kv > Key1: Value2

Cleaning up

When you are finished using a bucket, you can delete the bucket, and its resources, by using the rm operator:

nats kv rm my-kv

Object Store

JetStream, the persistence layer of NATS, not only allows for the higher qualities of service and features associated with 'streaming', but it also enables some functionalities not found in messaging systems.

One such feature is the Object store functionality, which allows client applications to create buckets (corresponding to streams) that can store a set of files. Files are stored and transmitted in chunks, allowing files of arbitrary size to be transferred safely over the NATS infrastructure.

Note: Object store is not a distributed storage system. All files in a bucket will need to fit on the target file system.

Basic Capabilities

The Object Store implements a chunking mechanism, allowing you to for example store and retrieve files (i.e. the object) of any size by associating them with a path or file name as the key.

Advanced Capabilities

Object Store Walkthrough

If you are running a local nats-server stop it and restart it with JetStream enabled using nats-server -js (if that's not already done)

You can then check that JetStream is enabled by using

nats account info

Which should output something like:

Connection Information:

               Client ID: 6
               Client IP: 127.0.0.1
                     RTT: 64.996µs
       Headers Supported: true
         Maximum Payload: 1.0 MiB
           Connected URL: nats://127.0.0.1:4222
       Connected Address: 127.0.0.1:4222
     Connected Server ID: ND2XVDA4Q363JOIFKJTPZW3ZKZCANH7NJI4EJMFSSPTRXDBFG4M4C34K

JetStream Account Information:

           Memory: 0 B of Unlimited
          Storage: 0 B of Unlimited
          Streams: 0 of Unlimited
        Consumers: 0 of Unlimited

If you see the below instead then JetStream is not enabled

JetStream Account Information:

   JetStream is not supported in this account

Creating an Object Store bucket

Just like you need to create streams before you can use them you need to first create an Object Store bucket

nats object add myobjbucket

which outputs

myobjbucket Object Store Status

         Bucket Name: myobjbucket
            Replicas: 1
                 TTL: unlimitd
              Sealed: false
                Size: 0 B
  Backing Store Kind: JetStream
    JetStream Stream: OBJ_myobjbucket

Putting a file in the bucket

nats object put myobjbucket ~/Movies/NATS-logo.mov

1.5 GiB / 1.5 GiB [====================================================================================]

Object information for myobjbucket > /Users/jnmoyne/Movies/NATS-logo.mov

               Size: 1.5 GiB
  Modification Time: 14 Apr 22 00:34 +0000
             Chunks: 12,656
             Digest: sha-256 8ee0679dd1462de393d81a3032d71f43d2bc89c0c8a557687cfe2787e926

Putting a file in the bucket by providing a name

By default the full file path is used as a key. Provide the key explicitly (e.g. a relative path ) with --name

nats object put --name /Movies/NATS-logo.mov myobjbucket ~/Movies/NATS-logo.mov

1.5 GiB / 1.5 GiB [====================================================================================]

Object information for myobjbucket > /Movies/NATS-logo.mov

               Size: 1.5 GiB
  Modification Time: 14 Apr 22 00:34 +0000
             Chunks: 12,656
             Digest: sha-256 8ee0679dd1462de393d81a3032d71f43d2bc89c0c8a557687cfe2787e926

Listing the objects in a bucket

nats object ls myobjbucket

╭───────────────────────────────────────────────────────────────────────────╮
│                              Bucket Contents                              │
├─────────────────────────────────────┬─────────┬───────────────────────────┤
│ Name                                │ Size    │ Time                      │
├─────────────────────────────────────┼─────────┼───────────────────────────┤
│ /Users/jnmoyne/Movies/NATS-logo.mov │ 1.5 GiB │ 2022-04-13T17:34:55-07:00 │
│ /Movies/NATS-logo.mov               │ 1.5 GiB │ 2022-04-13T17:35:41-07:00 │
╰─────────────────────────────────────┴─────────┴───────────────────────────╯

Getting an object from the bucket

nats object get myobjbucket ~/Movies/NATS-logo.mov

1.5 GiB / 1.5 GiB [====================================================================================]

Wrote: 1.5 GiB to /Users/jnmoyne/NATS-logo.mov in 5.68s average 279 MiB/s

Getting an object from the bucket with a specific output path

By default, the file will be stored relative to the local path under its name (not the full path). To specify an output path use --output

nats object get myobjbucket --output /temp/Movies/NATS-logo.mov /Movies/NATS-logo.mov

1.5 GiB / 1.5 GiB [====================================================================================]

Wrote: 1.5 GiB to /temp/Movies/NATS-logo.mov in 5.68s average 279 MiB/s

Removing an object from the bucket

nats object rm myobjbucket ~/Movies/NATS-logo.mov

? Delete 1.5 GiB byte file myobjbucket > /Users/jnmoyne/Movies/NATS-logo.mov? Yes
Removed myobjbucket > /Users/jnmoyne/Movies/NATS-logo.mov
myobjbucket Object Store Status

         Bucket Name: myobjbucket
            Replicas: 1
                 TTL: unlimitd
              Sealed: false
                Size: 16 MiB
  Backing Store Kind: JetStream
    JetStream Stream: OBJ_myobjbucket

Getting information about the bucket

nats object info myobjbucket

myobjbucket Object Store Status

         Bucket Name: myobjbucket
            Replicas: 1
                 TTL: unlimitd
              Sealed: false
                Size: 1.6 GiB
  Backing Store Kind: JetStream
    JetStream Stream: OBJ_myobjbucket

Watching for changes to a bucket

nats object watch myobjbucket

[2022-04-13 17:51:28] PUT myobjbucket > /Users/jnmoyne/Movies/NATS-logo.mov: 1.5 GiB bytes in 12,656 chunks
[2022-04-13 17:53:27] DEL myobjbucket > /Users/jnmoyne/Movies/NATS-logo.mov

Sealing a bucket

You can seal a bucket, meaning that no further changes are allowed on that bucket

nats object seal myobjbucket

? Really seal Bucket myobjbucket, sealed buckets can not be unsealed or modified Yes
myobjbucket has been sealed
myobjbucket Object Store Status

         Bucket Name: myobjbucket
            Replicas: 1
                 TTL: unlimitd
              Sealed: true
                Size: 1.6 GiB
  Backing Store Kind: JetStream
    JetStream Stream: OBJ_myobjbucket

Deleting a bucket

Using nats object rm myobjbucket will delete the bucket and all the files stored in it.

Headers

Message headers are used in a variety of JetStream contexts, such de-duplication, auto-purging of messages, metadata from republished messages, and more.

Nats- is a reserved namespace. Please use a different prefix for your own headers. This list may not be complete. Additional headers may be used for API internal messages or messages used for monitoring and control.

Publish

Headers that can be set by a client when a message being published. These headers are recognized by the server.

Name Description Example Version
Nats-Msg-Id Client-defined unique identifier for a message that will be used by the server apply de-duplication within the configured Duplicate Window. 9f01ccf0-8c34-4789-8688-231a2538a98b 2.2.0
Nats-Expected-Stream Used to assert the published message is received by some expected stream. my-stream 2.2.0
Nats-Expected-Last-Msg-Id Used to apply optimistic concurrency control at the stream-level. The value is the last expected Nats-Msg-Id and the server will reject a publish if the current ID does not match. 9f01ccf0-8c34-4789-8688-231a2538a98b 2.2.0
Nats-Expected-Last-Sequence Used to apply optimistic concurrency control at the stream-level. The value is the last expected sequence and the server will reject a publish if the current sequence does not match. 328 2.2.0
Nats-Expected-Last-Subject-Sequence Used to apply optimistic concurrency control at the subject-level. The value is the last expected sequence and the server will reject a publish if the current sequence does not match for the message's subject. 38 2.3.1
Nats-Expected-Last-Subject-Sequence-Subject A subject which may include wildcards. Used with Nats-Expected-Last-Subject-Sequence. Server will enforce last sequence against the given subject rather than the one being published. events.orders.1.> 2.11.0
Nats-Rollup Used to apply a purge of all prior messages in a stream or at the subject-level. The rollup message will stay in the stream. Requires the allow rollups to be set on the stream. all purges the full stream, sub purges the subject on which this messages was sent. Wildcards subjects are not allowed and will result in undefined behavior. 2.6.2
Nats-TTL Used to set a per message TTL. Requires the per message ttl flag to be set on the stream. 1h, 10s (go duration string format) 2.11

RePublish or direct get

When messages are being re-published (must be configured in stream settings) from a stream or retrieved with a direct get operation from stream these headers are being set.

Do not set these headers on client published messages.

Name Description Example Version
Nats-Stream Name of the stream the message was republished from. Nats-Stream: my-stream 2.8.3
Nats-Subject The original subject of the message. events.mouse_clicked 2.8.3
Nats-Sequence The original sequence of the message. 193 2.8.3
Nats-Last-Sequence The last sequence of the message having the same subject, otherwise zero if this is the first message for the subject. 190 2.8.3
Nats-Time-Stamp The original timestamp of the message. 2023-08-23T19:53:05.762416Z 2.10.0
Nats-Num-Pending Number of messages pending in the multi/batched get response. or details see: ADR-31 5 2.10.0
Nats-UpTo-Sequence On the last messages of multi/batched get response. The up-to-seq value of the original request. Helps the client to continue incomplete batch requests. For details see: ADR-31 2.11.0

Sources

Headers that are implicitly added to messages sourced from other streams.

The format of the header content may change in the future. Please parse conservatively and assume that additional fields may be added or that older nats-server version have fewer fields.

Name Description Example Version
Nats-Stream-Source Contains space delimited:
- Origin stream name (disambiguated with domain hash if cross domain sourced)
- The original sequence number
- The list of subject filters
- The list of destination transforms
- The original subject
 ORDERS:vSF0ECo6 17 foo.* bar.$1 foo.abc 2.2.0

Key-Value-Store

The KV-store API adds marker headers for some delete and purge operations.

Name Description Example Version
KV-Operation Marks a delete or a purge. Neither has immediate functional impact on the stream. The delete marker marks the key as delete even if the version history is kept intact. The purge MUST be combined with Nats-Rollup to actually purge the subject.
Since 2.14 PURGE can also be combined with Nats-TTL to auto delete the purge marker and bypass the need for KV nats kv compact		 DEL or PURGE 2.2.0

Tracing

When tracing is activated every subsystem that touches a message will produce Trace Events. These Events are aggregated per server and published to a destination subject.

Note that two variants exist. traceparent as per the trace context standard and ad hoc tracing through Nats-Trace-Dest.

Introduced in version 2.11 - see ADR-41

Name Description Example Version
traceparent Triggers tracing as per the https://www.w3.org/TR/trace-context/ standard. Requires tracing destination and sampling rate to be configured in the msg_trace section on the account level. See Tracing open telemetry style trace-ID-1234 2.11
Nats-Trace-Dest The subject that will receive the Trace messages trace.receiver.all 2.11
Nats-Trace-Only Optional. Defaults to false. Set to true to skip message delivery. If true only traces will be produced, but the messages is not sent to a subscribing client or stored in JetStream. true 2.11
Accept-Encoding Optional. Enables compression of the payload of the trace messages. gzip, snappy 2.11
Nats-Trace-Hop Internal. Do not set. Set by the server to count hops. <hop count> 2.11
Nats-Trace-Origin-Account Internal. Do not set. Set by the server when an account boundary is crossed <account name> 2.11

Scheduler

Message scheduling in streams. Needs to be enabled on the stream with "allow schedules" flag.

Introduced in version 2.12 with extensions in 2.14 - see ADR-51

Name Description Example Version
Nats-Schedule When the message will be sent to the target subject. Several formats are suppported: A timestamp for sending a messages once. Crontab format for repeated messages and simple alias for common crontab formats. 0 30 14 * * *, @hourly, @daily, @every 5m, @at 2009-11-10T23:00:00Z (RFC3339 format) 2.12 (repeating schedules: 2.14)
Nats-Schedule-TTL Optional. The TTL to be set on the final message on the target subject. 1h, 10s (valid go duration string) 2.12
Nats-Schedule-Target The target subject the final message will be sent to. Note that this must be distinct from the scheduling subject the message arrived in. orders 2.12
Nats-Schedule-Rollup Optional. The rollup to be set on the final message on the target subject. all, sub 2.14
Nats-Schedule-Source Optional. Instructs the schedule to read the last message on the given subject and publish it. If the Subject is empty, the schedule message's content is the same as the schedule's content. Wildcards are not supported. orders.customer_acme 2.14
Nats-Schedule-Time-Zone Optional. The time zone used for the Cron schedule. If not specified, the Cron schedule will be in UTC. Not allowed to be used if the schedule is not a Cron schedule. CET 2.14

The final scheduled message will contain the following headers.

Name Description Example Version
Nats-Scheduler The subject holding the schedule orders.schedule.1234 2.12
Nats-Schedule-Next Timestamp for next invocation for cron schedule messages or purge for delayed messages 2009-11-10T23:00:00Z, purge 2.12
Nats-TTL The TTL value when Nats-Schedule-TTL was set 1h, 10s 2.12
Nats-Rollup The rollup value when Nats-Schedule-Rollup was set all, sub 2.14

Atomic batch publish

Introduced in version 2.12 with optimizations in 2.14 - see ADR-50

Atomic batch publish will use the following headers. A client may reconstruct a batch using the headers below.

Name Description Example Version
Nats-Batch-Id Unique identifier for the batch. <uuid> (<=64 characters) 2.12
Nats-Batch-Sequence Monotonically increasing sequence number, starting at 1 1, 2 2.12
Nats-Batch-Commit Only on last message. 1 commit the batch including this message. eob commit the batch excluding this message. Any other value will terminate the batch. 1, eob 2.12 (eob: 2.14)

Fast-ingest batch publishing

Introduced in version 2.14 - see ADR-50

Note that fast-ingest is mentioned for completeness sake. Unlike atomic batches, fast-ingest is not controlled by headers, but rather by reply-subject format (ending with .$FI). All common headers can be combined with fast-ingest (e.g. de-duplication - see ), but it MUST NOT be combined with atomic batch publish.

Name Description Example Version
N/A N/A N/A 2.14

Internal

Headers used internally by API clients and the server. Should not be set by user.

This is list is not exhaustive. Headers used in error replies may not be documented.

Name Description Example Version
Nats-Required-Api-Level Optional. The required API level for the JetStream request. Servers from version 2.11 will return an error if larger than the support API level. 2 (Integer value) 2.12
Nats-Request-Info When messages cross account boundaries a header with origin information (account, user etc) may be added. 2.2.0
Nats-Marker-Reason When messages are removed from a KV where subject delete markers are supported, a delete marker will be placed. And notifications are sent to interested watchers. The message payload is empty and the removal reason is indicated through this header. See ADR-48 MaxAge, Remove, Purge 2.11
Nats-Incr Used in KV stores to atomically increment counter. Any valid integer (including 0) starting with a sign.. See ADR-49 +1, +42, -1, +0 2.12
Nats-Counter-Sources Tracking Nats-Incr when messages are sourced. For details see: ADR-49 2.12

Mirror

Headers used for internal flow-control messages for a mirror.

This is for information only and may change without notice.

Name Description Example Version
Nats-Last-Consumer 2.2.1
Nats-Last-Stream 2.2.1
Nats-Consumer-Stalled 2.4.0
Nats-Response-Type 2.6.4

Subject Mapping and Partitioning

Subject mapping and transforms is a powerful feature of the NATS server. Transformations (we will use mapping and transform interchangeably) apply to various situations when messages are generated and ingested, acting as translations and in some scenarios as filters.

注意

Mapping and transforms is an advanced topic. Before proceeding, please ensure you understand NATS concepts like clusters, accounts and streams.

Transforms can be defined (for details see below):

Transforms may be used for:

Priority and sequence of operations

mappings: {
	transform.order target.order
	target.order transform.order
}

On a central cluster:

server_name: "hub"
cluster: { name: "hub" }
mappings: {
	orders.* orders.central.{{wildcard(1)}}
}

OR

server_name: "hub"
cluster: { name: "hub" }
mappings: {
	orders.> orders.central.>}
}

On a leaf cluster

server_name: "store1"
cluster: { name: "store1" }
mappings: {
	orders.central.* orders.local.{{wildcard(1)}}
}

A stream config on the leaf cluster

{
  "name": "orders",
  "subjects": [ "orders.local.*"],
  "subject_transform":{"src":"orders.local.*","dest":"orders.{{wildcard(1)}}"},
  "retention": "limits",
  ...
  "republish": {
    "src": "orders.*",
    "dest": "orders.trace.{{wildcard(1)}}"
  },

Security

When using config file-based account management (not using JWT security), you can define the core NATS account level subject transforms in server configuration files, and simply need to reload the configuration whenever you change a transform for the change to take effect.

When using operator JWT security (distributed security) with the built-in resolver you define the transforms and the import/exports in the account JWT, so after modifying them, they will take effect as soon as you push the updated account JWT to the servers.

Testing and debugging

提示

You can easily test individual subject transform rules using the nats CLI tool command nats server mapping. See examples below.

From NATS server 2.11 (and NATS versions published thereafter) the handling of subjects, including mappings can be observed with nats trace

In the example below a message is first disambiguated from orders.device1.order1 -> orders.hub.device1.order1. Then imported into a stream and stored under its original name.

nats trace orders.device1.order1

Tracing message route to subject orders.device1.order1

Client "NATS CLI Version development" cid:16 cluster:"hub" server:"hub" version:"2.11.0-dev"
    Mapping subject:"orders.hub.device1.order1"
--J JetStream action:"stored" stream:"orders" subject:"orders.device1.order1"
--X No active interest

Legend: Client: --C Router: --> Gateway: ==> Leafnode: ~~> JetStream: --J Error: --X

Egress Count:

  JetStream: 1

Simple mappings

The example of foo:bar is straightforward. All messages the server receives on subject foo are remapped and can be received by clients subscribed to bar.

nats server mapping foo bar foo
> bar

When no subject is provided the command will operate in interactive mode:

nats server mapping foo bar
> Enter subjects to test, empty subject terminates.
>
> ? Subject foo
> bar

> ? Subject test
> Error: no matching transforms available

Example server config. Note that the mappings below apply only to the default $G account.

server_name: "hub"
cluster: { name: "hub" }
mappings: {
    orders.flush  orders.central.flush 
	orders.* orders.central.{{wildcard(1)}}
}

Mapping a full wildcard

server_name: "hub"
cluster: { name: "hub" }
mappings: {
    orders.>  orders.central.> 
}

With accounts. While this mapping applies to a specific account.

server_name: "hub"
cluster: { name: "hub" }

accounts {
    accountA: { 
        mappings: {
            orders.flush  orders.central.flush 
        	orders.* orders.central.{{wildcard(1)}}
        }
    }
}

Mapping a full wildcard '>'

A full wildcard token can be used ONCE in source expression and must be present on the destination expression as well exactly once.

Example: Prefixing a subject:

nats server mapping ">"  "baz.>" bar.a.b
> baz.bar.b.a

Subject token reordering

Wildcard tokens may be referenced by position number in the destination mapping using (only for versions 2.8.0 and above of nats-server). Syntax: {{wildcard(position)}}. E.g. {{wildcard(1)}} references the first wildcard token, {{wildcard(2)}} references the second wildcard token, etc..

Example: with this transform "bar.*.*" : "baz.{{wildcard(2)}}.{{wildcard(1)}}", messages that were originally published to bar.a.b are remapped in the server to baz.b.a. Messages arriving at the server on bar.one.two would be mapped to baz.two.one, and so forth. Try it for yourself using nats server mapping.

nats server mapping "bar.*.*"  "baz.{{wildcard(2)}}.{{wildcard(1)}}" bar.a.b
> baz.b.a

提示

An older style deprecated mapping syntax using $1.$2 en lieu of {{wildcard(1)}}.{{wildcard(2)}} may be seen in other examples.

Dropping subject tokens

You can drop tokens from the subject by not using all the wildcard tokens in the destination transform, with the exception of mappings defined as part of import/export between accounts in which case all the wildcard tokens must be used in the transform destination.

nats server mapping "orders.*.*" "foo.{{wildcard(2)}}" orders.local.order1
> orders.order1

提示

Import/export mapping must be mapped bidirectionally unambiguous.

Splitting tokens

There are two ways you can split tokens:

Splitting on a separator

You can split a token on each occurrence of a separator string using the split(separator) transform function.

Examples:

Splitting at an offset

You can split a token in two at a specific location from the start or the end of the token using the SplitFromLeft(wildcard index, offset) and SplitFromRight(wildcard index, offset) transform functions (note that the upper camel case on all subject transform function names is optional you can also use all lowercase function names if you prefer).

Examples:

Slicing tokens

You can slice tokens into multiple parts at a specific interval from the start or the end of the token by using the SliceFromLeft(wildcard index, number of characters) and SliceFromRight(wildcard index, number of characters) mapping functions.

Examples:

Partitioning with random numbers

A simple means of distributing load can be partitioned into random buckets. Use the random(<n>) function to create a random number between 0 and n-1 (module n).

For more details on partitioning, see below in Deterministic subject token partitioning.

nats server mapping "neworders.create" "neworders.{{random(15)}}.create" neworders.create
> neworders.7.create

Deterministic subject token partitioning

Deterministic token partitioning allows you to use subject-based addressing to deterministically divide (partition) a flow of messages where one or more of the subject tokens is mapped into a partition key. Deterministically means, the same tokens are always mapped into the same key. The mapping will appear random and may not be fair for a small number of subjects.

For example: new customer orders are published on neworders.<customer id>, you can partition those messages over 3 partition numbers (buckets), using the partition(number of partitions, wildcard token positions...) function which returns a partition number (between 0 and number of partitions-1) by using the following mapping "neworders.*" : "neworders.{{wildcard(1)}}.{{partition(3,1)}}".

nats server mapping "neworders.*" "neworders.{{wildcard(1)}}.{{partition(3,1)}}" neworders.customerid1
> neworders.customerid1.0

There are 2 variants of the partition function with 1 or more parameters.

| Variant /Example | Explanation | Equivalent to | | --------------------- | ----------------------- | | partition(<buckets>) | Create a hash from the complete subject (concatenated tokens)| partition(<buckets>,1,2,..,n) for a subject with n tokens | | partition(<buckets>,<token position>, ... | Create a hash from a list of tokens specified by position in the subject | N/A |

提示

Note that multiple token positions can be specified to form a kind of composite partition key. For example, a subject with the form foo.*.* can have a partition transform of foo.{{wildcard(1)}}.{{wildcard(2)}}.{{partition(5,1,2)}} which will result in five partitions in the form foo.*.*.<n>, but using the hash of the two wildcard tokens when computing the partition number.

nats server mapping "foo.*.*" "foo.{{wildcard(1)}}.{{wildcard(2)}}.{{partition(5,1,2)}}" foo.us.customerid 
> foo.us.customerid.0

This particular transform means that any message published on neworders.<customer id> will be mapped to neworders.<customer id>.<a partition number 0, 1, or 2>. i.e.:

Published on Mapped to
neworders.customerid1 neworders.customerid1.0
neworders.customerid2 neworders.customerid2.2
neworders.customerid3 neworders.customerid3.1
neworders.customerid4 neworders.customerid4.2
neworders.customerid5 neworders.customerid5.1
neworders.customerid6 neworders.customerid6.0

The transform is deterministic because (as long as the number of partitions is 3) 'customerid1' will always map to the same partition number. The mapping is hash-based, its distribution is random but tends towards 'perfectly balanced' distribution (i.e. the more keys you map the more the number of keys for each partition will tend to converge to the same number).

You can partition on more than one subject wildcard token at a time, e.g.: {{partition(10,1,2)}} distributes the union of token wildcards 1 and 2 over 10 partitions.

Published on Mapped to
foo.1.a foo.1.a.1
foo.1.b foo.1.b.0
foo.2.b foo.2.b.9
foo.2.a foo.2.a.2

What this deterministic partition transform enables is the distribution of the messages that are subscribed to using a single subscriber (on neworders.*) into three separate subscribers (respectively on neworders.*.0, neworders.*.1 and neworders.*.2) that can operate in parallel.

nats server mapping "foo.*.*" "foo.{{wildcard(1)}}.{{wildcard(2)}}.{{partition(3,1,2)}}"

When is deterministic partitioning uselful

The core NATS queue-groups and JetStream durable consumer mechanisms to distribute messages amongst a number of subscribers are partition-less and non-deterministic, meaning that there is no guarantee that two sequential messages published on the same subject are going to be distributed to the same subscriber. While in most use cases a completely dynamic, demand-driven distribution is what you need, it does come at the cost of guaranteed ordering because if two subsequent messages can be sent to two different subscribers which would then both process those messages at the same time at different speeds (or the message has to be re-transmitted, or the network is slow, etc.) and that could result in potential 'out of order' message delivery.

This means that if the application requires strictly ordered message processing, you need to limit distribution of messages to 'one at a time' (per consumer/queue-group, i.e. using the 'max acks pending' setting), which in turn hurts scalability because it means no matter how many workers you have subscribed, only one is doing any processing work at a time.

Being able to evenly split (i.e. partition) subjects in a deterministic manner (meaning that all the messages on a particular subject are always mapped to the same partition) allows you to distribute and scale the processing of messages in a subject stream while still maintaining strict ordering per subject. For example, inserting a partition number as a token in the message subject as part of the stream definition and then using subject filters to create a consumer per partition (or set of partitions).

Another scenario for deterministic partitioning is in the extreme message publication rate scenarios where you are reaching the limits of the throughput of incoming messages into a stream capturing messages using a wildcard subject. This limit can be ultimately reached at very high message rate due to the fact that a single nats-server process is acting as the RAFT leader (coordinator) for any given stream and can therefore become a limiting factor. In that case, distributing (i.e. partitioning) that stream into a number of smaller streams (each one with its own RAFT leader and therefore all these RAFT leaders are spread over all of the JetStream-enabled nats-servers in the cluster rather than a single one) in order to scale.

Yet another use case where deterministic partitioning can help is if you want to leverage local data caching of data (context or potentially heavy historical data for example) that the subscribing process need to access as part of the processing of the messages.

Weighted mappings

Traffic can be split by percentage from one subject transform to multiple subject transforms.

For A/B testing or canary releases

Here's an example for canary deployments, starting with version 1 of your service.

Applications would make requests of a service at myservice.requests. The responders doing the work of the server would subscribe to myservice.requests.v1. Your configuration would look like this:

  myservice.requests: [
    { destination: myservice.requests.v1, weight: 100% }
  ]

All requests to myservice.requests will go to version 1 of your service.

When version 2 comes along, you'll want to test it with a canary deployment. Version 2 would subscribe to myservice.requests.v2. Launch instances of your service.

Update the configuration file to redirect some portion of the requests made to myservice.requests to version 2 of your service.

For example the configuration below means 98% of the requests will be sent to version 1 and 2% to version 2.

    myservice.requests: [
        { destination: myservice.requests.v1, weight: 98% },
        { destination: myservice.requests.v2, weight: 2% }
    ]

Once you've determined Version 2 is stable you can switch 100% of the traffic over to it and you can then shut down the version 1 instance of your service.

For traffic shaping in testing

Traffic shaping is also useful in testing. You might have a service that runs in QA that simulates failure scenarios which could receive 20% of the traffic to test the service requestor.

myservice.requests.*: [{ destination: myservice.requests.{{wildcard(1)}}, weight: 80% }, { destination: myservice.requests.fail.{{wildcard(1)}}, weight: 20% }

For artificial loss

Alternatively, introduce loss into your system for chaos testing by mapping a percentage of traffic to the same subject. In this drastic example, 50% of the traffic published to foo.loss.a would be artificially dropped by the server.

foo.loss.>: [ { destination: foo.loss.>, weight: 50% } ]

You can both split and introduce loss for testing. Here, 90% of requests would go to your service, 8% would go to a service simulating failure conditions, and the unaccounted for 2% would simulate message loss.

myservice.requests: [{ destination: myservice.requests.v3, weight: 90% }, { destination: myservice.requests.v3.fail, weight: 8% }] the remaining 2% is "lost"

Cluster scoped mappings

If you are running a super-cluster you can define transforms that apply only to messages being published from a specific cluster.

For example if you have 3 clusters named east central and west and you want to map messages published on foo in the east cluster to foo.east, those published in the central cluster to foo.central and so on for west you can do so by using the cluster keyword in the mapping source and destination.

mappings = {
        "foo":[
               {destination:"foo.west", weight: 100%, cluster: "west"},
               {destination:"foo.central", weight: 100%, cluster: "central"},
               {destination:"foo.east", weight: 100%, cluster: "east"}
        ]
}

This means that the application can be portable in terms of deployment and does not need to be configured with the name of the cluster it happens to be connected in order to compose the subject: it just publishes to foo and the server will map it to the appropriate subject based on the cluster it's running in.

Subject mapping and transforms in streams

You can define subject mapping transforms as part of the stream configuration.

Transforms can be applied in multiple places in the stream configuration:

Note that when used in Mirror, Sources or Republish, the subject transforms are filters with optional transformation, while when used in the Stream configuration it only transforms the subjects of the matching messages and does not act as a filter.

{
  "name": "orders",
  "subjects": [ "orders.local.*"],
  "subject_transform":{"src":"orders.local.*","dest":"orders.{{wildcard(1)}}"},
  "retention": "limits",
  ...
  "sources": [
    {
      "name": "other_orders",
      "subject_transforms": [
        {
          "src": "orders.online.*",
          "dest": "orders.{{wildcard(1)}}"
        }
      ]
    }
  ],
  "republish": {
    "src": "orders.*",
    "dest": "orders.trace.{{wildcard(1)}}"
  }
    
}

提示

For sources and republish transforms the src expression will act as a filter. Non-matching subjects will be ignored.

For the stream level subject_transform non-matching subjects will stay untouched.

NATS Service Infrastructure

NATS is a client/server system in the fact that you have 'NATS client applications' (applications using one of the NATS client libraries) that connect to 'NATS servers' that provide the NATS service. The NATS servers work together to provide a NATS service infrastructure to their client applications.

NATS is extremely flexible and scalable and allows the service infrastructure to be as small as a single process running locally on your local machine and as large as an 'Internet of NATS' of Leaf Nodes, and Leaf Node clusters all interconnected in a secure way over a global shared NATS super-cluster.

Regardless of the size and complexity of the NATS service infrastructure being used, the only configuration needed by the client applications being the location (NATS URLs) of one or more NATS servers and depending on the required security, their credentials.

Note that if your application is written in Golang then you even have the option of embedding the NATS server functionality into the application itself (however you need to then configure your application instances with nats-server configuration information).

You do not actually need to run your NATS service infrastructure, instead you can instead make use of a public NATS infrastructure offered by a NATS Service Provider such as Synadia Cloud, think of Synadia Cloud as being an 'Internet of NATS' (literally an "InterNATS") and of Synadia as being an "InterNATS Service Provider".

The Evolution of your NATS service infrastructure

You will typically start by running a single instance of nats-server on your local development machine, and have your applications connect to it while you do your application development and local testing.

Next you will probably want to start testing and running those applications and servers in a VPC, or a region or in some on-prem location, so you will deploy either single NATS server or clusters of NATS servers in your VPCs/regions/on-prem/etc... locations and in each location have the applications connect their local nats-server or nats-server cluster. You can then connect those local nats-servers or local nats-server clusters together by making them leaf nodes connecting to a 'backbone' cluster or super-cluster, or by connecting them directly together via gateway connections.

If you have many client applications (e.g., applications deployed on end-user devices all over the Internet, or for example many IoT devices) or many servers in a lot of locations you will then scale your NATS service infrastructure by deploying clusters of NATS servers in multiple locations and multiple cloud providers and VPCs. You will then need to connect those clusters into a global super-cluster and then devise a scheme to intelligently direct your client applications to the right 'closest' NATS server cluster.

Running your own NATS service infrastructure

You can deploy and run your own NATS service infrastructure of nats-server instances, composed of servers, clusters of servers, super-cluster and leaf node NATS servers.

Virtualization and containerization considerations

If using Kubernetes we recommend you use the Helm charts.

NATS Adaptive Deployment Architectures

From a single process to a global super-cluster with leaf node servers, you can always adapt your NATS service deployment to your needs. From servers and VPCs in many clouds, to partially connected small edge devices and everything in between, you can always easily extend and scale your NATS service as your needs grow.

A single server

The simplest version of a NATS service infrastructure is a single nats-server process. The nats-server binary is highly optimized, very lightweight and extremely efficient in its resources' usage.

Client applications establish a connection to the URL of that nats-server process (e.g. "nats://localhost").

A cluster of servers

If you need a fault-tolerant NATS service or if you need to scale your service capacity, you can cluster a set of nats-server processes together in a cluster.

Client applications establish and maintain a connection to (one of) the nats server URL(s) composing the cluster (e.g. "nats://server1","nats://server2",...).

A super-cluster

You can go further than a single cluster and have disaster recovery and get global deployments (e.g. on multiple locations or regions, multiple VPCs or multiple Cloud providers) by deploying multiple clusters and connecting them together via gateway connections (which are interest pruned).

Client applications establish a connection to (one of) the nats server URL(s) of one of the clusters (e.g. "nats://us-west-1.company.com","nats://us-west-2.company.com",...).

With Leaf Nodes

You can easily 'extend' the NATS service provided by a cluster or super-cluster by deploying 'locally' one or more leaf node nats servers that proxy and route traffic between their client applications and the NATS service infrastructure. The context of 'locality' in this case is not just physical: it could mean a location, an edge device or a single development machine, but it could also service a VPC, a group of server processes for a specific application or different accounts, or even a business unit. Leaf node NATS servers can be configured to connect to their cluster over a WebSocket connection (rather than TLS or plain TCP).

Leaf nodes appear to the cluster as a single account connection. Leaf nodes can provide continuous NATS service for their clients, even while being temporarily disconnected from the cluster(s). You can even enable JetStream on the leaf nodes in order to create local streams that are mirrored (mirroring is store and forward and therefore can recover from connectivity outages) to global streams in the upstream cluster(s).

Client applications are configured with the URLs of their 'local' leaf node server(s) and establish a connection to (one of) the leaf node server(s) (e.g. "nats://leaf-node-1","nats://leaf-node-2",...).

See Also

NATS Clusters

https://youtu.be/srARy0m9SdI Clusters

NATS Super-clusters

https://youtu.be/6O_sNSJ2p70 Super-clusters

NATS Leaf Nodes

https://youtu.be/WH55czo1BNk Leaf Nodes

NATS Service Geo-affinity in Queues

https://youtu.be/jLTVhP08Tq0?t=190 Geo-affinity in Queues

Security

NATS has a lot of security features:

You can use accounts for multi-tenancy: each account has its own independent 'subject namespace' and you control the import/export of both streams of messages and services between accounts, and any number of users that client applications can be authenticated as. The subjects or subject wildcards that a user is allowed to publish and/or subscribe to can be controlled either through server configuration or as part of signed JWTs.

JWT authentication/authorization administration is decentralized because each account private key holder can manage their users and their authorizations on their own, without the need for any configuration change on the NATS servers by minting their own JWTs and distributing them to the users. There is no need for the NATS server to ever store any user private keys as they only need to validate the signature chain of trust contained in the user JWT presented by the client application to validate that they have the proper public key for that user.

The JetStream persistence layer of NATS also provides encryption at rest.

Connectivity

NATS supports several kinds of connectivity directly to the NATS servers.

There is also a number of adapters available to bridge traffic to and from other messaging systems