马克社区,马克吐温社区,马克java社区Redis实战(4)-数据结构List实战之队列特性实现消息多线程广播通知

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());
            }
        }
    }
}