0511_【掌握】SpringDataRedis实现Stream机制.pptx

《Redis开发实战》SpringDataRedis实现Stream机制

SpringDataRedis与StreamSpringDataRedis中为了便于Stream的消息接收操作，提供了专属的消息监听接口，开发者只需要通过监听容器并配置该监听接口的实例即可获取Stream消息内容

Stream消息监听与应答SpringDataRedis为了保证Stream消费处理的操作，每一个消费者都会启动一个专属的任务线程，而为了实现更好的线程分配处理，可以在项目中配置专属的线程池，而后每一个线程中的消费者都对应有一个完整的处理监听实例，考虑到Stream的消息机制，还需要在此监听实例中进行消息的应答处理

定义YootkStreamType的类，保存Stream基础信息packagecom.yootk.type;

publicclassYootkStreamType{

publicstaticfinalStringSTREAM=stream:yootk;//Stream流

publicstaticfinalStringGROUP=group:yootk;//组名称

}

创建消息监听类packagecom.yootk.listener;

@Component

publicclassStreamMessageListenerimplements

StreamListenerString,MapRecordString,String,Object{//消息监听

privatestaticfinalLoggerLOGGER=

LoggerFactory.getLogger(StreamMessageListener.class);

@Autowired

privateRedisTemplateString,ObjectredisTemplate;//Redis模版

@Override

publicvoidonMessage(MapRecordString,String,Objectmessage){

LOGGER.info(【Stream消息】ID={}、body={}、thread={},message.getId(),

message.getValue(),Thread.currentThread().getName());//消息输出this.redisTemplate.opsForStream().acknowledge(YootkStreamType.STREAM,

YootkStreamType.GROUP,message.getId());//消息应答

}

}

创建线程池配置类packagecom.yootk.config;

@Configuration

publicclassExecutorConfig{//线程池配置类

@Bean

publicExecutorexecutor(){//线程池

ThreadPoolTaskExecutorexecutor=newThreadPoolTaskExecutor();//线程池

intcore=Runtime.getRuntime().availableProcessors();//内核数量

executor.setCorePoolSize(core*2);//内核线程数量

executor.setAllowCoreThreadTimeOut(true);//允许超时

executor.setMaxPoolSize(core*2);//最大线程数

executor.setQueueCapacity(50);//延迟队列大小

executor.setThreadNamePrefix(yootk-stream-);//名称前缀

executor.setRejectedExecutionHandler(

newThreadPoolExecutor.AbortPolicy());//拒绝策略

executor.initialize();//初始化线程池

returnexecutor;

}