关键词:
一、用户获取新的消息通知有两种模式
-
上线登录后向系统主动索取
-
在线时系统向接收者主动推送新消息
设想下,用户的通知消息和新通知提醒数据都放在数据库中,数据库的读写操作频繁。如果消息量大,DB压力较大,可能出现数据瓶颈。这时候就可以引入消息队列RabbitMQ进行流量削峰。
向指定用户发送WebSocket消息并处理对方不在线的情况:
-
如果接收者在线,则直接发送消息;
-
否则将消息存储到redis,等用户上线后主动拉取未读消息。
二、Websocket+RabbitMQ消息推送架构图
从图中可以看出消息通知系统的基本流程是客户端A请求服务端核心模块,核心模块生产一条消息到消息队列,然后服务端消息模块消费消息,消费完之后就将消息推送给客户端B,流程很简单,没有太多技巧,唯一的巧妙之处就在消息模块这边的处理上。
一般还需要在「个人中心」需要有一个设置是否接收消息的设置项,满足用户个性化需求。
我们当前的流程有些取巧,原本应该是消费者发消息之前就去请求「用户消息设置」,用户设置成接收,才去产生消息的。而我们现在的流程中消费者不去关注用户设置,把所有消息都往「队列」里塞,让主流程去做过滤处理,这样各个生产者就不用每个都去单独处理,同时也少了一次网络交互。
三、消息通知的类型
几乎每个站点都有消息通知系统,可见通知系统的重要性不言而喻。通知系统看似简单,实际上比较复杂。那么本篇主要讲解常见的消息通知系统的设计和具体实现,包括数据库设计、逻辑关系分析等。
常见的站内通知类别:
- 公告 Announcement
-
提醒 Remind
- 资源订阅提醒「我关注的资源有更新、评论等事件时通知我」
- 资源发布提醒「我发布的资源有评论、收藏等事件时通知我」
- 系统提醒「平台会根据一些算法、规则等可能会对你的资源做一些事情,这时你会收到系统通知」
- 私信 Mailbox
以上三种消息有各自特点,实现也各不相同,其中「提醒」类通知是最复杂的:
通知事件:
通知事件就是当用户在网站或应用上产生了支付行为之后,如果你想给用户一个通知,告诉她系统已收到她的付款,那么你就要把这个「支付行为」定义为一个通知事件,并且保存这个通知事件到「通知事件表」里,以便通知系统作异步处理。通知系统会不断的处理「通知事件表」里的数据,分析每一个事件应该通知和不通知哪些人。
通知事件表「notify_event」
记录每一个用户行为产生的通知事件信息
表结构如下:
id: type: 'integer', primaryKey: true, autoIncrement:true
userID: type: 'string', required: true //用户ID
action: type: 'string', required: true //动作,如:捐款/更新/评论/收藏
objectID: type: 'string', required: true, //对象ID,如:文章ID;
objectType: type: 'string', required: true //对象所属类型,如:人、文章、活动、视频等;
createdAt:type: 'timestamp', required: true //创建时间;
用户行为定义
「action」即用户行为,如:赞了、评论了、喜欢了、捐款了、收藏了;一般来讲,我们把一个用户行为定义为一个通知类型,那么用户行为必须是需要提前定义好的。
由消息系统内部定义,为后台提供接口,用于通知设置。如下:
notify_action_type := ["donated","conllected","commented","updated"]
对象类型定义
「objectType」即用户行为作用的对象的所属类型,简单的说就是资源类型,如:项目、文章、评论、商品、视频、图片、用户。
由消息系统内部定义,为后台提供接口,用于通知设置。如下:
notify_object_type := ["project","comment"]
四、消息通知系统注意事项
4.1、Nginx代理webSocket时60s自动断开, 怎么保持长连接
利用nginx代理websocket的时候,发现客户端和服务器握手成功后,如果在60s时间内没有数据交互,连接就会自动断开。
nginx.conf 文件里location 中的proxy_read_timeout 默认60s断开,可以把他设置大一点,你可以设置成自己需要的时间,我这里设置的是十分钟(600s).
nginx配置如下:
server
listen 80;
server_name carrefourzone.senguo.cc;
#error_page 502 /static/502.html;
location /static/
root /home/chenming/Carrefour/carrefour.senguo.cc/source;
expires 7d;
location /
proxy_pass_header Server;
proxy_set_header Host $http_host;
proxy_redirect off;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Scheme $scheme;
proxy_pass http://127.0.0.1:9887;
proxy_http_version 1.1;
proxy_set_header Upgrade "websocket";
proxy_set_header Connection "Upgrade";
proxy_read_timeout 600s;
按照上述方法设置好后,我们可以发现,如果在10分钟之内没有数据交互的话,websocket连接就会自动断开,所以这种方式还是有点问题,如果我页面停留时间超过十分钟而且又没有数据交互的话,连接还是会断开的,所以需要同时结合WebSocket心跳机制的方法.
4.2、WebSocket的心跳激活机制
心跳机制是每隔一段时间会向服务器发送一个数据包,告诉服务器自己还活着,同时客户端会确认服务器端是否还活着,如果还活着的话,就会回传一个数据包给客户端来确定服务器端也还活着,否则的话,有可能是网络断开连接了。需要重连。
WebSocket 长连接需要在弱网环境和网络暂时断开的情况下,需要有一个稳定的重连机制,保证在网络不稳定的时候,客户端和服务端能够重连,继续通信。
在nginx延长超时时间的基础上,前端在超时时间内发心跳包,刷新再读时间,前端具体实现见如下代码(此处代码包含了前端整个websocket的实现过程,其中红色重点标注了发心跳包的内容):
// websocket连接
var websocket_connected_count = 0;
var onclose_connected_count = 0;
function newWebSocket()
var websocket = null;
// 判断当前环境是否支持websocket
if(window.WebSocket)
if(!websocket)
var ws_url ="wss://"+domain+"/updatewebsocket";
websocket = new WebSocket(ws_url);
else
Tip("not support websocket");
// 连接成功建立的回调方法
websocket.onopen = function(e)
heartCheck.reset().start(); // 成功建立连接后,重置心跳检测
Tip("connected successfully")
// 连接发生错误,连接错误时会继续尝试发起连接(尝试5次)
websocket.onerror = function()
console.log("onerror连接发生错误")
websocket_connected_count++;
if(websocket_connected_count <= 5)
newWebSocket()
// 接受到消息的回调方法
websocket.onmessage = function(e)
console.log("接受到消息了")
heartCheck.reset().start(); // 如果获取到消息,说明连接是正常的,重置心跳检测
var message = e.data;
if(message)
//执行接收到消息的操作,一般是刷新UI
// 接受到服务端关闭连接时的回调方法
websocket.onclose = function()
Tip("onclose断开连接");
// 监听窗口事件,当窗口关闭时,主动断开websocket连接,防止连接没断开就关闭窗口,server端报错
window.onbeforeunload = function()
websocket.close();
// 心跳检测, 每隔一段时间检测连接状态,如果处于连接中,就向server端主动发送消息,来重置server端与客户端的最大连接时间,如果已经断开了,发起重连。
var heartCheck =
timeout: 55000, // 9分钟发一次心跳,比server端设置的连接时间稍微小一点,在接近断开的情况下以通信的方式去重置连接时间。
serverTimeoutObj: null,
reset: function()
clearTimeout(this.timeoutObj);
clearTimeout(this.serverTimeoutObj);
return this;
,
start: function()
var self = this;
this.serverTimeoutObj = setInterval(function()
if(websocket.readyState == 1)
console.log("连接状态,发送消息保持连接");
websocket.send("ping");
heartCheck.reset().start(); // 如果获取到消息,说明连接是正常的,重置心跳检测
else
console.log("断开状态,尝试重连");
newWebSocket();
, this.timeout)
4.3、RabbitMQ消费端的必须捕捉异常
我遇到的Bug是对前端输入没有Emoj表情过滤导致JPA插入报错,且消费者程序没有try-catch捕捉异常,最终导致消费者程序异常终止。
4.4、WebSocket实现Web聊天室的多页面跨面问题
一般能想到的有:
-
通过IFrame
-
通过web都做成单页
-
通过sharedworker(也解决不了多浏览器),但是不是所有浏览器都支持
4.5、使用websocket实现群聊
需要很多用户(在不同的房间)进行实时聊天,也就是一个简单的聊天室,这里用的是WebSocket实现。这里需要对每一个连接都指定两个参数:用户的userId和所加入的房间id(roomId);
这里是用到一个map<房间id, 用户set>来保存房间对应的用户连接列表,当有用户进入一个房间的时候,就会先检测房间是否存在,如果不存在那就新建一个空的用户set,再加入本身到这个set中;这里需要考虑线程安全问题,因为用到的是一个hashMap,如同时又AB两个用户加入一个空房间,同时访问friends为空,然后都会新建一个set再加入进去,那么可能会出现一个情况就是A检测不存在房间,然后创建加入进去,B也同时检测到不存在,也重新创建一个用户set,这样就会覆盖原来的set,也就是说A用户就加入失败
推荐阅读:使用websocket实现群聊(多个群)
五、自定义HandshakeInterceptor,用于禁止未登录用户连接WebSocket
package cn.zifangsky.stompwebsocket.interceptor.websocket;
import cn.zifangsky.stompwebsocket.common.Constants;
import cn.zifangsky.stompwebsocket.common.SpringContextUtils;
import cn.zifangsky.stompwebsocket.model.User;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.server.ServerHttpRequest;
import org.springframework.http.server.ServerHttpResponse;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.server.HandshakeInterceptor;
import javax.servlet.http.HttpSession;
import java.text.MessageFormat;
import java.util.Map;
/**
* 自定义@link org.springframework.web.socket.server.HandshakeInterceptor,实现“需要登录才允许连接WebSocket”
*
*/
@Component
public class AuthHandshakeInterceptor implements HandshakeInterceptor
private final Logger logger = LoggerFactory.getLogger(getClass());
@Override
public boolean beforeHandshake(ServerHttpRequest serverHttpRequest, ServerHttpResponse serverHttpResponse, WebSocketHandler webSocketHandler, Map<String, Object> map) throws Exception
HttpSession session = SpringContextUtils.getSession();
User loginUser = (User) session.getAttribute(Constants.SESSION_USER);
if(loginUser != null)
logger.debug(MessageFormat.format("用户0请求建立WebSocket连接", loginUser.getUsername()));
return true;
else
logger.error("未登录系统,禁止连接WebSocket");
return false;
@Override
public void afterHandshake(ServerHttpRequest serverHttpRequest, ServerHttpResponse serverHttpResponse, WebSocketHandler webSocketHandler, Exception e)
参考链接:
使用websocket实现消息推送
WebSocket上联系客服功能在项目中很难避免,一般有下面三种实现方式: 使用http的get方式轮询接入第三方IM系统自己的IM系统 基于socket基于websocket第一种方式,最low的,实现简单,但是浪费用户流量;第二种方式,接入简... 查看详情
使用websocket实现消息推送
WebSocket上联系客服功能在项目中很难避免,一般有下面三种实现方式:使用http的get方式轮询接入第三方IM系统自己的IM系统基于socket基于websocket第一种方式,最low的,实现简单,但是浪费用户流量;第二种方式,接入简单,功能... 查看详情
使用websocket实现消息推送(下)(代码片段)
0x00WebSocket上一篇使用Java-WebSocket写了一套WebSocket的Demo,这一篇着重分析下WebSocket的一些实现细节,更加详细的协议细节可参考RFC6455。0x01WebSocket协议WebSocket协议:可参考RFC6455抓包工具:Charles和WireShark测试代码:参... 查看详情
springboot+websocket实时消息推送
...的后台管理系统实现新订单提醒推送功能,利用SpringBoot+WebSocket实时消息推送的方式进行实现。引入依赖,我使用的是SpringBoot版本2.2.6.RELEASE,自动管理依赖版本配置类WebSocketConfig,扫描并注册带有@ServerEndpoint注解的所有websocket... 查看详情
基于go的websocket消息推送的集群实现
参考技术A目前websocket技术已经很成熟,选型Go语言,当然是为了节省成本以及它强大的高并发性能。我使用的是第三方开源的websocket库即gorilla/websocket。由于我们线上推送的量不小,推送后端需要部署多节点保持高可用,所以需... 查看详情
springboot整合websocket实现实时消息推送(代码片段)
0.开发环境JDK:1.8SpringBoot:2.1.1.RELEASE1.引入依赖<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-websocket</artifactId>< 查看详情
使用websocket实现消息推送(心跳)(代码片段)
...跳的事情,这里再做一个补充。先说我的结论:WebSocket协议已经设计了心跳,这个功能可以到达检测链接是否可用心跳是用来检测链接是否可用的,不一定支持携带数据,可要看具体实现如果非要心跳中带上... 查看详情
php+sse服务器向客户端推送消息
...动接收服务端推送过来的消息,它是基于HTTP协议。SSE与WebSocket作用相似,都是建立浏览器与服务器之间的通信渠道,然后服务器向浏览器推送信息。但是WebSocket比SSE强大很多,SSE只能作为一个轻量级的消息推送方案,解决了从... 查看详情
springboot+netty+websocket实现消息推送(代码片段)
关于NettyNetty是一个利用Java的高级网络的能力,隐藏其背后的复杂性而提供一个易于使用的API的客户端/服务器框架。Maven依赖<dependencies> <!-- https://mvnrepository.com/artifact/io.netty/netty-all --> <dependency> <groupId&g 查看详情
使用websocket实现消息推送(上)(代码片段)
...get方式轮询接入第三方IM系统自己的IM系统基于socket基于websocket第一种方式,最low的,实现简单,但是浪费用户流量;第二种方式,接入简单,功能强大,但是可能需要一定的成本( 查看详情
springboot+netty+websocket实现消息推送(代码片段)
点击关注公众号,实用技术文章及时了解关于NettyNetty是一个利用Java的高级网络的能力,隐藏其背后的复杂性而提供一个易于使用的API的客户端/服务器框架。Maven依赖<dependencies> <!-- https://mvnrepository.com/artifact/io.net... 查看详情
springboot+netty+websocket实现消息推送(代码片段)
点击关注公众号,实用技术文章及时了解关于NettyNetty是一个利用Java的高级网络的能力,隐藏其背后的复杂性而提供一个易于使用的API的客户端/服务器框架。Maven依赖<dependencies> <!-- https://mvnrepository.com/artifact/io.net... 查看详情
java实现后台服务器消息推送
1.什么是WebSocketWebSocket协议是基于TCP的一种新的网络协议。它实现了浏览器与服务器全双工(full-duplex)通信——允许服务器主动发送信息给客户端。2.实现原理在实现websocket连线过程中,需要通过浏览器发出websocket连线请求,然后... 查看详情
websocket的消息推送只能固定在某一个页面吗?
...术A不会固定某一个页面一般都是用多frame来实现,即建立WebSocket是单独的一个frame,且此frame不随页面切换而刷新,至于其他需要在页面切换时刷新的元素放在其他frame里。websocket对前端其实还蛮简单(如果以前有对ajax的经验的话)... 查看详情
springboot整合websocket实现实时消息推送(代码片段)
....springframework.boot</groupId><artifactId>spring-boot-starter-websocket</artifactId></dependency>2.新建WebSocket配置类importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;importorg.springframework.web.socket.server... 查看详情
springboot集成websocket,轻松实现信息推送!
...前端,于是就使用到了MQTT,特此记录一下。一、什么是websocket?WebSocket协议是基于TCP的一种新的网络协议。它实现了客户端与服务器全双工通信,学过计算机网络都知道,既然是全双工,就说明了服务器可以主动发送信息给客... 查看详情
springboot使用websocket实现服务端推送--集群实现
...Tomcat的方式,怎么实现给某个人推送消息?比如WebSocket1连接到Tomcat1,但是在Tomcat2需要给WebSocket1发送消息,怎么办?一般的想法是像httpsession的集群处理方式一样,利用一个中间件Redis来保存session即可。但... 查看详情
springboot基于websocket进行推送
参考技术A客户端发起http请求,请求Netty服务器进行WebSocket连接,服务器接收后请求后进行注册信道并登记客户端IP地址,如此一来就建立了WebSocket通讯连接。上面的论述可以得出,我们可以比较Http和WebSocket两者之间的关系和区... 查看详情