Redis实战(4)-数据结构List实战之队列特性实现消息多线程 广播通知

作者: 修罗debug
版权声明:本文为博主原创文章,遵循 CC 4.0 by-sa 版权协议,转载请附上原文出处链接和本声明。



摘要:电商平台的管理后端一般有两大角色的用户可以使用,一个是系统管理员,一个是平台的卖家/商家,对于商家而言,管理自个儿的商品是日常工作中再为普通不过的事情了,而对于系统管理员而言,有时候需要发布一些活动公告通知商家进行报名参加,本文我们将基于List的队列特性实现公告消息的广播通知功能!

内容:在上篇文章中我们介绍了Redis的数据结构~列表List,简单介绍了其基本特性及其在电商应用后端管理平台下如何实现“商家”添加商品时的有序存储,以及如何以有序列表的形式进行展示!

在其中,我们给大家展示了列表List在存储和获取数据时的流程图,不晓得大伙儿还记不记得,如下图所示:


从该图中可以看出,当我们往Redis的列表List中添加数据时,数据的流动是具有“先进先出”的特性,即所谓的“FIFO”(有点队列Queue的特性!)的,而且数据是紧凑、一个挨着一个存储的!

即当我们在往缓存Redis的列表List添加数据时,可以采用“LPush 即从左边的方向添加”的方式往缓存Redis的List中添加,然后再采用“LPop 即从左边的方向弹出数据”或者“RPop 即从右边的方向弹出数据”的方式获取这一有序存储的列表数据!

知道了列表List的数据存储和读取流程,其实我们也就几乎知晓了在实际的项目实战开发中的代码实现了。

下面我们以“电商应用~平台管理员在平台发布活动公告信息之后,除了将公告信息塞入数据库DB之外,同时以LPush的方式将其塞入缓存Redis的列表List中,并在接口的另一端开启定时检测的方式,随时检测缓存中指定的列表Redis是否有通告信息过来,如果有,则采取RPop的方式弹出该公告信息,并以邮件的形式发送给商户!”,如下图所示:


下面,我们就进入代码实战环节吧!

(1)首先,当然是需要来个“通告信息表”啦,其完整的DDL(即数据定义语言)如下所示:

CREATE TABLE `notice` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`title` varchar(255) CHARACTER SET utf8mb4 DEFAULT NULL COMMENT '通告标题',
`content` varchar(255) CHARACTER SET utf8mb4 DEFAULT NULL COMMENT '内容',
`is_active` tinyint(4) DEFAULT '1',
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=18 DEFAULT CHARSET=utf8 COMMENT='通告';

(2)然后,当然是需要开发一个Controller啦(上文我们已经开发过了)!在该Controller中我们需要开设一个请求方法,给平台管理员添加“通告信息”,该请求方法在接收到公告信息之后需要将其塞入数据库DB中,同时也需要往缓存Redis的列表List中LPush一条公告信息,准备被监听检测!其完整的源代码如下所示:  

    //平台发送通知给到各位商户
@RequestMapping(value = "/notice/put",method = RequestMethod.POST,consumes = MediaType.APPLICATION_JSON_UTF8_VALUE)
public BaseResponse putNotice(@RequestBody @Validated Notice notice, BindingResult result){
String checkRes=ValidatorUtil.checkResult(result);
if (StrUtil.isNotBlank(checkRes)){
return new BaseResponse(StatusCode.Fail.getCode(),checkRes);
}
BaseResponse response=new BaseResponse(StatusCode.Success);
try {
log.info("--平台发送通知给到各位商户:{}",notice);

listService.pushNotice(notice);
}catch (Exception e){
response=new BaseResponse(StatusCode.Fail.getCode(),e.getMessage());
}
return response;
}

(3)紧接着,我们需要开发Controller对应的Service,其职责当然是用来处理真正的业务逻辑,即“添加完成通告信息”时,它负责将该商品信息添加进DB数据库,并添加进缓存Redis的列表List中,其完整的源代码如下所示:  

    @Autowired
private NoticeMapper noticeMapper;

//创建通告
@Transactional(rollbackFor = Exception.class)
public void pushNotice(Notice notice) throws Exception{
if (notice!=null){
notice.setId(null);
//TODO:将通告信息塞入数据库DB中
noticeMapper.insertSelective(notice);
final Integer id=notice.getId();

if (id>0){
//TODO:塞入List列表中(队列),准备被拉取异步通知至不同的商户的邮箱 - applicationEvent&Listener;Rabbitmq;jms
ListOperations<String,Notice> listOperations=redisTemplate.opsForList();
listOperations.leftPush(Constant.RedisListNoticeKey,notice);
}
}
}

(4)之后,我们需要创建一个“定时任务调度器”,用于“近实时”的检测缓存Redis中的列表List是否有通知公告信息,如果有,则将其RPop取出来,然后采取多线程的形式将其发送给“平台的商家”,让他们赶紧报名参加相关的活动!其完整的源代码如下所示:  

/**
* Redis列表-队列的消费者监听器
* @Author:debug (SteadyJack)
* @Link: weixin-> debug0868 qq-> 1948831260
* @Date: 2019/10/30 14:51
**/
@Component
@EnableScheduling
public class ListListenerScheduler {

private static final Logger log=
LoggerFactory.getLogger(ListListenerScheduler.class);

private static final String listenKey= Constant.RedisListNoticeKey;

@Autowired
private RedisTemplate redisTemplate;

@Autowired
private UserMapper userMapper;

@Autowired
private EmailService emailService;

//TODO:近实时的定时任务检测
//@Scheduled(cron = "0/10 * * * * ?")
@Scheduled(cron = "0/59 * * * * ?")
public void schedulerListenNotice(){
log.info("----定时任务调度队列监听、检测通告消息,监听list中的数据");

ListOperations<String,Notice> listOperations=redisTemplate.opsForList();
Notice notice=listOperations.rightPop(listenKey);
while (notice!=null){
//TODO:发送给到所有的商户的邮箱
this.noticeUser(notice);

notice=listOperations.rightPop(listenKey);
}
}

//TODO:发送通知给到不同的商户
@Async("threadPoolTaskExecutor")
private void noticeUser(Notice notice){
if (notice!=null){
//TODO:查询获取所有商户信息
List<User> list=userMapper.selectList();

//TODO:线程池/多线程触发群发邮件
try {
if (list!=null && !list.isEmpty()){
ExecutorService executorService=Executors.newFixedThreadPool(4);
List<NoticeThread> threads= Lists.newLinkedList();

list.forEach(user -> {
threads.add(new NoticeThread(user,notice,emailService));
});

executorService.invokeAll(threads);
}
}catch (Exception e){
log.error("近实时的定时任务检测-发送通知给到不同的商户-法二-线程池/多线程触发-发生异常:",e.fillInStackTrace());
}
}
}
}

(5)至此,我们的代码实战就完毕了,最后我们就基于Postman进入测试环节吧,几张图加以概括吧:



最后,上一下邮箱看看吧,可以发现确实收到了邮件:


好了,本篇文章我们就介绍到这里了,建议各位小伙伴一定要照着文章提供的样例代码撸一撸,只有撸过才能知道这玩意是咋用的,否则就成了“空谈者”!

对Redis相关技术栈以及实际应用场景实战感兴趣的小伙伴可以前往Debug搭建的技术社区的课程中心进行学习观看:https://www.fightjava.com/web/index/course/detail/12

其他相关的技术,感兴趣的小伙伴可以关注底部Debug的技术公众号,或者加Debug的微信,拉你进“微信版”的真正技术交流群!一起学习、共同成长!

补充:

1、本文涉及到的相关的源代码可以到此地址,check出来进行查看学习:

https://gitee.com/steadyjack/SpringBootRedis

2、目前Debug已将本文所涉及的内容整理录制成视频教程,感兴趣的小伙伴可以前往观看学习:https://www.fightjava.com/web/index/course/detail/12

3、关注一下Debug的技术微信公众号,最新的技术文章、课程以及技术专栏将会第一时间在公众号发布哦!