Using Redis Stream with Jedis: Basic Operations and Code Examples

In recent work the author needed a message middleware and a colleague chose Redis Stream as the solution. To integrate it into a performance‑testing framework the author began learning Redis Stream and shares basic usage demonstrations.

Redis Stream is a new data structure introduced in Redis 5.0, primarily used for message queues (MQ). Unlike Redis Pub/Sub, Stream persists messages, preventing loss when the network disconnects or the server crashes.

The author found this usage of Redis novel and proceeds to test basic functionalities.

Preparation Work

Dependencies

Be aware of version differences in the redis.clients library; mismatched versions may cause API incompatibilities. Use the following Maven or Gradle dependencies and ensure the Redis server version is 6.2.5.

<!-- https://mvnrepository.com/artifact/redis.clients/jedis -->
<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>4.2.3</version>
</dependency>

// https://mvnrepository.com/artifact/redis.clients/jedis
implementation group: 'redis.clients', name: 'jedis', version: '4.2.3'

Redis server version: Redis 6.2.5 .

XADD – Add Message to the End

If the key’s stream does not exist, it is created automatically. Create an redis.clients.jedis.params.XAddParams instance to configure parameters such as maxLen (rarely used).

def len = XAddParams.xAddParams();

Example usage:

public static void main(String[] args) {
    def base = new RedisBase("127.0.0.1", 6379);
    Jedis jedis = base.getJedis();
    def len = XAddParams.xAddParams();
    def map = new HashMap
();
    map.put("FunTester", Time.getDate() + TAB + 325);
    jedis.xadd("fun", len, map);
    jedis.close();
}

XTRIM – Trim Stream to Limit Length

This API sets the maximum length of the stream.

public static void main(String[] args) {
    def base = new RedisBase("127.0.0.1", 6379);
    Jedis jedis = base.getJedis();
    def xtrim = jedis.xtrim("fun", XTrimParams.xTrimParams().maxLen(10));
    output(xtrim);
    jedis.close();
}

The return value is the number of messages discarded.

XDEL – Delete Message

Deletes a specific message by its ID.

public static void main(String[] args) {
    def base = new RedisBase("127.0.0.1", 6379);
    Jedis jedis = base.getJedis();
    jedis.xdel("fun", new StreamEntryID(1653129389004, 1));
    jedis.close();
}

XLEN – Get Stream Length

Returns the number of entries in the stream.

jedis.xlen("fun");

XREAD – Blocking or Non‑Blocking Read

Create an redis.clients.jedis.params.XReadParams object to control the number of entries ( count ) and blocking time ( block ). A negative block value means non‑blocking, zero means block indefinitely.

def block = XReadParams.xReadParams().count(3).block(1000);

Define the second argument for Jedis#xread as a map of stream name to starting StreamEntryID . Two common options are:

Map
entry = ["fun": new StreamEntryID()]; // read historical messages
Map
entry = ["fun": StreamEntryID.LAST_ENTRY]; // read new messages after the request

Iterate over the returned messages:

List
>> xread = jedis.xread(block, entry);
output(xread.size());
Map.Entry
> get = xread.get(0);
def value = get.getValue();
value.each {
    println(it.getID());
    println(it.getFields().get("FunTester"));
}

Console output example:

16:40:56.065 main redis连接池IP：127.0.0.1，端口：6379，超时设置：5000
16:40:56.280 main 1
1653725282325-0
2022-05-28 16:08:02 325
1653725282325-1
2022-05-28 16:08:02 325
1653725282325-2
2022-05-28 16:08:02 325

XRANGE – Get Message List Within a Range

Retrieves messages between a start and end ID, automatically skipping deleted entries. Both string IDs and StreamEntryID objects are accepted.

jedis.xrange("fun", "1653129389045-0", "1653129389047-0");

The author plans to conduct performance testing of the Redis Stream API in future posts.

Have Fun ~ Tester!

FunTester Original Awards

Performance Testing Series

Java, Groovy, Go, Python

FunTester Community Highlights

Testing Theory Nuggets

Interface Function Testing Series

FunTester Video Series

Case Sharing: Solutions, Bugs, Crawlers

UI Automation Series

Testing Tools Series

Read the original article and visit my repository