我要评论应用场景说明：
由于redis并非专业的mq中间件，消息的防丢失策略并不完整，存在丢失消息的可能。该方案为在再pc web管理平台的右下角弹出，显示新接收到的消息数，哪怕没有收到这个通知，也可以自己在消息中心看看。所以对可靠性要求不高。如果业务场景要求可靠性高，还是请使用专业的mq中间件。该方案已在多个实际项目中运行。
流程架构：
websocket实现同一账户多点登录、websocket服务多节点部署推送方案。
简单架构图
假设用户a在两个地方登录，连接到两个websocketserver服务节点1和2，用户b连接到2节点。
websocketserver将websocket session保存在各自的map<string,session>中，key为userid，value为websocket session。节点1保存了用户a的websocket session，节点2保存了用户a、b的websocket session。
消息生产者发布消息的时候为json格式，如：[{"receive"="userid_a"，"msg"="您有1个未读消息"},{"receive"="userid_b"，"msg"="您有3个未读消息"}]，将消息发到redis的一个channel，如shownewestmsg。
websocketserver中订阅redis的channel=shownewestmsg，收到消息后根据消息中receive冲map中找到对应的websocket session，发消息给客户端。
核心代码：
1.该项目为springboot项目，先引入jar包，由于是从实际项目中抽出来写的记录，可能还缺jar请自行导入。
<dependency> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter-web</artifactid> </dependency> <!--websocket--> <dependency> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter-websocket</artifactid> </dependency> <!-- redis --> <dependency> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter-data-redis</artifactid> </dependency> <!-- 工具类 --> <dependency> <groupid>cn.hutool</groupid> <artifactid>hutool-all</artifactid> <version>5.3.6</version> </dependency> <dependency> <groupid>net.sf.json-lib</groupid> <artifactid>json-lib</artifactid> <version>2.4</version> <classifier>jdk15</classifier> </dependency>
2.websocket配置
import org.springframework.context.annotation.bean;
import org.springframework.context.annotation.configuration;
import org.springframework.web.socket.server.standard.serverendpointexporter;
/**
* spring websocket组件初始化
* @author csf
*
*/
//war包启动tomcat7及以下版本要关闭@configuration注解,否则将无法启动websocket服务
@configuration
public class websocketconfig
{
@bean
public serverendpointexporter serverendpointexporter()
{
return new serverendpointexporter();
}
}注意：war包启动tomcat7及以下版本要关闭@configuration注解,否则将无法启动websocket服务。
3.websocket服务端实现
import java.io.ioexception;
import java.util.arraylist;
import java.util.list;
import java.util.map;
import java.util.concurrent.concurrenthashmap;
import javax.annotation.postconstruct;
import javax.annotation.resource;
import javax.websocket.onclose;
import javax.websocket.onerror;
import javax.websocket.onmessage;
import javax.websocket.onopen;
import javax.websocket.session;
import javax.websocket.server.pathparam;
import javax.websocket.server.serverendpoint;
import org.slf4j.logger;
import org.slf4j.loggerfactory;
import org.springframework.stereotype.component;
import com.kingengine.plug.service.messageservice;
import cn.hutool.core.util.strutil;
import net.sf.json.jsonarray;
import net.sf.json.jsonobject;
/**
* websocket服务类
* @author csf
* @date 2020年8月10日
*/
@serverendpoint("/websocket/{custid}")
@component
public class websocketserver
{
@resource
private messageservice messageservice;
logger log = loggerfactory.getlogger(this.getclass());
// 当前在线连接数
private static int onlinecount = 0;
// 存放每个用户对应的websocket连接对象,key为custid_hhmmss，确保一个登录用户只建立一个连接
private static map<string, session> websocketsessionmap = new concurrenthashmap<string, session>();
// 与某个客户端的连接会话，需要通过它来给客户端发送数据
private session session;
// 接收用户id
private string custid = "";
private static websocketserver websocketserver;
// 通过@postconstruct实现初始化bean之前进行的操作
@postconstruct
public void init()
{
// 初使化时将已静态化的websocketserver实例化
websocketserver = this;
websocketserver.messageservice = this.messageservice;
}
/**
* 连接建立成功调用的方法
* @param session 连接会话，由框架创建
* @param custid 用户id， 为处理用户多点登录都能收到消息，需传该格式custid_hhmmss
* @author csf
* @date 2020年8月10日
*/
@onopen
public void onopen(session session, @pathparam("custid") string custid)
{
if (!websocketsessionmap.containskey(custid))
{
this.session = session;
websocketsessionmap.put(custid, session);
addonlinecount(); // 在线数加1
log.info("有新连接[{}]接入,当前websocket连接数为:{}", custid, getonlinecount());
}
this.custid = custid;
try
{
// 第一次建立连接，推送消息给客户端，只会执行一次。后续的新消息由com.kingengine.plug.redis.redisreceiver接收到redis订阅消息推送
// 获取未读消息数
// 由于前端传进来的custid是有时间后缀的，查询时需要去掉后缀。
string qrycustid = custid.split("_")[0];
jsonobject unreadmsg = websocketserver.messageservice.getunreadcount(qrycustid);
// 获取最新消息
/* jsonobject newmsg = websocketserver.messageservice.getnewestmsg(qrycustid);
// 发送消息
jsonarray msgarr = new jsonarray();
if (newmsg!=null)
{
msgarr.add(newmsg);
}*/
jsonarray msgarr = new jsonarray();
msgarr.add(unreadmsg);
sendmessage(custid, msgarr.tostring());
}
catch (exception e)
{
log.error("客户端连接websocket服务异常");
e.printstacktrace();
}
}
/**
* 连接关闭调用的方法
*/
@onclose
public void onclose(@pathparam("custid") string sessionkey)
{
if (websocketsessionmap.containskey(sessionkey))
{
try
{
websocketsessionmap.get(sessionkey).close();
websocketsessionmap.remove(sessionkey);
}
catch (ioexception e)
{
log.error("连接[{}]关闭失败。", sessionkey);
e.printstacktrace();
}
subonlinecount();
log.info("连接[{}]关闭，当前websocket连接数：{}", sessionkey, onlinecount);
}
}
/**
* 接收客户端发送的消息
* @param message 客户端发送过来的消息
* @param session websocket会话
*/
@onmessage
public void onmessage(string message, session session)
{
log.info("收到来自客户端" + custid + "的信息:" + message);
}
/**
* 连接错误时触发
* @param session
* @param error
*/
@onerror
public void onerror(session session, throwable error)
{
try
{
session.close();
}
catch (ioexception e)
{
log.error("发生错误，连接[{}]关闭失败。");
e.printstacktrace();
}
// log.error("websocket发生错误");
// error.printstacktrace();
}
/**
* 给指定的客户端推送消息，可单发和群发
* @param sessionkeys 发送消息给目标客户端sessionkey，多个逗号“,”隔开1234,2345...
* @param message
* @throws ioexception
* @author csf
* @date 2020年8月11日
*/
public void sendmessage(string sessionkeys, string message)
{
if (strutil.isnotblank(sessionkeys))
{
string[] sessionkeyarr = sessionkeys.split(",");
for (string key : sessionkeyarr)
{
try
{
// 可能存在一个账号多点登录
list<session> sessionlist = getlikebymap(websocketsessionmap, key);
for (session session : sessionlist)
{
session.getbasicremote().sendtext(message);
}
}
catch (ioexception e)
{
e.printstacktrace();
continue;// 某个客户端发送异常，不影响其他客户端发送
}
}
}
else
{
log.info("sessionkeys为空，没有目标客户端");
}
}
/**
* 给当前客户端推送消息，首次建立连接时调用
*/
public void sendmessage(string message)
throws ioexception
{
this.session.getbasicremote().sendtext(message);
}
/**
* 检查websocket连接是否在线
* @param sesstionkey websocketmap中维护的key
* @return 是否在线
*/
public static boolean checkonline(string sesstionkey)
{
if (websocketsessionmap.containskey(sesstionkey))
{
return true;
}
else
{
return false;
}
}
/**
* 获取包含key的所有map值
* @param map
* @param keylike
* @return
* @author csf
* @date 2020年8月13日
*/
private list<session> getlikebymap(map<string, session> map, string keylike)
{
list<session> list = new arraylist<>();
for (string key : map.keyset())
{
if (key.contains(keylike))
{
list.add(map.get(key));
}
}
return list;
}
public static synchronized int getonlinecount()
{
return onlinecount;
}
public static synchronized void addonlinecount()
{
websocketserver.onlinecount++;
}
public static synchronized void subonlinecount()
{
websocketserver.onlinecount--;
}
}
4.redis消息订阅配置
import org.springframework.cache.annotation.enablecaching;
import org.springframework.context.annotation.bean;
import org.springframework.context.annotation.configuration;
import org.springframework.data.redis.connection.redisconnectionfactory;
import org.springframework.data.redis.core.stringredistemplate;
import org.springframework.data.redis.listener.patterntopic;
import org.springframework.data.redis.listener.redismessagelistenercontainer;
import org.springframework.data.redis.listener.adapter.messagelisteneradapter;
@configuration
@enablecaching
public class rediscacheconfig
{
@bean
redismessagelistenercontainer container(redisconnectionfactory connectionfactory, messagelisteneradapter listeneradapter)
{
redismessagelistenercontainer container = new redismessagelistenercontainer();
container.setconnectionfactory(connectionfactory);
// 可以添加多个 messagelistener，配置不同的交换机
container.addmessagelistener(listeneradapter, new patterntopic("shownewestmsg"));// 订阅最新消息频道
return container;
}
@bean
messagelisteneradapter listeneradapter(redisreceiver receiver)
{
// 消息监听适配器
return new messagelisteneradapter(receiver, "onmessage");
}
@bean
stringredistemplate template(redisconnectionfactory connectionfactory)
{
return new stringredistemplate(connectionfactory);
}
}5.redis配置，直接放在springboot项目application.properties或application.yml中
# 数据库索引（默认为0） spring.redis.database=0 spring.redis.host=192.168.1.100 spring.redis.port=6379 spring.redis.password=123456 # 连接池最大连接数（使用负值表示没有限制） spring.redis.pool.max-active=8 # 连接池最大阻塞等待时间（使用负值表示没有限制） spring.redis.pool.max-wait=-1 # 连接池中的最大空闲连接 spring.redis.pool.max-idle=8 # 连接池中的最小空闲连接 spring.redis.pool.min-idle=0 # 连接超时时间（毫秒） spring.redis.timeout=5000
6.接收消息生产者发布的消息，推送给对应的客户端
import java.io.unsupportedencodingexception;
import org.slf4j.logger;
import org.slf4j.loggerfactory;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.data.redis.connection.message;
import org.springframework.data.redis.connection.messagelistener;
import org.springframework.stereotype.component;
import com.kingengine.plug.websocket.websocketserver;
import cn.hutool.core.codec.base64;
import cn.hutool.core.util.strutil;
import net.sf.json.jsonarray;
import net.sf.json.jsonobject;
/**
* 消息监听对象，接收订阅消息
* @author csf
* @date 2020年8月13日
*/
@component
public class redisreceiver implements messagelistener
{
logger log = loggerfactory.getlogger(this.getclass());
@autowired
websocketserver websocketserver;
/**
* 处理接收到的订阅消息
*/
@override
public void onmessage(message message, byte[] pattern)
{
string channel = new string(message.getchannel());// 订阅的频道名称
string msg = "";
try
{
msg = new string(message.getbody(), "gbk");//注意与发布消息编码一致，否则会乱码
if (strutil.isnotblank(msg)){
if ("shownewestmsg".endswith(channel))// 最新消息
{
jsonobject json = jsonobject.fromobject(msg);
websocketserver.sendmessage(json.get("receive"),json.get("msg"));
}else{
//todo 其他订阅的消息处理
}
}else{
log.info("消息内容为空，不处理。");
}
}
catch (exception e)
{
log.error("处理消息异常："+e.tostring())
e.printstacktrace();
}
}
}
7.消息发布测试
import java.io.unsupportedencodingexception;
import java.util.hashmap;
import java.util.map;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.data.redis.core.stringredistemplate;
import org.springframework.web.bind.annotation.postmapping;
import org.springframework.web.bind.annotation.requestmapping;
import org.springframework.web.bind.annotation.restcontroller;
import net.sf.json.jsonobject;
@requestmapping("redis")
@restcontroller
public class redistestcontroller
{
@autowired
stringredistemplate template;
/**
* 发布消息测试
*@param userid
* @param msg
* @return
*/
@postmapping("sendmessage")
public string sendmessage(string userid,string msg)
{
try
{
string newmessge=new string(msg.getbytes("gbk"),"gbk");
map<string,string> map = new hashmap<string, string>();
map.put("receive", userid);
map.put("msg", newmessge);
template.convertandsend("shownewestmsg",
jsonobject.fromobject(map).tostring());
}
catch (unsupportedencodingexception e)
{
e.printstacktrace();
}
return "消息发布成功！";
}
}8.客户端代码
<!doctype html>
<html>
<head>
<title>websocket测试</title>
<meta http-equiv="content-type" content="text/html; charset=utf-8"/>
</head>
<body>
<div>
来自服务端消息：
<p id="message"></p>
</div>
</body>
<script src="http://apps.bdimg.com/libs/jquery/1.6.4/jquery.min.js"></script>
<script>
let websocketclient;
if (window.websocket)
{
let custid="132456_" + math.random();//该参数会作为websocketserver中存储session的key，要保证唯一。
websocketclient = new websocket("ws://127.0.0.1:8082/bootapp/websocket/" + custid);
//连通之后的回调事件
websocketclient.onopen = function () {
websocketclient.send("这里是地球，收到请回答。。。");
// websocket.send('{"type":"1","data":"121"}');
};
//接收后台服务端的消息
websocketclient.onmessage = function (evt) {
console.log("数据已接收:" + evt.data);
showmessage("未读消息：" + evt.data);
};
//连接关闭的回调事件
websocketclient.onclose = function () {
alert("连接已关闭...");
};
}else{
alert("浏览器不支持websocket");
}
function showmessage(message) {
$("#message").html(message);
}
</script>
</html>核心代码至此。
到此这篇关于redis做websocket分布式消息推送服务的实现的文章就介绍到这了,更多相关redis websocket分布式消息推送服务内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！
发表评论