基于Redission使用Redis的Stream

        redisson = Redisson.create(config);

        RStream<String, String> stream = redisson.getStream("test3");
        //初始化，不知为啥，但不这样做create不到group
        stream.add("0","0");
        //创建一个group，一个group需要在stream数据添加前创建，否则这个group只能读它创建以后写入stream的数据
        stream.createGroup("testGroup31");
        //往stream添加消息
        for(Integer i=0;i<30;i++){
            stream.add(i.toString(), i.toString());
        }

        //消费消息
        for(Integer i=0;i<6;i++){
            Integer finalI = i;
            Thread t = new Thread( ()->{
                try {
                    Thread.sleep(1000);
                }catch (Exception e){

                }

                Map<StreamMessageId, Map<String, String>> s = stream.readGroup("testGroup31", "consumer"+ finalI.toString(),1);
                if(s!=null && s.size()>0){
                    for (Map.Entry<StreamMessageId, Map<String, String>> entry : s.entrySet()) {
                        Map<String, String> m2 = entry.getValue();
                        for(Map.Entry<String,String> entry1:m2.entrySet()){
                            System.out.println(Thread.currentThread().getName()+" : Key = " + entry1.getKey() + ", Value = " + entry1.getValue());
                        }
                        //消费了消息，要应答一下
                        stream.ack("testGroup31",entry.getKey());
                        //如果消费了消息想删除，可以删除掉
                        //stream.remove(entry.getKey());
                    }
                }

            });
            t.start();
        }

image.png

image.png

这里,有个group名字叫testGroup31，里面有消费者6个，pending代表目前有6个数据被读取了，但没有ack。last_delivered-id代表这个group目前读到哪条消息。