Redis Streams

Introduction to Redis streams

A Redis stream is a data structure that acts like an append-only log but also implements several operations to overcome some of the limits of a typical append-only log. These include random access in O(1) time and complex consumption strategies, such as consumer groups. You can use streams to record and simultaneously syndicate events in real time. Examples of Redis stream use cases include:

• Event sourcing (e.g., tracking user actions, clicks, etc.)
• Sensor monitoring (e.g., readings from devices in the field)
• Notifications (e.g., storing a record of each user's notifications in a separate stream)

Redis generates a unique ID for each stream entry. You can use these IDs to retrieve their associated entries later or to read and process all subsequent entries in the stream. Note that because these IDs are related to time, the ones shown here may vary and will be different from the IDs you see in your own Redis instance.

Redis streams support several trimming strategies (to prevent streams from growing unbounded) and more than one consumption strategy (see XREAD, XREADGROUP, and XRANGE). Starting with Redis 8.2, the XACKDEL, XDELEX, XADD, and XTRIM commands provide fine-grained control over how stream operations interact with multiple consumer groups, simplifying the coordination of message processing across different applications.

Beginning with Redis 8.6, Redis streams support idempotent message processing (at-most-once production) to prevent duplicate entries when using at-least-once delivery patterns. This feature enables reliable message submission with automatic deduplication. See Idempotent Message Processing for more information.

Examples

• When our racers pass a checkpoint, we add a stream entry for each racer that includes the racer's name, speed, position, and location ID:

  Language: >_ Redis CLI Python JavaScript (node-redis) Java-Sync Go C#-Sync (SE.Redis) Ruby Rust-Sync Rust-Async

  Show full output

  Copied!

  Copied!

  Foundational: Add entries to a stream using XADD with auto-generated IDs (creates new entries with field-value pairs)

  > XADD race:france * rider Castilla speed 30.2 position 1 location_id 1
  "1692632086370-0"
  > XADD race:france * rider Norem speed 28.8 position 3 location_id 1
  "1692632094485-0"
  > XADD race:france * rider Prickett speed 29.7 position 2 location_id 1
  "1692632102976-0"

  ▼ Commands: XADD

  • XADD ( @write , @stream , @fast )

    Appends a new message to a stream. Creates the key if it doesn't exist.

  Redis CLI guide

  Also, check out our other client tools Redis Insight and Redis for VS Code.

  res1 = r.xadd(
      "race:france",
      {"rider": "Castilla", "speed": 30.2, "position": 1, "location_id": 1},
  )
  print(res1)  # >>> 1692629576966-0
  
  res2 = r.xadd(
      "race:france",
      {"rider": "Norem", "speed": 28.8, "position": 3, "location_id": 1},
  )
  print(res2)  # >>> 1692629594113-0
  
  res3 = r.xadd(
      "race:france",
      {"rider": "Prickett", "speed": 29.7, "position": 2, "location_id": 1},
  )
  print(res3)  # >>> 1692629613374-0
  

  ▼ Commands: XADD

  • XADD ( @write , @stream , @fast )

    Appends a new message to a stream. Creates the key if it doesn't exist.

    ▶ Method

    • xadd(
      • name: KeyT,
      • fields: Dict[FieldT, EncodableT],
      • id: StreamIdT = "*",
      • maxlen: Optional[int] = None,
      • approximate: bool = True,
      • nomkstream: bool = False,
      • minid: Union[StreamIdT, None] = None,
      • limit: Optional[int] = None,
      • ref_policy: Optional[Literal["KEEPREF", "DELREF", "ACKED"]] = None,
      • idmpauto: Optional[str] = None,
      • idmp: Optional[tuple[str, bytes]] = None
      ) → ResponseT

  Python Quick-Start

  const res1 = await client.xAdd(
    'race:france', '*', {
      'rider': 'Castilla',
      'speed': '30.2',
      'position': '1',
      'location_id': '1'
    }
  );
  console.log(res1); // >>> 1700073067968-0 N.B. actual values will differ from these examples
  
  const res2 = await client.xAdd(
    'race:france', '*', {
      'rider': 'Norem',
      'speed': '28.8',
      'position': '3',
      'location_id': '1'
    },
  );
  console.log(res2); // >>> 1692629594113-0
  
  const res3 = await client.xAdd(
    'race:france', '*', {
      'rider': 'Prickett',
      'speed': '29.7',
      'position': '2',
      'location_id': '1'
    },
  );
  console.log(res3); // >>> 1692629613374-0
  

  ▼ Commands: XADD

  • XADD ( @write , @stream , @fast )

    Appends a new message to a stream. Creates the key if it doesn't exist.

    ▶ Method

    • XADD(
      • ...args: Tail<Parameters<typeof parseXAddArguments>>
      ) → Any

  Node.js Quick-Start

      StreamEntryID res1 = jedis.xadd("race:france",new HashMap<String,String>(){{put("rider","Castilla");put("speed","30.2");put("position","1");put("location_id","1");}} , XAddParams.xAddParams());
  
      System.out.println(res1); // >>> 1701760582225-0
  
      StreamEntryID res2 = jedis.xadd("race:france",new HashMap<String,String>(){{put("rider","Norem");put("speed","28.8");put("position","3");put("location_id","1");}} , XAddParams.xAddParams());
  
      System.out.println(res2); // >>> 1701760582225-1
  
      StreamEntryID res3 = jedis.xadd("race:france",new HashMap<String,String>(){{put("rider","Prickett");put("speed","29.7");put("position","2");put("location_id","1");}} , XAddParams.xAddParams());
  
      System.out.println(res3); // >>> 1701760582226-0
  

  ▼ Commands: XADD

  • XADD ( @write , @stream , @fast )

    Appends a new message to a stream. Creates the key if it doesn't exist.

    ▶ Methods

    • xadd(
      • key: String,
      • id: StreamEntryID,
      • hash: Map<String, String>
      ) → StreamEntryID // the ID of the added entry
    • xadd(
      • key: String,
      • params: XAddParams,
      • hash: Map<String, String>
      ) → StreamEntryID // the ID of the added entry
    • xadd(
      • key: String,
      • id: StreamEntryID,
      • hash: Map<String, String>
      ) → StreamEntryID // the ID of the added entry
    • xadd(
      • key: String,
      • hash: Map<String, String>,
      • params: XAddParams
      ) → default StreamEntryID // the ID of the added entry
    • xadd(
      • key: Any,
      • params: Any,
      • hash: Any
      ) → return // the ID of the added entry

  Java-Sync Quick-Start

  	res1, err := rdb.XAdd(ctx, &redis.XAddArgs{
  		Stream: "race:france",
  		Values: map[string]interface{}{
  			"rider":       "Castilla",
  			"speed":       30.2,
  			"position":    1,
  			"location_id": 1,
  		},
  	}).Result()
  
  	if err != nil {
  		panic(err)
  	}
  
  	// fmt.Println(res1) // >>> 1692632086370-0
  
  	res2, err := rdb.XAdd(ctx, &redis.XAddArgs{
  		Stream: "race:france",
  		Values: map[string]interface{}{
  			"rider":       "Norem",
  			"speed":       28.8,
  			"position":    3,
  			"location_id": 1,
  		},
  	}).Result()
  
  	if err != nil {
  		panic(err)
  	}
  
  	// fmt.PrintLn(res2) // >>> 1692632094485-0
  
  	res3, err := rdb.XAdd(ctx, &redis.XAddArgs{
  		Stream: "race:france",
  		Values: map[string]interface{}{
  			"rider":       "Prickett",
  			"speed":       29.7,
  			"position":    2,
  			"location_id": 1,
  		},
  	}).Result()
  
  	if err != nil {
  		panic(err)
  	}
  
  	// fmt.Println(res3) // >>> 1692632102976-0
  

  ▼ Commands: XADD

  • XADD ( @write , @stream , @fast )

    Appends a new message to a stream. Creates the key if it doesn't exist.

    ▶ Method

    • XAdd(
      • ctx: context.Context,
      • a: *XAddArgs
      ) → *StringCmd

  Go Quick-Start

          RedisValue res1 = db.StreamAdd(
              "race:france",
              [
                  new("rider", "Castilla"),
                  new("speed", 30.2),
                  new("position", 1),
                  new("location_id", 1)
              ]
          );
          Console.WriteLine(res1);    // >>> 1712668482289-0
  
          RedisValue res2 = db.StreamAdd(
              "race:france",
              [
                  new("rider", "Norem"),
                  new("speed", 28.8),
                  new("position", 3),
                  new("location_id", 1)
              ]
          );
          Console.WriteLine(res2);    // >>> 1712668766534-1
  
          RedisValue res3 = db.StreamAdd(
              "race:france",
              [
                  new("rider", "Prickett"),
                  new("speed", 29.7),
                  new("position", 2),
                  new("location_id", 1)
              ]
          );
          Console.WriteLine(res3);    // >>> 1712669055705-0
  
  

  ▼ Commands: XADD

  • XADD ( @write , @stream , @fast )

    Appends a new message to a stream. Creates the key if it doesn't exist.

    ▶ Methods

    • StreamAdd(
      • key: RedisKey, // The key of the stream.
      • streamField: RedisValue,
      • streamValue: RedisValue,
      • messageId: RedisValue?, // The ID to assign to the stream entry, defaults to an auto-generated ID ("*").
      • maxLength: int?, // The maximum length of the stream.
      • useApproximateMaxLength: bool, // If true, the "~" argument is used to allow the stream to exceed max length by a small number. This improves performance when removing messages.
      • flags: CommandFlags // The flags to use for this operation.
      ) → RedisValue // The ID of the newly created message.
    • StreamAdd(
      • key: RedisKey, // The key of the stream.
      • streamPairs: NameValueEntry[], // The fields and their associated values to set in the stream entry.
      • messageId: RedisValue?, // The ID to assign to the stream entry, defaults to an auto-generated ID ("*").
      • maxLength: int?, // The maximum length of the stream.
      • useApproximateMaxLength: bool, // If true, the "~" argument is used to allow the stream to exceed max length by a small number. This improves performance when removing messages.
      • flags: CommandFlags // The flags to use for this operation.
      ) → RedisValue // The ID of the newly created message.
    • StreamAdd(
      • key: RedisKey, // The key of the stream.
      • streamField: RedisValue,
      • streamValue: RedisValue,
      • messageId: RedisValue?, // The ID to assign to the stream entry, defaults to an auto-generated ID ("*").
      • maxLength: long?, // The maximum length of the stream.
      • useApproximateMaxLength: bool, // If true, the "~" argument is used to allow the stream to exceed max length by a small number. This improves performance when removing messages.
      • limit: long?, // Specifies the maximal count of entries that will be evicted.
      • trimMode: StreamTrimMode, // Determines how stream trimming should be performed.
      • flags: CommandFlags // The flags to use for this operation.
      ) → RedisValue // The ID of the newly created message.
    • StreamAdd(
      • key: RedisKey, // The key of the stream.
      • streamField: RedisValue,
      • streamValue: RedisValue,
      • idempotentId: StreamIdempotentId,
      • maxLength: long?, // The maximum length of the stream.
      • useApproximateMaxLength: bool, // If true, the "~" argument is used to allow the stream to exceed max length by a small number. This improves performance when removing messages.
      • limit: long?, // Specifies the maximal count of entries that will be evicted.
      • trimMode: StreamTrimMode, // Determines how stream trimming should be performed.
      • flags: CommandFlags // The flags to use for this operation.
      ) → RedisValue // The ID of the newly created message.
    • StreamAdd(
      • key: RedisKey, // The key of the stream.
      • streamPairs: NameValueEntry[], // The fields and their associated values to set in the stream entry.
      • messageId: RedisValue?, // The ID to assign to the stream entry, defaults to an auto-generated ID ("*").
      • maxLength: long?, // The maximum length of the stream.
      • useApproximateMaxLength: bool, // If true, the "~" argument is used to allow the stream to exceed max length by a small number. This improves performance when removing messages.
      • limit: long?, // Specifies the maximal count of entries that will be evicted.
      • trimMode: StreamTrimMode, // Determines how stream trimming should be performed.
      • flags: CommandFlags // The flags to use for this operation.
      ) → RedisValue // The ID of the newly created message.

  C#-Sync (SE.Redis) Quick-Start

  res1 = r.xadd('race:france', {
    'rider' => 'Castilla',
    'speed' => 30.2,
    'position' => 1,
    'location_id' => 1
  })
  puts res1 # 1692632086370-0, for example
  
  res2 = r.xadd('race:france', {
    'rider' => 'Norem',
    'speed' => 28.8,
    'position' => 3,
    'location_id' => 1
  })
  puts res2 # 1692632094485-0, for example
  
  res3 = r.xadd('race:france', {
    'rider' => 'Prickett',
    'speed' => 29.7,
    'position' => 2,
    'location_id' => 1
  })
  puts res3 # 1692632102976-0, for example
  

  ▼ Commands: XADD

  • XADD ( @write , @stream , @fast )

    Appends a new message to a stream. Creates the key if it doesn't exist.

    ▶ Method

    • xadd(
      • key: String, // the stream key
      • entry: Hash, // one or multiple field-value pairs
      • approximate: Boolean, // whether to add ~ modifier of maxlen/minid or not
      • maxlen: Integer, // max length of entries to keep
      • minid: Integer, // min id of entries to keep
      • nomkstream: Boolean, // whether to add NOMKSTREAM
      • id: String // the entry id, default value is *
      ) → String // the entry id

  Ruby Quick-Start

          let res1 = {
              let res: Option<String> = r
                  .xadd(
                      "race:france",
                      "*",
                      &[
                          ("rider", "Castilla"),
                          ("speed", "30.2"),
                          ("position", "1"),
                          ("location_id", "1"),
                      ],
                  )
                  .expect("xadd 1");
              res.expect("missing stream id")
          };
          println!("{res1}"); // >>> 1692632086370-0
  
          let res2 = {
              let res: Option<String> = r
                  .xadd(
                      "race:france",
                      "*",
                      &[
                          ("rider", "Norem"),
                          ("speed", "28.8"),
                          ("position", "3"),
                          ("location_id", "1"),
                      ],
                  )
                  .expect("xadd 2");
              res.expect("missing stream id")
          };
          println!("{res2}"); // >>> 1692632094485-0
  
          let res3 = {
              let res: Option<String> = r
                  .xadd(
                      "race:france",
                      "*",
                      &[
                          ("rider", "Prickett"),
                          ("speed", "29.7"),
                          ("position", "2"),
                          ("location_id", "1"),
                      ],
                  )
                  .expect("xadd 3");
              res.expect("missing stream id")
          };
          println!("{res3}"); // >>> 1692632102976-0
  

  ▼ Commands: XADD

  • XADD ( @write , @stream , @fast )

    Appends a new message to a stream. Creates the key if it doesn't exist.

    ▶ Methods

    • xadd(
      • key: K,
      • id: ID,
      • items: &'a [(F, V
      ) → Any
    • xadd_map(
      • key: K,
      • id: ID,
      • map: BTM
      ) → (Option<String>)
    • xadd_options(
      • key: K,
      • id: ID,
      • items: I,
      • options: &'a streams
      ) → (Option<String>)
    • xadd_maxlen(
      • key: K,
      • maxlen: streams,
      • id: ID,
      • items: &'a [(F, V
      ) → Any
    • xadd_maxlen_map(
      • key: K,
      • maxlen: streams,
      • id: ID,
      • map: BTM
      ) → (Option<String>)

  Rust-Sync Quick-Start

          let res1: Option<String> = r
              .xadd(
                  "race:france",
                  "*",
                  &[
                      ("rider", "Castilla"),
                      ("speed", "30.2"),
                      ("position", "1"),
                      ("location_id", "1"),
                  ],
              )
              .await
              .expect("xadd 1");
          let res1 = res1.expect("missing stream id");
          println!("{res1}"); // >>> 1692632086370-0
  
          let res2: Option<String> = r
              .xadd(
                  "race:france",
                  "*",
                  &[
                      ("rider", "Norem"),
                      ("speed", "28.8"),
                      ("position", "3"),
                      ("location_id", "1"),
                  ],
              )
              .await
              .expect("xadd 2");
          let res2 = res2.expect("missing stream id");
          println!("{res2}"); // >>> 1692632094485-0
  
          let res3: Option<String> = r
              .xadd(
                  "race:france",
                  "*",
                  &[
                      ("rider", "Prickett"),
                      ("speed", "29.7"),
                      ("position", "2"),
                      ("location_id", "1"),
                  ],
              )
              .await
              .expect("xadd 3");
          let res3 = res3.expect("missing stream id");
          println!("{res3}"); // >>> 1692632102976-0
  

  ▼ Commands: XADD

  • XADD ( @write , @stream , @fast )

    Appends a new message to a stream. Creates the key if it doesn't exist.

    ▶ Methods

    • xadd(
      • key: K,
      • id: ID,
      • items: &'a [(F, V
      ) → Any
    • xadd_map(
      • key: K,
      • id: ID,
      • map: BTM
      ) → (Option<String>)
    • xadd_options(
      • key: K,
      • id: ID,
      • items: I,
      • options: &'a streams
      ) → (Option<String>)
    • xadd_maxlen(
      • key: K,
      • maxlen: streams,
      • id: ID,
      • items: &'a [(F, V
      ) → Any
    • xadd_maxlen_map(
      • key: K,
      • maxlen: streams,
      • id: ID,
      • map: BTM
      ) → (Option<String>)

  Rust-Async Quick-Start

• Read two stream entries starting at ID 1692632086370-0:

  Language: >_ Redis CLI Python JavaScript (node-redis) Java-Sync Go C#-Sync (SE.Redis) Ruby Rust-Sync Rust-Async

  Show full output

  Copied!

  Copied!

  Foundational: Retrieve stream entries within a range of IDs using XRANGE when you need to access historical data

  > XRANGE race:france 1692632086370-0 + COUNT 2
  1) 1) "1692632086370-0"
     2) 1) "rider"
        2) "Castilla"
        3) "speed"
        4) "30.2"
        5) "position"
        6) "1"
        7) "location_id"
        8) "1"
  2) 1) "1692632094485-0"
     2) 1) "rider"
        2) "Norem"
        3) "speed"
        4) "28.8"
        5) "position"
        6) "3"
        7) "location_id"
        8) "1"

  ▼ Commands: XRANGE

  • XRANGE ( @read , @stream , @slow )

    Returns the messages from a stream within a range of IDs.

  Redis CLI guide

  Also, check out our other client tools Redis Insight and Redis for VS Code.

  res4 = r.xrange("race:france", "1691765278160-0", "+", 2)
  print(
      res4
  )  # >>> [
  #   ('1692629576966-0',
  #       {'rider': 'Castilla', 'speed': '30.2', 'position': '1', 'location_id': '1'}
  #   ),
  #   ('1692629594113-0',
  #       {'rider': 'Norem', 'speed': '28.8', 'position': '3', 'location_id': '1'}
  #   )
  # ]
  

  ▼ Commands: XRANGE

  • XRANGE ( @read , @stream , @slow )

    Returns the messages from a stream within a range of IDs.

    ▶ Method

    • xrange(
      • name: KeyT,
      • min: StreamIdT = "-",
      • max: StreamIdT = "+",
      • count: Optional[int] = None
      ) → ResponseT

  Python Quick-Start

  const res4 = await client.xRange('race:france', '1691765278160-0', '+', {COUNT: 2});
  console.log(res4); // >>> [{ id: '1692629576966-0', message: { rider: 'Castilla', speed: '30.2', position: '1', location_id: '1' } }, { id: '1692629594113-0', message: { rider: 'Norem', speed: '28.8', position: '3', location_id: '1' } }]
  

  ▼ Commands: XRANGE

  • XRANGE ( @read , @stream , @slow )

    Returns the messages from a stream within a range of IDs.

    ▶ Method

    • XRANGE(
      • key: RedisArgument,
      • ...args: Parameters<typeof xRangeArguments>
      ) → Any

  Node.js Quick-Start

      List<StreamEntry> res4 = jedis.xrange("race:france","1701760582225-0","+",2);
  
      System.out.println(res4); // >>> [1701760841292-0 {rider=Castilla, speed=30.2, location_id=1, position=1}, 1701760841292-1 {rider=Norem, speed=28.8, location_id=1, position=3}]
  

  ▼ Commands: XRANGE

  • XRANGE ( @read , @stream , @slow )

    Returns the messages from a stream within a range of IDs.

    ▶ Methods

    • xrange(
      • key: byte[],
      • start: byte[], // minimum StreamEntryID for the retrieved range, passing null will
      • end: byte[] // maximum StreamEntryID for the retrieved range, passing null will
      ) → List<Object> // the entries with IDs matching the specified range, from the higher ID to the lower ID matching.
    • xrange(
      • key: byte[],
      • start: byte[], // minimum StreamEntryID for the retrieved range, passing null will
      • end: byte[], // maximum StreamEntryID for the retrieved range, passing null will
      • count: int // The entries with IDs matching the specified range.
      ) → List<Object> // the entries with IDs matching the specified range, from the higher ID to the lower ID matching.
    • xrange(
      • key: String,
      • start: StreamEntryID, // minimum StreamEntryID for the retrieved range, passing null will
      • end: StreamEntryID // maximum StreamEntryID for the retrieved range, passing null will
      ) → List<StreamEntry> // the entries with IDs matching the specified range, from the higher ID to the lower ID matching.
    • xrange(
      • key: String,
      • end: StreamEntryID start final StreamEntryID, // maximum StreamEntryID for the retrieved range, passing null will
      • count: int // The entries with IDs matching the specified range.
      ) → List<StreamEntry> // the entries with IDs matching the specified range, from the higher ID to the lower ID matching.
    • xrange(
      • key: String,
      • start: String, // minimum StreamEntryID for the retrieved range, passing null will
      • end: String // maximum StreamEntryID for the retrieved range, passing null will
      ) → List<StreamEntry> // the entries with IDs matching the specified range, from the higher ID to the lower ID matching.

  Java-Sync Quick-Start

  	res4, err := rdb.XRangeN(ctx, "race:france", "1691765278160-0", "+", 2).Result()
  
  	if err != nil {
  		panic(err)
  	}
  
  	fmt.Println(res4)
  	// >>> [{1692632086370-0 map[location_id:1 position:1 rider:Castilla...
  

  ▼ Commands: XRANGE

  • XRANGE ( @read , @stream , @slow )

    Returns the messages from a stream within a range of IDs.

    ▶ Methods

    • XRange(
      • ctx: context.Context,
      • stream: Any,
      • start: Any,
      • stop: string
      ) → *XMessageSliceCmd
    • XRangeN(
      • ctx: context.Context,
      • stream: Any,
      • start: Any,
      • stop: string,
      • count: int64
      ) → *XMessageSliceCmd

  Go Quick-Start

          StreamEntry[] res4 = db.StreamRange("race:france", "1712668482289-0", "+", 2);
  
          foreach (StreamEntry entry in res4)
          {
              Console.WriteLine($"{entry.Id}: [{string.Join(", ", entry.Values.Select(b => $"{b.Name}: {b.Value}"))}]");
          }
  
          // >>> 1712668482289-0: [rider: Castilla, speed: 30.199999999999999, position: 1, location_id: 1]
          // >>> 1712668766534-1: [rider: Norem, speed: 28.800000000000001, position: 3, location_id: 1]
  

  ▼ Commands: XRANGE

  • XRANGE ( @read , @stream , @slow )

    Returns the messages from a stream within a range of IDs.

    ▶ Methods

    • StreamRange(
      • key: RedisKey, // The key of the stream.
      • minId: RedisValue?, // The minimum ID from which to read the stream. The method will default to reading from the beginning of the stream.
      • maxId: RedisValue?, // The maximum ID to read to within the stream. The method will default to reading to the end of the stream.
      • count: int?, // The maximum number of messages to return.
      • messageOrder: Order, // The order of the messages. Ascending will execute XRANGE and Descending will execute XREVRANGE.
      • flags: CommandFlags // The flags to use for this operation.
      ) → StreamEntry[] // Returns an instance of StreamEntry for each message returned.
    • StreamRange(
      • key: RedisKey, // The key of the stream.
      • minId: RedisValue?, // The minimum ID from which to read the stream. The method will default to reading from the beginning of the stream.
      • maxId: RedisValue?, // The maximum ID to read to within the stream. The method will default to reading to the end of the stream.
      • count: int?, // The maximum number of messages to return.
      • messageOrder: Order, // The order of the messages. Ascending will execute XRANGE and Descending will execute XREVRANGE.
      • flags: CommandFlags // The flags to use for this operation.
      ) → StreamEntry[] // Returns an instance of StreamEntry for each message returned.

  C#-Sync (SE.Redis) Quick-Start

  r.del('race:france')
  r.xadd('race:france', {
    'rider' => 'Castilla',
    'speed' => '30.2',
    'position' => '1',
    'location_id' => '1'
  }, id: '1692632086370-0')
  r.xadd('race:france', {
    'rider' => 'Norem',
    'speed' => '28.8',
    'position' => '3',
    'location_id' => '1'
  }, id: '1692632094485-0')
  r.xadd('race:france', {
    'rider' => 'Prickett',
    'speed' => '29.7',
    'position' => '2',
    'location_id' => '1'
  }, id: '1692632102976-0')
  r.xadd('race:france', {
    'rider' => 'Castilla',
    'speed' => '29.9',
    'position' => '1',
    'location_id' => '2'
  }, id: '1692632147973-0')
  res4 = r.xrange('race:france', '1692632086370-0', '+', count: 2)
  puts res4.inspect
  # [["1692632086370-0", {"rider"=>"Castilla", "speed"=>"30.2", "position"=>"1", "location_id"=>"1"}],
  #  ["1692632094485-0", {"rider"=>"Norem", "speed"=>"28.8", "position"=>"3", "location_id"=>"1"}]]
  

  ▼ Commands: XRANGE

  • XRANGE ( @read , @stream , @slow )

    Returns the messages from a stream within a range of IDs.

    ▶ Method

    • xrange(
      • key: String, // the stream key
      • start: String, // first entry id of range, default is -
      • range_end: String, // last entry id of range, default is +
      • count: Integer // the number of entries as limit
      ) → Array<Array<String, Hash>> // the ids and entries pairs

  Ruby Quick-Start

          if let Ok(res) = r.xrange_count("race:france", "1692632086370-0", "+", 2) {
              let res: StreamRangeReply = res;
              let view: Vec<_> = res
                  .ids
                  .iter()
                  .map(|entry| {
                      (
                          entry.id.clone(),
                          vec![
                              ("rider".to_string(), entry.get::<String>("rider").expect("missing rider")),
                              ("speed".to_string(), entry.get::<String>("speed").expect("missing speed")),
                              ("position".to_string(), entry.get::<String>("position").expect("missing position")),
                              (
                                  "location_id".to_string(),
                                  entry.get::<String>("location_id").expect("missing location_id"),
                              ),
                          ],
                      )
                  })
                  .collect();
              println!("{view:?}");
              // >>> [("1692632086370-0", [("rider", "Castilla"), ("speed", "30.2"), ("position", "1"), ("location_id", "1")]), ("1692632094485-0", [("rider", "Norem"), ("speed", "28.8"), ("position", "3"), ("location_id", "1")])]
          }
  

  ▼ Commands: XRANGE

  • XRANGE ( @read , @stream , @slow )

    Returns the messages from a stream within a range of IDs.

    ▶ Methods

    • xrange(
      • key: K,
      • start: S,
      • end: E
      ) → (streams::StreamRangeReply)
    • xrange_all(
      • key: K
      ) → (streams::StreamRangeReply)
    • xrange_count(
      • key: K,
      • start: S,
      • end: E,
      • count: C
      ) → (streams::StreamRangeReply)

  Rust-Sync Quick-Start

          if let Ok(res) = r.xrange_count("race:france", "1692632086370-0", "+", 2).await {
              let res: StreamRangeReply = res;
              let view: Vec<_> = res
                  .ids
                  .iter()
                  .map(|entry| {
                      (
                          entry.id.clone(),
                          vec![
                              ("rider".to_string(), entry.get::<String>("rider").expect("missing rider")),
                              ("speed".to_string(), entry.get::<String>("speed").expect("missing speed")),
                              ("position".to_string(), entry.get::<String>("position").expect("missing position")),
                              (
                                  "location_id".to_string(),
                                  entry.get::<String>("location_id").expect("missing location_id"),
                              ),
                          ],
                      )
                  })
                  .collect();
              println!("{view:?}");
              // >>> [("1692632086370-0", [("rider", "Castilla"), ("speed", "30.2"), ("position", "1"), ("location_id", "1")]), ("1692632094485-0", [("rider", "Norem"), ("speed", "28.8"), ("position", "3"), ("location_id", "1")])]
          }
  

  ▼ Commands: XRANGE

  • XRANGE ( @read , @stream , @slow )

    Returns the messages from a stream within a range of IDs.

    ▶ Methods

    • xrange(
      • key: K,
      • start: S,
      • end: E
      ) → (streams::StreamRangeReply)
    • xrange_all(
      • key: K
      ) → (streams::StreamRangeReply)
    • xrange_count(
      • key: K,
      • start: S,
      • end: E,
      • count: C
      ) → (streams::StreamRangeReply)

  Rust-Async Quick-Start

• Read up to 100 new stream entries, starting at the end of the stream, and block for up to 300 ms if no entries are being written:

  Language: >_ Redis CLI Python JavaScript (node-redis) Java-Sync Go C#-Sync (SE.Redis) Ruby Rust-Sync Rust-Async

  Show full output

  Copied!

  Copied!

  Use XREAD with BLOCK to wait for new entries when you need to consume messages as they arrive

  > XREAD COUNT 100 BLOCK 300 STREAMS race:france $
  (nil)

  ▼ Commands: XREAD

  • XREAD ( @read , @stream , @slow , @blocking )

    Returns messages from multiple streams with IDs greater than the ones requested. Blocks until a message is available otherwise.

  Redis CLI guide

  Also, check out our other client tools Redis Insight and Redis for VS Code.

  res5 = r.xread(streams={"race:france": 0}, count=100, block=300)
  print(
      res5
  )
  # >>> [
  #   ['race:france',
  #       [('1692629576966-0',
  #           {'rider': 'Castilla', 'speed': '30.2', 'position': '1', 'location_id': '1'}
  #       ),
  #       ('1692629594113-0',
  #           {'rider': 'Norem', 'speed': '28.8', 'position': '3', 'location_id': '1'}
  #       ),
  #       ('1692629613374-0',
  #           {'rider': 'Prickett', 'speed': '29.7', 'position': '2', 'location_id': '1'}
  #       )]
  # ]
  # ]
  

  ▼ Commands: XREAD

  • XREAD ( @read , @stream , @slow , @blocking )

    Returns messages from multiple streams with IDs greater than the ones requested. Blocks until a message is available otherwise.

    ▶ Method

    • xread(
      • streams: Dict[KeyT, StreamIdT],
      • count: Optional[int] = None,
      • block: Optional[int] = None
      ) → ResponseT

  Python Quick-Start

  const res5 = await client.xRead({
    key: 'race:france',
    id: '0-0'
  }, {
    COUNT: 100,
    BLOCK: 300
  });
  console.log(res5); // >>> [{ name: 'race:france', messages: [{ id: '1692629576966-0', message: { rider: 'Castilla', speed: '30.2', position: '1', location_id: '1' } }, { id: '1692629594113-0', message: { rider: 'Norem', speed: '28.8', position: '3', location_id: '1' } }, { id: '1692629613374-0', message: { rider: 'Prickett', speed: '29.7', position: '2', location_id: '1' } }] }]
  

  ▼ Commands: XREAD

  • XREAD ( @read , @stream , @slow , @blocking )

    Returns messages from multiple streams with IDs greater than the ones requested. Blocks until a message is available otherwise.

    ▶ Method

    • XREAD(
      • streams: XReadStreams,
      • options?: XReadOptions
      ) → Any

  Node.js Quick-Start

      List<Map.Entry<String, List<StreamEntry>>> res5= jedis.xread(XReadParams.xReadParams().block(300).count(100),new HashMap<String,StreamEntryID>(){{put("race:france",new StreamEntryID());}});
      System.out.println(
        res5
      ); // >>> [race:france=[1701761996660-0 {rider=Castilla, speed=30.2, location_id=1, position=1}, 1701761996661-0 {rider=Norem, speed=28.8, location_id=1, position=3}, 1701761996661-1 {rider=Prickett, speed=29.7, location_id=1, position=2}]]
  

  ▼ Commands: XREAD

  • XREAD ( @read , @stream , @slow , @blocking )

    Returns messages from multiple streams with IDs greater than the ones requested. Blocks until a message is available otherwise.

    ▶ Methods

    • xread(
      • xReadParams: XReadParams,
      • streams: Entry<byte[], byte[]>...
      ) → List<Object>
    • xreadBinary(
      • streams: XReadParams xReadParams Map<byte[], StreamEntryID>
      ) → List<Map.Entry<byte[], List<StreamEntryBinary>>>
    • xreadBinaryAsMap(
      • streams: XReadParams xReadParams Map<byte[], StreamEntryID>
      ) → Map<byte[], List<StreamEntryBinary>>
    • xreadAsMap(
      • xReadParams: XReadParams,
      • streams: Map<String, StreamEntryID>
      ) → Map<String, List<StreamEntry>>
    • xread(
      • streams: XReadParams xReadParams Map<String, StreamEntryID>
      ) → List<Map.Entry<String, List<StreamEntry>>>

  Java-Sync Quick-Start

  	res5, err := rdb.XRead(ctx, &redis.XReadArgs{
  		Streams: []string{"race:france", "0"},
  		Count:   100,
  		Block:   300,
  	}).Result()
  
  	if err != nil {
  		panic(err)
  	}
  
  	fmt.Println(res5)
  	// >>> // [{race:france [{1692632086370-0 map[location_id:1 position:1...
  

  ▼ Commands: XREAD

  • XREAD ( @read , @stream , @slow , @blocking )

    Returns messages from multiple streams with IDs greater than the ones requested. Blocks until a message is available otherwise.

    ▶ Methods

    • XRead(
      • ctx: context.Context,
      • a: *XReadArgs
      ) → *XStreamSliceCmd
    • XReadStreams(
      • ctx: context.Context,
      • streams: ...string
      ) → *XStreamSliceCmd

  Go Quick-Start

          StreamEntry[] res5 = db.StreamRead("race:france", 0, 100);
  
          foreach (StreamEntry entry in res4)
          {
              Console.WriteLine($"{entry.Id}: [{string.Join(", ", entry.Values.Select(b => $"{b.Name}: {b.Value}"))}]");
          }
  
          // >>> 1712668482289-0: [rider: Castilla, speed: 30.199999999999999, position: 1, location_id: 1]
          // >>> 1712668766534-1: [rider: Norem, speed: 28.800000000000001, position: 3, location_id: 1]
          // >>> 1712669055705-0: [rider: Prickett, speed: 29.699999999999999, position: 2, location_id: 1]
  

  ▼ Commands: XREAD

  • XREAD ( @read , @stream , @slow , @blocking )

    Returns messages from multiple streams with IDs greater than the ones requested. Blocks until a message is available otherwise.

    ▶ Methods

    • StreamRead(
      • key: RedisKey,
      • position: RedisValue,
      • count: int?,
      • flags: CommandFlags // The flags to use for this operation.
      ) → StreamEntry[] // A value of RedisStream for each stream.
    • StreamRead(
      • streamPositions: StreamPosition[], // Array of streams and the positions from which to begin reading for each stream.
      • countPerStream: int?, // The maximum number of messages to return from each stream.
      • flags: CommandFlags // The flags to use for this operation.
      ) → RedisStream[] // A value of RedisStream for each stream.
    • StreamRead(
      • key: RedisKey,
      • position: RedisValue,
      • count: int?,
      • flags: CommandFlags // The flags to use for this operation.
      ) → StreamEntry[] // A value of RedisStream for each stream.
    • StreamRead(
      • streamPositions: StreamPosition[], // Array of streams and the positions from which to begin reading for each stream.
      • countPerStream: int?, // The maximum number of messages to return from each stream.
      • flags: CommandFlags // The flags to use for this operation.
      ) → RedisStream[] // A value of RedisStream for each stream.

  C#-Sync (SE.Redis) Quick-Start

  r.del('race:france')
  r.xadd('race:france', {'rider' => 'Castilla'}, id: '1692632086370-0')
  res5 = r.xread(['race:france'], ['$'], count: 100, block: 300)
  puts res5.inspect # {}
  

  ▼ Commands: XREAD

  • XREAD ( @read , @stream , @slow , @blocking )

    Returns messages from multiple streams with IDs greater than the ones requested. Blocks until a message is available otherwise.

    ▶ Method

    • xread(
      • keys: Array<String>, // one or multiple stream keys
      • ids: Array<String>, // one or multiple entry ids
      • count: Integer, // the number of entries as limit per stream
      • block: Integer // the number of milliseconds as blocking timeout
      ) → Hash // the entries

  Ruby Quick-Start

          let opts = StreamReadOptions::default().count(100).block(300);
          if let Ok(res) = r.xread_options(&["race:france"], &["$"], &opts) {
              let res: Option<StreamReadReply> = res;
              println!("{res:?}"); // >>> None
          }
  

  ▼ Commands: XREAD

  • XREAD ( @read , @stream , @slow , @blocking )

    Returns messages from multiple streams with IDs greater than the ones requested. Blocks until a message is available otherwise.

    ▶ Methods

    • xread(
      • keys: &'a [K],
      • ids: &'a [ID]
      ) → (Option<streams::StreamReadReply>)
    • xread_options(
      • keys: &'a [K],
      • ids: &'a [ID],
      • options: &'a streams
      ) → (Option<streams::StreamReadReply>)

  Rust-Sync Quick-Start

          let opts = StreamReadOptions::default().count(100).block(300);
          if let Ok(res) = r.xread_options(&["race:france"], &["$"], &opts).await {
              let res: Option<StreamReadReply> = res;
              println!("{res:?}"); // >>> None
          }
  

  ▼ Commands: XREAD

  • XREAD ( @read , @stream , @slow , @blocking )

    Returns messages from multiple streams with IDs greater than the ones requested. Blocks until a message is available otherwise.

    ▶ Methods

    • xread(
      • keys: &'a [K],
      • ids: &'a [ID]
      ) → (Option<streams::StreamReadReply>)
    • xread_options(
      • keys: &'a [K],
      • ids: &'a [ID],
      • options: &'a streams
      ) → (Option<streams::StreamReadReply>)

  Rust-Async Quick-Start

Performance

Adding an entry to a stream is O(1). Accessing any single entry is O(n), where n is the length of the ID. Since stream IDs are typically short and of a fixed length, this effectively reduces to a constant time lookup. For details on why, note that streams are implemented as radix trees.

Simply put, Redis streams provide highly efficient inserts and reads. See each command's time complexity for the details.

Streams basics

Streams are an append-only data structure. The fundamental write command, called XADD, appends a new entry to the specified stream.

Each stream entry consists of one or more field-value pairs, somewhat like a dictionary or a Redis hash:

> XADD race:france * rider Castilla speed 29.9 position 1 location_id 2
"1692632147973-0"

res6 = r.xadd(
    "race:france",
    {"rider": "Castilla", "speed": 29.9, "position": 1, "location_id": 2},
)
print(res6)  # >>> 1692629676124-0

const res6 = await client.xAdd(
  'race:france', '*', {
    'rider': 'Castilla',
    'speed': '29.9',
    'position': '1',
    'location_id': '2'
  }
);
console.log(res6); // >>> 1692629676124-0

    StreamEntryID res6 = jedis.xadd("race:france",new HashMap<String,String>(){{put("rider","Castilla");put("speed","29.9");put("position","2");put("location_id","1");}} , XAddParams.xAddParams());
    System.out.println(res6); // >>> 1701762285679-0

	res6, err := rdb.XAdd(ctx, &redis.XAddArgs{
		Stream: "race:france",
		Values: map[string]interface{}{
			"rider":       "Castilla",
			"speed":       29.9,
			"position":    1,
			"location_id": 2,
		},
	}).Result()

	if err != nil {
		panic(err)
	}

	//fmt.Println(res6) // >>> 1692632147973-0

        RedisValue res6 = db.StreamAdd(
            "race:france",
            [
                new("rider", "Castilla"),
                new("speed", 29.9),
                new("position", 1),
                new("location_id", 2)
            ]
        );

        Console.WriteLine(res6);    // >>> 1712675674750-0

res6 = r.xadd('race:france', {
  'rider' => 'Castilla',
  'speed' => 29.9,
  'position' => 1,
  'location_id' => 2
})
puts res6 # 1692632147973-0, for example

        if let Ok(res) = r.xadd(
            "race:france",
            "*",
            &[
                ("rider", "Castilla"),
                ("speed", "29.9"),
                ("position", "1"),
                ("location_id", "2"),
            ],
        ) {
            let res: Option<String> = res;
            let res = res.expect("missing stream id");
            println!("{res}"); // >>> 1692632147973-0
        }

        if let Ok(res) = r
            .xadd(
                "race:france",
                "*",
                &[
                    ("rider", "Castilla"),
                    ("speed", "29.9"),
                    ("position", "1"),
                    ("location_id", "2"),
                ],
            )
            .await
        {
            let res: Option<String> = res;
            let res = res.expect("missing stream id");
            println!("{res}"); // >>> 1692632147973-0
        }