使用java jedis封装Redis Stream操作案例
Posted 赵侠客
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了使用java jedis封装Redis Stream操作案例相关的知识，希望对你有一定的参考价值。
前言
Redis在5.0后增加了Stream功能，在日常的项目中Redis用到的比较多，但是Stream这个功能用的却是比较少，今天学习了一下Stream的基本使用功能，可以方便在接下来项目中遇到合适的再场景使用。
接口代码
public interface IStreamServer
/**
* 添加消息
* @param key
* @param streamEntryID
* @param content
* @return
*/
StreamEntryID xadd(String key, StreamEntryID streamEntryID, Map<String, String> content);
/**
* 创建分组
* @param stream
* @param group
* @param makeStream
* @return
*/
String xgroupCreate(String stream, String group, Boolean makeStream);
/**
* 倒序获取历史消息
* @param key
* @param end
* @param start
* @param count
* @return
*/
List<StreamEntry> xrevrange(String key, StreamEntryID end, StreamEntryID start, int count);
/**
* 正序获取历史消息
* @param key
* @param start
* @param end
* @param count
* @return
*/
List<StreamEntry> xrange(String key, StreamEntryID start, StreamEntryID end, int count);
/**
* 按分组获取消息
* @param group
* @param consumer
* @param count
* @param streams
* @return
*/
List<Map.Entry<String, List<StreamEntry>>> xreadGroup(String group, String consumer, int count, Map.Entry<String, StreamEntryID>... streams);
/**
* 获取消息
* @param count 获取数据
* @param streams 起始消息ID
* @return
*/
List<Map.Entry<String, List<StreamEntry>>> xread(int count, Map.Entry<String, StreamEntryID>... streams);
实现类代码
public class StreamServiceImpl implements IStreamServer
@Resource
private Pool<Jedis> jedisPool;
@Override
public StreamEntryID xadd(String key, StreamEntryID streamEntryID, Map<String, String> content)
try (Jedis jedis = jedisPool.getResource())
return jedis.xadd(key, streamEntryID, content);
@Override
public String xgroupCreate(String stream, String group, Boolean makeStream)
try (Jedis jedis = jedisPool.getResource())
return jedis.xgroupCreate(stream, group, null, makeStream);
@Override
public List<StreamEntry> xrevrange(String key, StreamEntryID end, StreamEntryID start, int count)
try (Jedis jedis = jedisPool.getResource())
return jedis.xrevrange(key, end, start, count);
@Override
public List<StreamEntry> xrange(String key, StreamEntryID start, StreamEntryID end, int count)
try (Jedis jedis = jedisPool.getResource())
return jedis.xrange(key, start, end, count);
@Override
public List<Map.Entry<String, List<StreamEntry>>> xreadGroup(String group, String consumer, int count, Map.Entry<String, StreamEntryID>... streams)
try (Jedis jedis = jedisPool.getResource())
return jedis.xreadGroup(group, consumer, count, 0, false, streams);
@Override
public List<Map.Entry<String, List<StreamEntry>>> xread(int count, Map.Entry<String, StreamEntryID>... streams)
try (Jedis jedis = jedisPool.getResource())
return jedis.xread(count, 0, streams);
测试代码
public class RedisStreamTest extends BaseTest
@Resource
private IStreamServer iStreamServer;
@Test
public void setGroup() throws Exception
//创建 流名为 video 分组为 group1
String res = iStreamServer.xgroupCreate("video", "group1", true);
//创建 流名为 audio分组为 group1
String res2 = iStreamServer.xgroupCreate("audio", "group1", true);
@Test
public void xadd() throws Exception
Long time = System.currentTimeMillis();
//向audio中插入100条消息
for (int i = 0; i < 100; i++)
StreamEntryID streamEntryID = new StreamEntryID(String.format("%s-%s", time, i));
StreamEntryID id = iStreamServer.xadd("audio", streamEntryID, ImmutableMap.of("id", String.valueOf(i)));
System.out.println(id);
//向video中插入100条消息
for (int i = 0; i < 100; i++)
StreamEntryID streamEntryID = new StreamEntryID(String.format("%s-%s", time, i));
StreamEntryID id = iStreamServer.xadd("video", streamEntryID, ImmutableMap.of("id", String.valueOf(i)));
System.out.println(id);
@Test
public void xreadGroup() throws Exception
Map<String, StreamEntryID> t = MapUtil.of("video", null);
Map<String, StreamEntryID> t2 = MapUtil.of("audio", null);
Map.Entry<String, StreamEntryID> video = t.entrySet().stream().findFirst().get();
Map.Entry<String, StreamEntryID> audio = t2.entrySet().stream().findFirst().get();
//client1 用 group1 分组 从 video/audio 流中获取两个消息 ,同一分组中不同客户端可共享消费消息(client1消费一条，client2消息偏移量增加1 )，不同分组中的客户端消息不影响（client1消费一条，不影响client2消息偏移量）
List<Map.Entry<String, List<StreamEntry>>> list = iStreamServer.xreadGroup("group1", "client1", 2, video, audio);
list.forEach(x ->
System.out.println(x.getKey() + "-->" + x.getValue());
);
//输出
/**
video-->[1610607445337-2 id=2, 1610607445337-3 id=3]
audio-->[1610608368683-2 id=2, 1610608368683-3 id=3]
**/
@Test
public void xrange() throws Exception
//video中按正序 从1610607445337-10 到1610607445337-20 中获取3条消息
List<StreamEntry> streamEntries = iStreamServer.xrange("video", new StreamEntryID("1610607445337-10"), new StreamEntryID("1610607445337-20"), 3);
streamEntries.forEach(x -> System.out.println(x));
//输出
/**
1610607445337-10 id=10
1610607445337-11 id=11
1610607445337-12 id=12
**/
@Test
public void xrevrange() throws Exception
//video中按倒序 从1610607445337-80 到1610607445337-0 中获取3条消息
List<StreamEntry> streamEntries = iStreamServer.xrevrange("video", new StreamEntryID("1610607445337-80"), new StreamEntryID("1610607445337-0"), 3);
streamEntries.forEach(x -> System.out.println(x));
//输出
/**
1610607445337-80 id=80
1610607445337-79 id=79
1610607445337-78 id=78
**/
@Test
public void xread() throws Exception
//从1610607445337-2 开始获取
Map<String, StreamEntryID> t = MapUtil.of("video", new StreamEntryID("1610607445337-2"));
Map<String, StreamEntryID> t2 = MapUtil.of("audio", null);
Map.Entry<String, StreamEntryID> video = t.entrySet().stream().findFirst().get();
Map.Entry<String, StreamEntryID> audio = t2.entrySet().stream().findFirst().get();
// 从video 1610607445337-2 开始，从audio 0开始获取 2条消息
List<Map.Entry<String, List<StreamEntry>>> list = iStreamServer.xread(2, video, audio);
list.forEach(x ->
System.out.println(x.getKey() + "-->" + x.getValue());
);
//输出
/**
audio-->[1610608368683-0 id=0, 1610608368683-1 id=1]
video-->[1610607445337-3 id=3, 1610607445337-4 id=4]
**/
总结
1.Redis Stream 消息支持持久化，可随意获取历史消息记录
2.Redis Stream 支持客户端分组功能，同一分组中的客户端可共同消费流中的消息
以上是关于使用java jedis封装Redis Stream操作案例的主要内容，如果未能解决你的问题，请参考以下文章