使用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}]
        **/
    }
}