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

2017-01-13 15:08:08来源:csdn作者:huangwenyi1010人点击

第七城市
开心一笑

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

唯美图片

提出问题

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

解决问题1.业务描述

这几天,公司有个业务,具体内容如下:

业务图片

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

跑马灯消息提示内容总共有四种:


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

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

2.技术分析2.1业务实现流程

做业务功能实现的时候,流程基本都是这样的:熟悉业务 —>>>分析业务—>>>拆分业务—>>> 寻找拆分任务的技术解决方案 —>>>编码实现 —>>>愉快的玩耍


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


2.2技术选择跑马灯功能

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

<marquee behavior="alternate">我来回滚动</marquee>
Spring+Websocket实现消息

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

Redis队列存储消息

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

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

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

两种方法的比较:

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

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

就这样纠结啊,纠结啊,我觉得选择第二中方式,出于不想写代码的原因,加上第二种方式逼格高,效率高等等。

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

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

我定义4个队列(queue),分别存放 P1 P2 P3 P4 四种基本的消息,取数据的时候,我先从P1队列开始取,获取不到时,依次从P2P3P4去消息。

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

Spring + Quartz实现定时刷新

因为跑马灯的功能要实现实时刷新,也就是当有新的消息产生的时候,要实时刷新跑马灯的内容,我选择的方案是:在后端开启一个定时器,实时的去Redis缓存队列获取相关的信息,推送给前端。

3.代码实现3.1定时器代码实现

我在pcsMainTaskService这个业务类实现一个定时器,定时器的方法是pcsMarqueeRefresh:

<bean name="pcsMarqueeRefreshParseJob"
p:targetObject-ref="pcsMainTaskService" p:targetMethod="pcsMarqueeRefresh"
p:concurrent="false"/>
<bean>
<property name="jobDetail" ref="pcsMarqueeRefreshParseJob"/>
<property name="cronExpression" value="0/10 * * * * ?"/>
</bean>
<util:list>
<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());
}
//一般任务
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】标签 HTML跑马灯 【2】Spring+Websocket实现消息的推送 【3】Spring3.0与Quartz的整合实现定时任务调度 【4】利用Redis 实现消息队列 【5】用redis实现支持优先级的消息队列 【6】java redis使用之利用jedis实现redis消息队列

唯美图片

其他

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


第七城市

最新文章

123

最新摄影

微信扫一扫

第七城市微信公众平台