<!-- WebSocket--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-websocket</artifactId> </dependency> <dependency> <groupId>org.redisson</groupId> <artifactId>redisson-spring-boot-starter</artifactId> </dependency>
1.开启WebSocket支持
import org.springframework.boot.web.servlet.ServletContextInitializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
import javax.servlet.ServletContext;
import javax.servlet.ServletException;
/**
* @Description: 开启WebSocket支持
*/
@Configuration
public class WebSocketConfig implements ServletContextInitializer {
/**
* 这个bean的注册,用于扫描带有@ServerEndpoint的注解成为websocket,如果你使用外置的tomcat就不需要该配置文件
*/
@Bean
public ServerEndpointExporter serverEndpointExporter() {
return new ServerEndpointExporter();
}
@Override
public void onStartup(ServletContext servletContext) throws ServletException {
}
}
2.WebSocket操作类
import com.yami.shop.admin.dict.DictWebSocket;
import com.yami.shop.admin.util.StringUtils;
import com.yami.shop.admin.util.WebSocketRedisUtil;
import com.yami.shop.sys.model.SysUser;
import com.yami.shop.sys.service.SysUserService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import net.sf.json.JSONObject;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
/**
* @Description: WebSocket操作类
*/
@ServerEndpoint("/websocket/{userId}")
@Component
@Slf4j
public class WebSocketSever {
private static SysUserService sysUserService;
// 注入的时候,给类的 service 注入
@Autowired
public void setTestService(SysUserService sysUserService) {
WebSocketSever.sysUserService = sysUserService;
}
// 与某个客户端的连接会话,需要通过它来给客户端发送数据
private Session session;
// session集合,存放对应的session
private static ConcurrentHashMap<Integer, Session> sessionPool = new ConcurrentHashMap<>();
// concurrent包的线程安全Set,用来存放每个客户端对应的WebSocket对象。
private static CopyOnWriteArraySet<WebSocketSever> webSocketSet = new CopyOnWriteArraySet<>();
/**
* 缓存的消息,因为会存在同时写消息-读消息,写消息-删除消息的情况,需要保证线程安全*
*/
// private static ConcurrentHashMap<String, List<String>> cacheMessage = new ConcurrentHashMap<>();
/**
* 建立WebSocket连接
*
* @param session
* @param userId 用户ID
*/
@OnOpen
public void onOpen(Session session, @PathParam(value = "userId") Integer userId) {
log.info("WebSocket建立连接中,连接用户ID:{}", userId);
SysUser user = sysUserService.getSysUserById(Long.valueOf(userId));
if (user == null) {
log.error(DictWebSocket.WEB_SOCKET_USERID +"不正确");
return;
}
try {
Session historySession = sessionPool.get(userId);
// historySession不为空,说明已经有人登陆账号,应该删除登陆的WebSocket对象
if (historySession != null) {
webSocketSet.remove(historySession);
historySession.close();
}
} catch (IOException e) {
log.error("重复登录异常,错误信息:" + e.getMessage(), e);
}
// 建立连接
this.session = session;
webSocketSet.add(this);
sessionPool.put(userId, session);
log.info("建立连接完成,当前在线人数为:{}", webSocketSet.size());
// testMassage();
cacheMessageContains();
}
/**
* 查询是否有离线消息并推送
*
*/
public static void cacheMessageContains(){
//是否有暂存的消息,如果有则发送消息
// boolean contains = RedisUtil.lGetListSize(DictWebSocket.OFFLINE_MESSAGE);
List<Object> Strings = WebSocketRedisUtil.getCacheChatMessage(DictWebSocket.OFFLINE_MESSAGE);
if (Strings!=null) {
//取出消息列表
List<Object> list = Strings;
if (list == null) {
log.info("暂无缓存的消息");
}
list.forEach(message -> {
//暂时群发消息
sendAllMessage(message.toString());
});
log.info("用户缓存的消息发送成功一共:"+list.size()+"条");
list = null;
WebSocketRedisUtil.deleteCacheChatMessage(DictWebSocket.OFFLINE_MESSAGE);
}
}
/**
* 暂存离线消息
*
*/
public static void cacheMessagePut (String message){
// RedisUtil
//获取消息列表
// List<String> list = cacheMessage.get(DictWebSocket.OFFLINE_MESSAGE);
// if (list == null) {
// // list会存在并发修改异常,需要一个线程安全的List
// list = new CopyOnWriteArrayList<>();
// cacheMessage.put(DictWebSocket.OFFLINE_MESSAGE, list);
// }
// //把新消息添加到消息列表
// list.add(message);
if (!StringUtils.isEmpty(message)){
boolean isCache = WebSocketRedisUtil.saveCacheChatMessage(DictWebSocket.OFFLINE_MESSAGE, message);
if (isCache){
log.info("消息暂存成功" + message);
}else{
log.error("消息暂存失败" + message);
}
}
}
/**
* 群发消息
*
* @param message 发送的消息
*/
public static void sendAllMessage(String message) {
if (webSocketSet.size()<=0){
cacheMessagePut(message);
return;
}
log.info("发送消息:{}", message);
Integer index=0;
for (WebSocketSever webSocket : webSocketSet) {
try {
webSocket.session.getBasicRemote().sendText(message);
index++;
} catch (IOException e) {
log.error("群发消息发生错误:" + e.getMessage(), e);
}
}
log.info("总共发送 "+index+" 条");
}
/**
* 发生错误
*
* @param throwable e
*/
@OnError
public void onError(Throwable throwable) {
throwable.printStackTrace();
}
/**
* 连接关闭
*/
@OnClose
public void onClose() {
webSocketSet.remove(this);
log.info("连接断开,当前在线人数为:{}", webSocketSet.size());
}
/**
* 接收客户端消息
*
* @param message 接收的消息
*/
@OnMessage
public void onMessage(String message, Session session) {
try {
String userId = session.getPathParameters().get(DictWebSocket.WEB_SOCKET_USERID);
userSession(Integer.valueOf(userId),message);
log.info("session==<><><:{}", session);
log.info("收到客户端 "+userId+" 发来的消息:{}", message);
if (!StringUtils.isEmpty(message)) {
JSONObject jsonObject = JSONObject.fromObject(message);
if (jsonObject.get(DictWebSocket.PING).equals(true)) {
pingMassage(Integer.valueOf(userId));
}
}
} catch (Exception e) {
}
}
/**
* 查询接收到的消息发送者是否已链接 未连接无视消息
*/
public static void userSession(Integer userId,String message){
Session historySession = sessionPool.get(userId);
if (historySession!=null){
return;
}
log.error("收到未连接用户"+userId+" 发来的消息:{}", message);
throw new RuntimeException();
}
/**
* 保持客户端心跳链接
*/
public static void pingMassage(Integer userId) {
Map<String, Object> map = new HashMap<>();
map.put(DictWebSocket.PING, true);
map.put(DictWebSocket.WEB_SOCKET_USERID, userId);
map.put("msg", "心跳链接");
sendMessageByUser(userId, JSONObject.fromObject(map).toString());
}
//在一个特定的时间执行这个方法
//cron 表达式
/**
* 推送消息到指定用户
*
* @param userId 用户ID
* @param message 发送的消息
*/
public static void sendMessageByUser(Integer userId, String message) {
log.info("用户ID:" + userId + ",推送内容:" + message);
Session session = sessionPool.get(userId);
try {
session.getBasicRemote().sendText(message);
} catch (IOException e) {
log.error("推送消息到指定用户发生错误:" + e.getMessage(), e);
}
}
}
3.WebSocket消息机制redis工具类
import com.yami.shop.common.util.RedisUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.util.List;
/**
* WebSocket消息推送redis工具类
*
* @author ruoyi
*/
@Component
@Slf4j
public class WebSocketRedisUtil {
/**
* 功能描述:将JavaBean对象的信息缓存进Redis
*
* @param message 信息JavaBean
* @return 是否保存成功
*/
public static boolean saveCacheChatMessage(String key, String message) {
//判断key是否存在
if (RedisUtil.hasKey(key)) {
//将javabean对象添加到缓存的list中
long redisSize = RedisUtil.lGetListSize(key);
System.out.println("redis当前数据条数" + redisSize);
Long index = RedisUtil.rightPushValue(key, message);
System.out.println("redis执行rightPushList返回值:" + index);
return redisSize<index;
} else {
//不存在key时,将chatVO存进缓存,并设置过期时间
// JSONArray jsonArray=new JSONArray();
// jsonArray.add(message);
boolean isCache = RedisUtil.lSet(key, message);
//保存成功,设置过期时间 暂不设置失效时间
if (isCache) {
// RedisUtil.expire(key, 3L, TimeUnit.DAYS);
System.out.println("存储成功"+message);
}
return isCache;
}
}
/**
* 功能描述:从缓存中读取信息
*
* @param key 缓存信息的键
* @return 缓存中信息list
*/
public static List<Object> getCacheChatMessage(String key) {
List<Object> chatList = null;
//判断key是否存在
if (RedisUtil.hasKey(key)) {
chatList = RedisUtil.getOpsForList(key);
} else {
log.info("redis缓存中无此键值:" + key);
}
return chatList;
}
/**
* 功能描述: 在缓存中删除信息
*
* @param key 缓存信息的键
*/
public static void deleteCacheChatMessage(String key) {
//判断key是否存在
if (RedisUtil.hasKey(key)) {
RedisUtil.del(key);
}
}
}
4.RedisUtil普通redis工具类
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.util.CollectionUtils;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
/**
* @author lh
*/
@Slf4j
public class RedisUtil {
private static RedisTemplate<String, Object> redisTemplate = SpringContextUtils.getBean("redisTemplate", RedisTemplate.class);
public static final StringRedisTemplate STRING_REDIS_TEMPLATE = SpringContextUtils.getBean("stringRedisTemplate",StringRedisTemplate.class);
//=============================common============================
/**
* 指定缓存失效时间
*
* @param key 键
* @param time 时间(秒)
* @return
*/
public static boolean expire(String key, long time) {
try {
if (time > 0) {
redisTemplate.expire(key, time, TimeUnit.SECONDS);
}
return true;
} catch (Exception e) {
log.error("设置redis指定key失效时间错误:", e);
return false;
}
}
/**
* 根据key 获取过期时间
*
* @param key 键 不能为null
* @return 时间(秒) 返回0代表为永久有效 失效时间为负数,说明该主键未设置失效时间(失效时间默认为-1)
*/
public static Long getExpire(String key) {
return redisTemplate.getExpire(key, TimeUnit.SECONDS);
}
/**
* 判断key是否存在
*
* @param key 键
* @return true 存在 false 不存在
*/
public static Boolean hasKey(String key) {
try {
return redisTemplate.hasKey(key);
} catch (Exception e) {
log.error("redis判断key是否存在错误:", e);
return false;
}
}
/**
* 删除缓存
*
* @param key 可以传一个值 或多个
*/
@SuppressWarnings("unchecked")
public static void del(String... key) {
if (key != null && key.length > 0) {
if (key.length == 1) {
redisTemplate.delete(key[0]);
} else {
redisTemplate.delete(Arrays.asList(key));
}
}
}
//============================String=============================
/**
* 普通缓存获取
*
* @param key 键
* @return 值
*/
@SuppressWarnings("unchecked")
public static <T> T get(String key) {
return key == null ? null : (T) redisTemplate.opsForValue().get(key);
}
/**
* 普通缓存放入
*
* @param key 键
* @param value 值
* @return true成功 false失败
*/
public static boolean set(String key, Object value) {
try {
redisTemplate.opsForValue().set(key, value);
return true;
} catch (Exception e) {
log.error("设置redis缓存错误:", e);
return false;
}
}
/**
* 普通缓存放入并设置时间
*
* @param key 键
* @param value 值
* @param time 时间(秒) time要大于0 如果time小于等于0 将设置无限期
* @return true成功 false 失败
*/
public static boolean set(String key, Object value, long time) {
try {
if (time > 0) {
redisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS);
} else {
set(key, value);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 递增 此时value值必须为int类型 否则报错
*
* @param key 键
* @param delta 要增加几(大于0)
* @return
*/
public static Long incr(String key, long delta) {
if (delta < 0) {
throw new RuntimeException("递增因子必须大于0");
}
return STRING_REDIS_TEMPLATE.opsForValue().increment(key, delta);
}
/**
* 递减
*
* @param key 键
* @param delta 要减少几(小于0)
* @return
*/
public static Long decr(String key, long delta) {
if (delta < 0) {
throw new RuntimeException("递减因子必须大于0");
}
return STRING_REDIS_TEMPLATE.opsForValue().increment(key, -delta);
}
// ===================================自定义工具扩展===========================================
/**
* HashGet
*
* @param key 键 不能为null
* @param item 项 不能为null
* @return 值
*/
public Object hget ( String key, String item ) {
return redisTemplate.opsForHash().get(key, item);
}
// /**
// * 获取hashKey对应的所有键值
// *
// * @param key 键
// * @return 对应的多个键值
// */
// public static Map<Object, Object> hmget (String key ) {
// return redisTemplate.opsForHash().entries(key);
// }
/**
* 获取hashKey对应的所有键值
*
* @param key 键
* @return 对应的多个键值
// */
// public static List<Object> hmget (String key ) {
// return redisTemplate.opsForList().ge;
// }
/**
* 获取list缓存的长度
*
* @param key 键
* @return
*/
public static long lGetListSize ( String key ) {
try {
return redisTemplate.opsForList().size(key);
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
/**
* 功能描述:在list的右边添加元素
* 如果键不存在,则在执行推送操作之前将其创建为空列表
*
* @param key 键
* @return value 值
* @author RenShiWei
* Date: 2020/2/6 23:22
*/
public static Long rightPushValue ( String key, Object value ) {
return redisTemplate.opsForList().rightPush(key, value);
}
/**
* 功能描述:获取缓存中所有的List key
*
* @param key 键
*/
public static List<Object> getOpsForList ( String key) {
return redisTemplate.opsForList().range(key, 0, redisTemplate.opsForList().size(key));
}
/**
* 将list放入缓存
*
* @param key 键
* @param value 值
* @return
*/
public static boolean lSet ( String key, Object value ) {
try {
redisTemplate.opsForList().rightPush(key, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 将list放入缓存
*
* @param key 键
* @param value 值
* @return
*/
public boolean lSet ( String key, List<Object> value ) {
try {
redisTemplate.opsForList().rightPushAll(key, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
}
5.DictWebSocket key值
public interface DictWebSocket {
public String WEB_SOCKET_USERID="userId";//链接编号
public String PING="ping";//客户端心跳
public String SOCKET_MESSAGE="socketMessage";//提示客户端展示消息
public String OFFLINE_MESSAGE="offlineMessage";//提示客户端展示消息
}
vue端涉及业务就不贴了
WebSocket是一种在单个TCP连接上进行全双工通信的协议。WebSocket通信协议于2011年被IETF定为标准RFC 6455,并由RFC7936补充规范。WebSocket API也被W3C定为标准。文章来源:https://www.toymoban.com/news/detail-483390.html
WebSocket使得客户端和服务器之间的数据交换变得更加简单,允许服务端主动向客户端推送数据。在WebSocket API中,浏览器和服务器只需要完成一次握手,两者之间就直接可以创建持久性的连接,并进行双向数据传输。文章来源地址https://www.toymoban.com/news/detail-483390.html
到了这里,关于WebSocket+Redis实现消息推送机制以及离线消息推送(vue+sping boot)的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!