JetStream

Detailed documentation on the NATS JetStream component

Component format

To set up JetStream pub/sub, create a component of type pubsub.jetstream. See the pub/sub broker component file to learn how ConsumerID is automatically generated. Read the How-to: Publish and Subscribe guide on how to create and apply a pub/sub configuration.

  1. apiVersion: dapr.io/v1alpha1
  2. kind: Component
  3. metadata:
  4.   name: jetstream-pubsub
  5. spec:
  6.   type: pubsub.jetstream
  7.   version: v1
  8.   metadata:
  9.   - name: natsURL
  10.     value: "nats://localhost:4222"
  11.   - name: jwt # Optional. Used for decentralized JWT authentication.
  12.     value: "eyJhbGciOiJ...6yJV_adQssw5c"
  13.   - name: seedKey # Optional. Used for decentralized JWT authentication.
  14.     value: "SUACS34K232O...5Z3POU7BNIL4Y"
  15.   - name: tls_client_cert # Optional. Used for TLS Client authentication.
  16.     value: "/path/to/tls.crt"
  17.   - name: tls_client_key # Optional. Used for TLS Client authentication.
  18.     value: "/path/to/tls.key"
  19.   - name: token # Optional. Used for token based authentication.
  20.     value: "my-token"
  21.   - name: name
  22.     value: "my-conn-name"
  23.   - name: streamName
  24.     value: "my-stream"
  25.   - name: durableName 
  26.     value: "my-durable-subscription"
  27.   - name: queueGroupName
  28.     value: "my-queue-group"
  29.   - name: startSequence
  30.     value: 1
  31.   - name: startTime # In Unix format
  32.     value: 1630349391
  33.   - name: flowControl
  34.     value: false
  35.   - name: ackWait
  36.     value: 10s
  37.   - name: maxDeliver
  38.     value: 5
  39.   - name: backOff
  40.     value: "50ms, 1s, 10s"
  41.   - name: maxAckPending
  42.     value: 5000
  43.   - name: replicas
  44.     value: 1
  45.   - name: memoryStorage
  46.     value: false
  47.   - name: rateLimit
  48.     value: 1024
  49.   - name: heartbeat
  50.     value: 15s
  51.   - name: ackPolicy
  52.     value: explicit
  53.   - name: deliverPolicy
  54.     value: all
  55.   - name: domain
  56.     value: hub
  57.   - name: apiPrefix
  58.     value: PREFIX

Spec metadata fields

FieldRequiredDetailsExample
natsURLYNATS server address URL“nats://localhost:4222”
jwtNNATS decentralized authentication JWT“eyJhbGciOiJ…6yJV_adQssw5c”
seedKeyNNATS decentralized authentication seed key“SUACS34K232O…5Z3POU7BNIL4Y”
tls_client_certNNATS TLS Client Authentication Certificate“/path/to/tls.crt”
tls_client_keyNNATS TLS Client Authentication Key“/path/to/tls.key”
tokenNNATS token based authentication“my-token”
nameNNATS connection name“my-conn-name”
streamNameNName of the JetStream Stream to bind to“my-stream”
durableNameNDurable name“my-durable”
queueGroupNameNQueue group name“my-queue”
startSequenceNStart Sequence1
startTimeNStart Time in Unix format1630349391
flowControlNFlow Controltrue
ackWaitNAck Wait10s
maxDeliverNMax Deliver15
backOffNBackOff“50ms, 1s, 5s, 10s”
maxAckPendingNMax Ack Pending5000
replicasNReplicas3
memoryStorageNMemory Storagefalse
rateLimitNRate Limit1024
heartbeatNHeartbeat10s
ackPolicyNAck Policyexplicit
deliverPolicyNOne of: all, last, new, sequence, timeall
domainN[JetStream Leafondes]HUB
apiPrefixN[JetStream Leafnodes]PREFIX

Create a NATS server

You can run a NATS Server with JetStream enabled locally using Docker:

  1. docker run -d -p 4222:4222 nats:latest -js

You can then interact with the server using the client port: localhost:4222.

Install NATS JetStream on Kubernetes by using the helm:

  1. helm repo add nats https://nats-io.github.io/k8s/helm/charts/
  2. helm install --set nats.jetstream.enabled=true my-nats nats/nats

This installs a single NATS server into the default namespace. To interact with NATS, find the service with:

  1. kubectl get svc my-nats

For more information on helm chart settings, see the Helm chart documentation.

Create JetStream

It is essential to create a NATS JetStream for a specific subject. For example, for a NATS server running locally use:

  1. nats -s localhost:4222 stream add myStream --subjects mySubject

Example: Competing consumers pattern

Let’s say you’d like each message to be processed by only one application or pod with the same app-id. Typically, the consumerID metadata spec helps you define competing consumers.

Since consumerID is not supported in NATS JetStream, you need to specify durableName and queueGroupName to achieve the competing consumers pattern. 例如：

  1. apiVersion: dapr.io/v1alpha1
  2. kind: Component
  3. metadata:
  4.   name: pubsub
  5. spec:
  6.   type: pubsub.jetstream
  7.   version: v1
  8.   metadata:
  9.   - name: name
  10.     value: "my-conn-name"
  11.   - name: streamName
  12.     value: "my-stream"
  13.   - name: durableName 
  14.     value: "my-durable-subscription"
  15.   - name: queueGroupName
  16.     value: "my-queue-group"