使用java jedis封装Redis Stream操作案例

我会带着你远行 2023-01-06 11:56 537阅读 0赞

## 前言 ##

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 支持客户端分组功能，同一分组中的客户端可共同消费流中的消息