内容简介:在上一篇文章(关于服务端如何给指定用户的客户端发送消息,一般可以通过以下三种方案来实现:注:本篇文章的完整源码可以参考:
在上一篇文章( www.zifangsky.cn/1355.html )中我介绍了在Spring项目中使用WebSocket的几种实现方式。但是,上篇文章中只介绍了服务端采用广播模式给所有客户端发送消息,然而我们有时需要服务端给指定用户的客户端发送消息(比如:发送Web通知、实时打印用户任务的日志、两个用户点对点聊天等)。
关于服务端如何给指定用户的客户端发送消息,一般可以通过以下三种方案来实现:
- 方案一 :WebSocket使用“Java提供的@ServerEndpoint注解”实现或者使用“Spring低层级API”实现,在建立连接时从
HttpSession
中获取用户登录后的用户名,然后把“ 用户名+该WebSocket连接 ”存储到ConcurrentHashMap
。给指定用户发送消息,只需要 根据接收者的用户名获取对方已经建立的WebSocket连接,接着给他发送消息即可 。 - 方案二 :在页面的监听路径前面动态添加当前登录的“ 用户ID/用户名 ”,这样给指定用户发送消息,只需要发送广播消息到监听了前面那个路径的客户端即可。
- 方案三 :这种方案类似于方案一。使用Spring的高级API实现WebSocket,然后自定义
HandshakeHandler
类并重写determineUser
方法,其目的是为了在建立连接时使用用户登录后的用户名作为此次WebSocket的凭证,最后我们就可以使用messagingTemplate.convertAndSendToUser
方法给指定用户发送消息了。
注:本篇文章的完整源码可以参考: github.com/zifangsky/W…
使用SimpMessagingTemplate发送消息
使用 org.springframework.messaging.simp.SimpMessagingTemplate
类可以在服务端的任意地方给客户端发送消息。此外,在我们配置Spring支持STOMP后 SimpMessagingTemplate
类就会被自动装配到Spring的上下文中,因此我们只需要在想要使用的地方使用 @Autowired
注解注入SimpMessagingTemplate即可使用。
需要说明的是, SimpMessagingTemplate
类有两个重要的方法,它们分别是:
-
public void convertAndSend(D destination, Object payload)
:给监听了路径destination
的所有客户端发送消息payload
-
public void convertAndSendToUser(String user, String destination, Object payload)
:给监听了路径destination
的用户user
发送消息payload
一个简单示例:
package cn.zifangsky.stompwebsocket.controller; import cn.zifangsky.stompwebsocket.model.websocket.Greeting; import cn.zifangsky.stompwebsocket.model.websocket.HelloMessage; import cn.zifangsky.stompwebsocket.service.RedisService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.messaging.simp.SimpMessagingTemplate; import org.springframework.messaging.simp.user.SimpUserRegistry; import org.springframework.stereotype.Controller; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.ResponseBody; import javax.annotation.Resource; /** * 测试{@link org.springframework.messaging.simp.SimpMessagingTemplate}类的基本用法 * @author zifangsky * @date 2018/10/10 * @since 1.0.0 */ @Controller @RequestMapping(("/wsTemplate")) public class MessageTemplateController { private final Logger logger = LoggerFactory.getLogger(getClass()); @Autowired private SimpMessagingTemplate messagingTemplate; @Autowired private SimpUserRegistry userRegistry; @Resource(name = "redisServiceImpl") private RedisService redisService; /** * 简单测试SimpMessagingTemplate的用法 */ @PostMapping("/greeting") @ResponseBody public String greeting(@RequestBody Greeting greeting) { this.messagingTemplate.convertAndSend("/topic/greeting", new HelloMessage("Hello," + greeting.getName() + "!")); return "ok"; } } 复制代码
很显然,这里发送的地址是上篇文章中最后那个示例监听的地址,在客户端页面建立连接后,我们使用 Postman
请求一下上面这个方法,效果如下:
然后我们可以发现页面中也收到消息了:
向指定用户发送WebSocket消息并处理对方不在线的情况
给指定用户发送消息:
- 如果接收者在线,则直接发送消息;
- 否则将消息存储到redis,等用户上线后主动拉取未读消息。
(1)自定义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” * * @author zifangsky * @date 2018/10/11 * @since 1.0.0 */ @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) { } } 复制代码
(2)自定义HandshakeHandler,用于在建立WebSocket的时候使用自定义的Principal:
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.stereotype.Component; import org.springframework.web.socket.WebSocketHandler; import org.springframework.web.socket.server.support.DefaultHandshakeHandler; import javax.servlet.http.HttpSession; import java.security.Principal; import java.text.MessageFormat; import java.util.Map; /** * 自定义{@link org.springframework.web.socket.server.support.DefaultHandshakeHandler},实现“生成自定义的{@link java.security.Principal}” * * @author zifangsky * @date 2018/10/11 * @since 1.0.0 */ @Component public class MyHandshakeHandler extends DefaultHandshakeHandler{ private final Logger logger = LoggerFactory.getLogger(getClass()); @Override protected Principal determineUser(ServerHttpRequest request, WebSocketHandler wsHandler, Map<String, Object> attributes) { HttpSession session = SpringContextUtils.getSession(); User loginUser = (User) session.getAttribute(Constants.SESSION_USER); if(loginUser != null){ logger.debug(MessageFormat.format("WebSocket连接开始创建Principal,用户:{0}", loginUser.getUsername())); return new MyPrincipal(loginUser.getUsername()); }else{ logger.error("未登录系统,禁止连接WebSocket"); return null; } } } 复制代码
相应地,这里的 MyPrincipal
继承了 java.security.Principal
类:
package cn.zifangsky.stompwebsocket.interceptor.websocket; import java.security.Principal; /** * 自定义{@link java.security.Principal} * * @author zifangsky * @date 2018/10/11 * @since 1.0.0 */ public class MyPrincipal implements Principal { private String loginName; public MyPrincipal(String loginName) { this.loginName = loginName; } @Override public String getName() { return loginName; } } 复制代码
(3)自定义ChannelInterceptor,用于在用户断开连接的时候记录日志:
package cn.zifangsky.stompwebsocket.interceptor.websocket; import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.simp.stomp.StompCommand; import org.springframework.messaging.simp.stomp.StompHeaderAccessor; import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.stereotype.Component; import java.security.Principal; import java.text.MessageFormat; /** * 自定义{@link org.springframework.messaging.support.ChannelInterceptor},实现断开连接的处理 * * @author zifangsky * @date 2018/10/10 * @since 1.0.0 */ @Component public class MyChannelInterceptor implements ChannelInterceptor{ private final Logger logger = LoggerFactory.getLogger(getClass()); @Override public void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex) { StompHeaderAccessor accessor = StompHeaderAccessor.wrap(message); StompCommand command = accessor.getCommand(); //用户已经断开连接 if(StompCommand.DISCONNECT.equals(command)){ String user = ""; Principal principal = accessor.getUser(); if(principal != null && StringUtils.isNoneBlank(principal.getName())){ user = principal.getName(); }else{ user = accessor.getSessionId(); } logger.debug(MessageFormat.format("用户{0}的WebSocket连接已经断开", user)); } } } 复制代码
(4)WebSocket相关的完整配置:
package cn.zifangsky.stompwebsocket.config; import cn.zifangsky.stompwebsocket.interceptor.websocket.AuthHandshakeInterceptor; import cn.zifangsky.stompwebsocket.interceptor.websocket.MyChannelInterceptor; import cn.zifangsky.stompwebsocket.interceptor.websocket.MyHandshakeHandler; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Configuration; import org.springframework.messaging.simp.config.ChannelRegistration; import org.springframework.messaging.simp.config.MessageBrokerRegistry; import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker; import org.springframework.web.socket.config.annotation.StompEndpointRegistry; import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer; /** * WebSocket相关配置 * * @author zifangsky * @date 2018/9/30 * @since 1.0.0 */ @Configuration @EnableWebSocketMessageBroker public class WebSocketConfig implements WebSocketMessageBrokerConfigurer{ @Autowired private AuthHandshakeInterceptor authHandshakeInterceptor; @Autowired private MyHandshakeHandler myHandshakeHandler; @Autowired private MyChannelInterceptor myChannelInterceptor; @Override public void registerStompEndpoints(StompEndpointRegistry registry) { registry.addEndpoint("/stomp-websocket").withSockJS(); registry.addEndpoint("/chat-websocket") .addInterceptors(authHandshakeInterceptor) .setHandshakeHandler(myHandshakeHandler) .withSockJS(); } @Override public void configureMessageBroker(MessageBrokerRegistry registry) { //客户端需要把消息发送到/message/xxx地址 registry.setApplicationDestinationPrefixes("/message"); //服务端广播消息的路径前缀,客户端需要相应订阅/topic/yyy这个地址的消息 registry.enableSimpleBroker("/topic"); } @Override public void configureClientInboundChannel(ChannelRegistration registration) { registration.interceptors(myChannelInterceptor); } } 复制代码
(5)Controller中的消息处理如下:
package cn.zifangsky.stompwebsocket.controller; import cn.zifangsky.stompwebsocket.common.Constants; import cn.zifangsky.stompwebsocket.common.SpringContextUtils; import cn.zifangsky.stompwebsocket.enums.ExpireEnum; import cn.zifangsky.stompwebsocket.model.User; import cn.zifangsky.stompwebsocket.model.websocket.Greeting; import cn.zifangsky.stompwebsocket.model.websocket.HelloMessage; import cn.zifangsky.stompwebsocket.service.RedisService; import cn.zifangsky.stompwebsocket.utils.JsonUtils; import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.messaging.simp.SimpMessagingTemplate; import org.springframework.messaging.simp.user.SimpUser; import org.springframework.messaging.simp.user.SimpUserRegistry; import org.springframework.stereotype.Controller; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.ResponseBody; import javax.annotation.Resource; import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpSession; import java.text.MessageFormat; import java.util.HashMap; import java.util.List; import java.util.Map; /** * 测试{@link org.springframework.messaging.simp.SimpMessagingTemplate}类的基本用法 * @author zifangsky * @date 2018/10/10 * @since 1.0.0 */ @Controller @RequestMapping(("/wsTemplate")) public class MessageTemplateController { private final Logger logger = LoggerFactory.getLogger(getClass()); @Autowired private SimpMessagingTemplate messagingTemplate; @Autowired private SimpUserRegistry userRegistry; @Resource(name = "redisServiceImpl") private RedisService redisService; /** * 简单测试SimpMessagingTemplate的用法 */ @PostMapping("/greeting") @ResponseBody public String greeting(@RequestBody Greeting greeting) { this.messagingTemplate.convertAndSend("/topic/greeting", new HelloMessage("Hello," + greeting.getName() + "!")); return "ok"; } /** * 给指定用户发送WebSocket消息 */ @PostMapping("/sendToUser") @ResponseBody public String chat(HttpServletRequest request) { //消息接收者 String receiver = request.getParameter("receiver"); //消息内容 String msg = request.getParameter("msg"); HttpSession session = SpringContextUtils.getSession(); User loginUser = (User) session.getAttribute(Constants.SESSION_USER); HelloMessage resultData = new HelloMessage(MessageFormat.format("{0} say: {1}", loginUser.getUsername(), msg)); this.sendToUser(loginUser.getUsername(), receiver, "/topic/reply", JsonUtils.toJson(resultData)); return "ok"; } /** * 给指定用户发送消息,并处理接收者不在线的情况 * @param sender 消息发送者 * @param receiver 消息接收者 * @param destination 目的地 * @param payload 消息正文 */ private void sendToUser(String sender, String receiver, String destination, String payload){ SimpUser simpUser = userRegistry.getUser(receiver); //如果接收者存在,则发送消息 if(simpUser != null && StringUtils.isNoneBlank(simpUser.getName())){ this.messagingTemplate.convertAndSendToUser(receiver, destination, payload); } //否则将消息存储到redis,等用户上线后主动拉取未读消息 else{ //存储消息的 Redis 列表名 String listKey = Constants.REDIS_UNREAD_MSG_PREFIX + receiver + ":" + destination; logger.info(MessageFormat.format("消息接收者{0}还未建立WebSocket连接,{1}发送的消息【{2}】将被存储到Redis的【{3}】列表中", receiver, sender, payload, listKey)); //存储消息到Redis中 redisService.addToListRight(listKey, ExpireEnum.UNREAD_MSG, payload); } } /** * 拉取指定监听路径的未读的WebSocket消息 * @param destination 指定监听路径 * @return java.util.Map<java.lang.String,java.lang.Object> */ @PostMapping("/pullUnreadMessage") @ResponseBody public Map<String, Object> pullUnreadMessage(String destination){ Map<String, Object> result = new HashMap<>(); try { HttpSession session = SpringContextUtils.getSession(); //当前登录用户 User loginUser = (User) session.getAttribute(Constants.SESSION_USER); //存储消息的Redis列表名 String listKey = Constants.REDIS_UNREAD_MSG_PREFIX + loginUser.getUsername() + ":" + destination; //从Redis中拉取所有未读消息 List<Object> messageList = redisService.rangeList(listKey, 0, -1); result.put("code", "200"); if(messageList !=null && messageList.size() > 0){ //删除Redis中的这个未读消息列表 redisService.delete(listKey); //将数据添加到返回集,供前台页面展示 result.put("result", messageList); } }catch (Exception e){ result.put("code", "500"); result.put("msg", e.getMessage()); } return result; } } 复制代码
注:这里对应的几个Redis操作的方法如下:
@Override public boolean delete(String key) { return redisTemplate.delete(key); } @Override public void addToListLeft(String listKey, ExpireEnum expireEnum, Object... values) { //绑定操作 BoundListOperations<String, Object> boundValueOperations = redisTemplate.boundListOps(listKey); //插入数据 boundValueOperations.leftPushAll(values); //设置过期时间 boundValueOperations.expire(expireEnum.getTime(),expireEnum.getTimeUnit()); } @Override public void addToListRight(String listKey, ExpireEnum expireEnum, Object... values) { //绑定操作 BoundListOperations<String, Object> boundValueOperations = redisTemplate.boundListOps(listKey); //插入数据 boundValueOperations.rightPushAll(values); //设置过期时间 boundValueOperations.expire(expireEnum.getTime(),expireEnum.getTimeUnit()); } @Override public List<Object> rangeList(String listKey, long start, long end) { //绑定操作 BoundListOperations<String, Object> boundValueOperations = redisTemplate.boundListOps(listKey); //查询数据 return boundValueOperations.range(start, end); } 复制代码
(6)示例页面:
<html xmlns:th="http://www.thymeleaf.org"> <head> <meta content="text/html;charset=UTF-8"/> <meta http-equiv="Content-Type" content="text/html; charset=utf-8"/> <meta http-equiv="X-UA-Compatible" content="IE=edge"/> <meta name="viewport" content="width=device-width, initial-scale=1"/> <title>Chat With STOMP Message</title> <script src="https://cdnjs.cloudflare.com/ajax/libs/jquery/3.3.1/jquery.min.js"></script> <script src="https://cdnjs.cloudflare.com/ajax/libs/sockjs-client/1.1.4/sockjs.min.js"></script> <script src="https://cdnjs.cloudflare.com/ajax/libs/stomp.js/2.3.3/stomp.min.js"></script> <script th:src="@{/layui/layui.js}"></script> <script th:src="@{/layui/lay/modules/layer.js}"></script> <link th:href="@{/layui/css/layui.css}" rel="stylesheet"> <link th:href="@{/layui/css/modules/layer/default/layer.css}" rel="stylesheet"> <link th:href="@{/css/style.css}" rel="stylesheet"> <style type="text/css"> #connect-container { margin: 0 auto; width: 400px; } #connect-container div { padding: 5px; margin: 0 7px 10px 0; } .message input { padding: 5px; margin: 0 7px 10px 0; } .layui-btn { display: inline-block; } </style> <script type="text/javascript"> var stompClient = null; $(function () { var target = $("#target"); if (window.location.protocol === 'http:') { target.val('http://' + window.location.host + target.val()); } else { target.val('https://' + window.location.host + target.val()); } }); function setConnected(connected) { var connect = $("#connect"); var disconnect = $("#disconnect"); var echo = $("#echo"); if (connected) { connect.addClass("layui-btn-disabled"); disconnect.removeClass("layui-btn-disabled"); echo.removeClass("layui-btn-disabled"); } else { connect.removeClass("layui-btn-disabled"); disconnect.addClass("layui-btn-disabled"); echo.addClass("layui-btn-disabled"); } connect.attr("disabled", connected); disconnect.attr("disabled", !connected); echo.attr("disabled", !connected); } //连接 function connect() { var target = $("#target").val(); var ws = new SockJS(target); stompClient = Stomp.over(ws); stompClient.connect({}, function () { setConnected(true); log('Info: STOMP connection opened.'); //连接成功后,主动拉取未读消息 pullUnreadMessage("/topic/reply"); //订阅服务端的/topic/reply地址 stompClient.subscribe("/user/topic/reply", function (response) { log(JSON.parse(response.body).content); }) },function () { //断开处理 setConnected(false); log('Info: STOMP connection closed.'); }); } //断开连接 function disconnect() { if (stompClient != null) { stompClient.disconnect(); stompClient = null; } setConnected(false); log('Info: STOMP connection closed.'); } //向指定用户发送消息 function sendMessage() { if (stompClient != null) { var receiver = $("#receiver").val(); var msg = $("#message").val(); log('Sent: ' + JSON.stringify({'receiver': receiver, 'msg':msg})); $.ajax({ url: "/wsTemplate/sendToUser", type: "POST", dataType: "json", async: true, data: { "receiver": receiver, "msg": msg }, success: function (data) { } }); } else { layer.msg('STOMP connection not established, please connect.', { offset: 'auto' ,icon: 2 }); } } //从服务器拉取未读消息 function pullUnreadMessage(destination) { $.ajax({ url: "/wsTemplate/pullUnreadMessage", type: "POST", dataType: "json", async: true, data: { "destination": destination }, success: function (data) { if (data.result != null) { $.each(data.result, function (i, item) { log(JSON.parse(item).content); }) } else if (data.code !=null && data.code == "500") { layer.msg(data.msg, { offset: 'auto' ,icon: 2 }); } } }); } //日志输出 function log(message) { console.debug(message); } </script> </head> <body> <noscript><h2 style="color: #ff0000">Seems your browser doesn't support Javascript! Websockets rely on Javascript being enabled. Please enable Javascript and reload this page!</h2></noscript> <div> <div id="connect-container" class="layui-elem-field"> <legend>Chat With STOMP Message</legend> <div> <input id="target" type="text" class="layui-input" size="40" style="width: 350px" value="/chat-websocket"/> </div> <div> <button id="connect" class="layui-btn layui-btn-normal" onclick="connect();">Connect</button> <button id="disconnect" class="layui-btn layui-btn-normal layui-btn-disabled" disabled="disabled" onclick="disconnect();">Disconnect </button> </div> <div class="message"> <input id="receiver" type="text" class="layui-input" size="40" style="width: 350px" placeholder="接收者姓名" value=""/> <input id="message" type="text" class="layui-input" size="40" style="width: 350px" placeholder="消息内容" value=""/> </div> <div> <button id="echo" class="layui-btn layui-btn-normal layui-btn-disabled" disabled="disabled" onclick="sendMessage();">Send Message </button> </div> </div> </div> </body> </html> 复制代码
启动项目后,分别在两个浏览器中使用不同的账号登录,接着互相给对方发送消息,效果如下:
界面一:
界面二:
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持 码农网
猜你喜欢:- [译]WebRTC基础实践 - 9. 拍照并传给对方
- NASA被黑客盗取500MB任务文件,对方只用了个Raspberry Pi
- c# – “SMTP主机未指定” – 但是是否指定?
- 运维安全 | 如何限制指定账户不能SSH只能SFTP在指定目录
- Zabbix监控指定端口
- Android指定专用APN
本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们。
Beginning PHP and MySQL 5
W. Jason Gilmore / Apress / 2006-01-23 / USD 44.99
Beginning PHP and MySQL 5: From Novice to Professional, Second Edition offers comprehensive information about two of the most prominent open source technologies on the planet: the PHP scripting langua......一起来看看 《Beginning PHP and MySQL 5》 这本书的介绍吧!