Java之Redis队列+Websocket+定时器实现跑马灯实时刷新

开心1笑

【老婆对老公说：老公你说我奇怪不奇怪啊，我既然可以用眼睛吃饭也。老公：说的新鲜，怎样用眼睛怎样吃饭啊。老婆说：我看你1眼就饱了……】

提出问题

项目中如何利用Redis队列+定时器（quartz）+websocket实现实时刷新的跑马灯功能???

解决问题

1.业务描写

这几天，公司有个业务，具体内容以下：

在仪表盘banner区域转动播放提示信息。也就是实现1个实时播放消息的跑马灯功能。播放的是1个任务内容(数据库有1张表pm_task)。

跑马灯消息提示内容总共有4种：

• 任务下发——P3（消息播放队列优先级）
  任务被下发时进行提示。
  文字提示内容：任务已下发：任务编号 任务名称
• 1般任务复核通过——P4
  任务复核通过时进行提示。
  文字提示内容：任务已通过：任务编号 任务名称
• 关键决策任务复核通过——P1
  任务复核通过时进行提示。
  文字提示内容：关键决策任务通过：任务编号 任务名称
• 关键验证任务复核通过——P2
  任务复核通过时进行提示。
  文字提示内容：关键验证任务通过：任务编号 任务名称

转动播放时，每一个提示信息之间应由字符间隔。播放速度到时根据具体代码运行情况进行分析，播放速度应不超过1般浏览速度。当有提示信息生成时，末位补进提示信息队列。
特殊情况：
1. 当信息同时生成时，同级任务信息按时间进行排序。
2. 消息插播优先级：P1>P2>P3>P4。
3. 插播为即时插播，未播放完的消息不移除播放队列。

2.技术分析

2.1业务实现流程

做业务功能实现的时候，流程基本都是这样的：熟习业务 —>>>分析业务—>>>拆分业务—>>> 寻觅拆分任务的技术解决方案 —>>>编码实现 —>>>愉快的顽耍

在业务没想清楚之前，千万不要动手写代码。

2.2技术选择

跑马灯功能

网上搜索前端跑马灯功能实现，1堆，可以看看文章最后参考文章那1节。具体实现就是HTML的1个便签< marquee>

<marquee behavior="alternate">我来回转动</marquee> 

Spring+Websocket实现消息

消息推送，公司既有的框架就是Websocket，所以可以在用户进入页面的时候，定阅相干通道，用户退出页面的时候，取消相干的通道。在需要推送消息时候，实现消息推送既可。

Redis队列存储消息

后端产生的消息，事实上有2种存储方法：

1.我利用数据库，建立1张表，产生的每条消息都保存到表（xxx_marquee_msg）里面。当前端跑马灯需要数据的时候，从数据库读取1条优先级高的数据，返回给前端。与此同时，我把该条数据删除，实现1个类似队列这样的1个功能。

2.我利用Redis的阻塞队列功能，将数据寄存到redis队列中。前端需要的时候，我再从队列中获得数据。

两种方法的比较：

利用数据库方法实现，简单，业务逻辑好控制，缺点是：你得实现表的增删改查操作，需要些很对的代码，从控制层，业务层，DAO层，1层1层的写，1堆代码，麻烦。

相比之下，如果用Redis的阻塞队列来实现，我不需要写增删改查操操作，只需要get和push消息到队列中便可，同时由于在缓存中，效力高，缺点是：业务逻辑不好控制，比如我要实现队列的排序，优先级，相对来讲都比较麻烦。

就这样纠结啊，纠结啊，我觉得选择第2中方式，出于不想写代码的缘由，加上第2种方式逼格高，效力高等等。

Redis消息队列优先级解决方法

仔细看下需求，你会发现，需求中要求消息是排优先级的，这点就有点头疼了，不过好在，我们的消息只有4中优先级，所以具体解决方案以下：

我定义4个队列（queue），分别寄存 P1 P2 P3 P4 4种基本的消息，取数据的时候，我先从P1队列开始取，获得不到时，顺次从P2 P3 P4去消息。

可以参考这篇文章用redis实现支持优先级的消息队列

Spring + Quartz实现定时刷新

由于跑马灯的功能要实现实时刷新，也就是当有新的消息产生的时候，要实时刷新跑马灯的内容，我选择的方案是：在后端开启1个定时器，实时的去Redis缓存队列获得相干的信息，推送给前端。

3.代码实现

3.1定时器代码实现

我在pcsMainTaskService这个业务类实现1个定时器，定时器的方法是pcsMarqueeRefresh：

<bean name="pcsMarqueeRefreshParseJob" class="org.springframework.scheduling.quartz.MethodInvokingJobDetailFactoryBean"
      p:targetObject-ref="pcsMainTaskService" p:targetMethod="pcsMarqueeRefresh"
      p:concurrent="false"/>


<bean id="pcsMarqueeRefreshTrigger" class="org.springframework.scheduling.quartz.CronTriggerFactoryBean">
    <property name="jobDetail" ref="pcsMarqueeRefreshParseJob"/>
    <property name="cronExpression" value="0/10 * * * * ?"/>
</bean>


<util:list id="schedulerTriggers">
    <ref bean="pcsMarqueeRefreshTrigger"></ref>
</util:list>

3.2业务代码实现

    /**
     * 描写：跑马灯刷新（定时器）
     */
    public void pcsMarqueeRefresh() throws Exception{
        // 推送内容
        String pushContent =  null;
        if(StringUtils.isEmpty(pushContent)){
            pushContent = RedisUtils.getFromQueue(MarqueeRefreshUtils.REDIS_MARQUEE_P1_KEY);
        }
        if(StringUtils.isEmpty(pushContent)){
            pushContent = RedisUtils.getFromQueue(MarqueeRefreshUtils.REDIS_MARQUEE_P2_KEY);
        }
        if(StringUtils.isEmpty(pushContent)){
            pushContent = RedisUtils.getFromQueue(MarqueeRefreshUtils.REDIS_MARQUEE_P3_KEY);
        }
        if(StringUtils.isEmpty(pushContent)){
            pushContent = RedisUtils.getFromQueue(MarqueeRefreshUtils.REDIS_MARQUEE_P4_KEY);
        }
        //推送消息
        if(StringUtils.isNotEmpty(pushContent)){
            //查询系统所有的用户
            List<String> userIds = sysUserService.find(new ArrayList<>()).stream().map(SysUser::getId).collect(Collectors.toList());
            //websocket推送消息
            redisPubSubService.publish(new RedisMessage(pushContent, userIds, MarqueeRefreshUtils.MARQUEE_CHANNEL,false));
        }
    }

3.3 WebSocket消息推送代码实现

消息推送代码比较简单，获得系统用户，往通道（MarqueeRefreshUtils.MARQUEE_CHANNEL）推送消息。

//查询系统所有的用户
            List<String> userIds = sysUserService.find(new ArrayList<>()).stream().map(SysUser::getId).collect(Collectors.toList());
            //websocket推送消息
            redisPubSubService.publish(new RedisMessage(pushContent, userIds, MarqueeRefreshUtils.MARQUEE_CHANNEL,false));

3.4消息存取实现

package com.evada.de.projcommand.utils;

import com.evada.de.common.enums.projcommond.TaskDeliverStatus;
import com.evada.de.common.enums.projcommond.TaskTypeEnum;
import com.evada.de.common.util.RedisUtils;
import com.evada.de.projcommand.model.PcsTask;

/**
 * 描写：跑马灯消息刷新
 * Created by huangwy on 2017/1/9.
 */
public class MarqueeRefreshUtils {

    // 队列总共分为4个级别，分别为 P1 P2 P3 P4
    public static final String REDIS_MARQUEE_P1_KEY = "inno.pcs.marquee.refresh.p1";
    public static final String REDIS_MARQUEE_P2_KEY = "inno.pcs.marquee.refresh.p2";
    public static final String REDIS_MARQUEE_P3_KEY = "inno.pcs.marquee.refresh.p3";
    public static final String REDIS_MARQUEE_P4_KEY = "inno.pcs.marquee.refresh.p4";
    // 定阅频道
    public static final String MARQUEE_CHANNEL = "inno.pcs.marquee.refresh";
    //消息前缀
    public static final String PRE_P1_MESSAGE = "关键决策任务通过:";
    public static final String PRE_P2_MESSAGE = "关键验证任务通过:";
    public static final String PRE_P3_MESSAGE = "任务已下发:";
    public static final String PRE_P4_MESSAGE = "任务已通过:";


    public static void pushToQueue(PcsTask pcsTask){
        if(!(pcsTask.getWorkitemStatus().equals("3") 
                || pcsTask.getDeliverStatus().equals(TaskDeliverStatus.TASK_DELIVER.toString()))){
            return;
        }
        StringBuffer content = new StringBuffer();
        //关键决策任务
        if(TaskTypeEnum.KEY_DECISION_TASK.toString().equals(pcsTask.getType()) && pcsTask.getWorkitemStatus().equals("3")){
            content.append(PRE_P1_MESSAGE).append(pcsTask.getCode()).append(" ").append(pcsTask.getName());
            RedisUtils.putToQueue(REDIS_MARQUEE_P1_KEY,content.toString());
        }
        //关键验证任务
        if(TaskTypeEnum.KEY_VALIDATION_TASK.toString().equals(pcsTask.getType()) && pcsTask.getWorkitemStatus().equals("3")){
            content.append(PRE_P2_MESSAGE).append(pcsTask.getCode()).append(" ").append(pcsTask.getName());
            RedisUtils.putToQueue(REDIS_MARQUEE_P2_KEY,content.toString());
        }
        //任务已下发
        if(TaskTypeEnum.KEY_VALIDATION_TASK.toString().equals(pcsTask.getType())
                && pcsTask.getDeliverStatus().equals(TaskDeliverStatus.TASK_DELIVER.toString())){
            content.append(PRE_P3_MESSAGE).append(pcsTask.getCode()).append(" ").append(pcsTask.getName());
            RedisUtils.putToQueue(REDIS_MARQUEE_P3_KEY,content.toString());
        }
        //1般任务
        if(TaskTypeEnum.GENERAL_TASK.toString().equals(pcsTask.getType()) && pcsTask.getWorkitemStatus().equals("3")){
            content.append(PRE_P4_MESSAGE).append(pcsTask.getCode()).append(" ").append(pcsTask.getName());
            RedisUtils.putToQueue(REDIS_MARQUEE_P4_KEY,content.toString());
        }
    }
}

3.5缓存工具类实现

该工具类主要是实现队列数据的存和取，相对来讲比较简单：

package com.evada.de.common.util;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

@Component
public class RedisUtils {
    private static RedisTemplate tmp;

    @Autowired
    RedisUtils(RedisTemplate redisTemplate) {
        tmp = redisTemplate;
    }


    /**
     * set value to queue
     * @param key
     * @param value
     * @return
     */
    public static Long putToQueue(final String key, final String value) {
        Long l = (Long) tmp.execute(new RedisCallback<Object>() {
            public Object doInRedis(RedisConnection connection)throws DataAccessException {
                return connection.lPush(key.getBytes(), value.getBytes());
            }
        });
        return l;
    }

    /**
     * get value from queue
     * @param key
     * @return
     */
    public static String getFromQueue(final String key) {
        byte[] b = (byte[]) tmp.execute(new RedisCallback<Object>() {
            public Object doInRedis(RedisConnection connection)throws DataAccessException {
                return connection.lPop(key.getBytes());
            }
        });
        if(b != null){
            return new String(b);
        }
        return null;
    }

}

好了，写到这里基本就实现了，很简单有木有~~~

读书感悟

来自 古斯塔夫·勒庞《乌合之众》

• 人1到群体中，智商就严重下降，为了取得认同，个体愿意抛弃是非，用智商去换取那份让人备感安全的归属感。
• 我们以为自己是理性的，我们以为自己的1举1动都是有其道理的。但事实上，我们的绝大多很多天常行动，都是1些我们自己根本没法了解的隐蔽动机的结果。
• 个人1旦成为群体的1员，他所作所为就不会再承当责任，这时候每一个人都会暴露出自己不遭到的束缚的1面。群体寻求和相信的历来不是甚么真相和理性，而是盲从、残暴、偏执和狂热，只知道简单而极真个感情。
• 群体只会干两种事——锦上添花或落井下石。
• 孤立的个人具有主宰自己的反应行动的能力，群体则缺少这类能力。群体中的个人极易受刺激因素的影响，转眼之间就从最血腥的狂热变成最极真个宽宏大量和英雄主义。群体很容易做出刽子手的举动，一样也很容易慷慨就义，为每种信仰的成功而不惜血流成河。
• 群体表现出来的感情不论是好是坏，其突出的特点就是极其简单而夸大。

经典故事

【1个失意年轻人寻觅成功，哲人给1颗花生说：“用力捏它。”年轻人用力1捏，花生壳碎了，剩下仁。哲人又叫他搓，结果搓掉红色的皮，只留下白白的果实。哲人再叫他捏，不论他如何用力，却捏不碎花生仁。哲人说：“虽然屡受磨难，失去了很多，但要有1颗不屈的心。”】

参考文章

【1】标签 HTML跑马灯
【2】Spring+Websocket实现消息的推送
【3】Spring3.0与Quartz的整合实现定时任务调度
【4】利用Redis 实现消息队列
【5】用redis实现支持优先级的消息队列
【6】java redis使用之利用jedis实现redis消息队列

其他

如果有带给你1丝丝小快乐，就让快乐继续传递下去，欢迎点赞、顶、欢迎留下宝贵的意见、多谢支持！