RedisStream队列

Bertha 。 2021-09-25 15:12 267阅读 0赞

spring-data-redis 实现RedisStream 队列读取发送 demo

import java.time.Duration;
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    import javax.annotation.Resource;
    
    import org.apache.log4j.Logger;
    
    import org.springframework.data.redis.connection.stream.Consumer;
    import org.springframework.data.redis.connection.stream.MapRecord;
    
    import org.springframework.data.redis.connection.stream.ReadOffset;
    import org.springframework.data.redis.connection.stream.RecordId;
    import org.springframework.data.redis.connection.stream.StreamOffset;
    import org.springframework.data.redis.connection.stream.StreamReadOptions;
    import org.springframework.data.redis.core.StringRedisTemplate;
    import org.springframework.stereotype.Component;
    import com.game.util.JsonTool;
    import com.game.util.SpringManager;
    
    /**
     * RedisStream队列
     * 
     * @author gaoyannan
     *
     */
    @Component("redisStream")
    public class RedisStream {
    	public final static ExecutorService executors = Executors.newCachedThreadPool();
    	@Resource(name = "redisTemplateContainer")
    	private StringRedisTemplate redisTemplate;
    
    	public static RedisStream util() {
    		return SpringManager.getBean(RedisStream.class);
    	}
    
    	static Logger logger = Logger.getLogger(RedisStream.class);
    	public static final String key = "test:mystream2";
    	/***
    	 * 分组名字
    	 */
    	public static final String group = "group";
    	/**
    	 * 消费者名字
    	 */
    	public static final String consumer = "consumer";
    	
    	/**
    	 * add消息
    	 * @param content
    	 */
    	public void XADD(String content) {
    		
    		for (int i = 0; i < 10; i++) {
    			final int y=i;
    			executors.execute(new Runnable() {
    				
    				@Override
    				public void run() {
    					
    						Map<Object, Object> hm=new HashMap<Object, Object>(); 
    						hm.put("key", content+y);
    						redisTemplate.boundStreamOps(key).add(hm);
    					}
    			});
    		}
    		
    		
    	}
    	/**
    	 * 阻塞读取消息
    	 */
    	public void XREADGROUP() {
    		pending();
    		logger.info("MQ初始化信息开始");
    		//初始化队列
    		Map<Object, Object> hm=new HashMap<Object, Object>(); 
    		hm.put("-1", "-1");
    		RecordId recordId = this.redisTemplate.boundStreamOps(key).add(hm);
    		this.redisTemplate.boundStreamOps(key).delete(recordId.getValue());
    		//初始化组
    		
    		this.redisTemplate.boundStreamOps(key).destroyGroup(group);//销毁一个消费者组。
    		this.redisTemplate.boundStreamOps(key).createGroup(ReadOffset.latest(), group);//创建一个消费者组。
    		logger.info("MQ初始化信息完毕");
    		
    		while (true) {
    			try {
    					List<MapRecord<String, Object, Object>>   list=this.redisTemplate.boundStreamOps(key).read(Consumer.from(group, consumer), StreamReadOptions.empty().block(Duration.ofDays(1)), ReadOffset.lastConsumed());
    					logger.info("mq收到消息："+"消息长度="+list.size() );
    					pending();
    					int i=0;
    					int err=0;
    					for (MapRecord<String, Object, Object> mapRecord : list) {
    						RecordId recordId1 = mapRecord.getId();
    						long del=0;
    						del=this.redisTemplate.boundStreamOps(key).acknowledge(group, recordId1.getValue());
    						if(del!=1) {
    							err++;
    							logger.info(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>Redis Stream mq acknowledge消息异常 mapRecord="+JsonTool.encode(mapRecord));
    						}else {
    							i++;
    						}
    						redisTemplate.boundStreamOps(key).trim(10);
    					}	
    					logger.info("mq消息：成功acknowledge消息="+i +" 合计="+(i+err));
    					logger.info("mq消息：失败acknowledge消息="+err +" 合计="+(i+err));
    					pending();
    			} catch (Exception e) {
    				continue;
    			}
    		}
    	}
    	
    	@SuppressWarnings("unchecked")
    	public void pending() {
    		try {
    			/**
    			 * 只能读取新的未读消息
    			 */
    			List<MapRecord<String, Object, Object>>  list2 = redisTemplate.opsForStream().read(Consumer.from(group, consumer), StreamOffset.create(key, ReadOffset.from("0")));
    			logger.info("mq 未读消息："+" 未读消息长度="+list2.size() );
    		} catch (Exception e) {
    			// TODO: handle exception
    			e.printStackTrace();
    		}
    	}
    	public void init() {
    			executors.execute(new Runnable() {
    				
    				@Override
    				public void run() {
    					XREADGROUP();
    				}
    			});
    	}
    
    }