Redis實戰(4)-數據結構List實戰之隊列特性實現消息多線程 廣播通知

概述：本系列博文所涉及的相關內容來源於debug親自錄製的實戰課程：緩存中間件Redis技術入門與應用場景實戰（SpringBoot2.x + 搶紅包系統設計與實戰），感興趣的小夥伴能夠點擊自行前往學習（畢竟以視頻的形式來掌握技術 會更快！） ，文章所屬專欄：緩存中間件Redis技術入門與實戰

摘要：電商平臺的管理後端通常有兩大角色的用戶可使用，一個是系統管理員，一個是平臺的賣家/商家，對於商家而言，管理自個兒的商品是平常工做中再爲普通不過的事情了，而對於系統管理員而言，有時候須要發佈一些活動公告通知商家進行報名參加，本文咱們將基於List的隊列特性實現公告消息的廣播通知功能！

內容：在上篇文章中咱們介紹了Redis的數據結構~列表List，簡單介紹了其基本特性及其在電商應用後端管理平臺下如何實現「商家」添加商品時的有序存儲，以及如何以有序列表的形式進行展現！（文章連接：Redis實戰(3)-數據結構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相關技術棧以及實際應用場景實戰感興趣的小夥伴能夠我們51cto學院 debug親自錄製的課程進行學習：緩存中間件Redis技術入門與應用場景實戰（SpringBoot2.x + 搶紅包系統設計與實戰）

補充：

一、本文涉及到的相關的源代碼能夠到此地址，check出來進行查看學習：https://gitee.com/steadyjack/SpringBootRedis

二、目前debug已將本文所涉及的內容整理錄製成視頻教程，感興趣的小夥伴能夠前往觀看學習：https://edu.51cto.com/course/20384.html