转载

Spring Websocket实现向指定的用户发送消息

概述

本文我们介绍通过Spring websocket实现向特定的用户发送消息。

本文的内容如下：

1. 首先实现简单的登录功能，这里向特定用户发送消息的必要条件

2. 用户登录系统后，才可以登录websocket，并重写MyPrincipal

3. 实现向特定用户发送消息的功能

4. 测试

首先实现简单的登录功能，这是向特定用户发送消息的必要条件

TestMQCtl：控制类

提供模拟登录，登录成功后转到websocket页面

/**
     * 模拟登录     */
    @RequestMapping(value = "loginIn", method = RequestMethod.POST)
    public String login(HttpServletRequest request, @RequestParam(required=true) String name, String pwd){
        HttpSession httpSession = request.getSession();
        // 如果登录成功，则保存到会话中
        httpSession.setAttribute("loginName", name);
        return "websocket/sendtouser/ws-sendtouser-rabbitmq";
    }

    /**
     * 转到登录页面
     */
    @RequestMapping(value = "login", method = RequestMethod.GET)
    public String loginPage(){
        // 转到登录页面
        return "websocket/sendtouser/login";
    }

    /**
     * websocket页面
     * @return
     */
    @RequestMapping(value="/broadcast-rabbitmq/index")
    public String broadcastIndex(){
        return "websocket/sendtouser/ws-sendtouser-rabbitmq";
    }

简单的form表单，将请求提到loginIn，并转到ws-sendtouser-rabbitmq.jsp页面

<form action="loginIn" method="post">
    用户名:<input type="text" name="name" />
    <p>
        密码:<input type="password" name="password" />
    <p>
        <input type="submit" value="submit" />
</form>

ws-sendtouser-rabbitmq.jsp

连接websocket并订阅消息，这个jsp之前的文章已经介绍过了这里不详细描述。页面通过向/ws/icc/websocket启动websocket，然后订阅/user/topic/demo消息

<script type="text/javascript">
    var stompClient = null;

    function setConnected(connected) {
        document.getElementById('connect').disabled = connected;
        document.getElementById('disconnect').disabled = !connected;
        document.getElementById('conversationDiv').style.visibility = connected ? 'visible' : 'hidden';
        $('#response').html();
    }

    function connect() {
        // websocket的连接地址，此值等于WebSocketMessageBrokerConfigurer中registry.addEndpoint("/ws/icc/websocket").withSockJS()配置的地址
        var socket = new SockJS('/ws/icc/websocket'); //1
        stompClient = Stomp.over(socket);
        stompClient.connect({}, function(frame) {
            setConnected(true);
            console.log('Connected: ' + frame);
            // 客户端订阅消息的目的地址：此值等于BroadcastCtl中@SendTo注解的里配置的值。
            stompClient.subscribe(
                '/user/topic/demo',
                function(respnose){
                showResponse(JSON.parse(respnose.body));
                }
                );
        });
    }


    function disconnect() {
        if (stompClient != null) {
            stompClient.disconnect();
        }
        setConnected(false);
        console.log("Disconnected");
    }

    function showResponse(message) {
        var response = $("#response");
        response.html(message.name + "<br>" + response.html());
    }
</script>

用户登录系统后，才可以登录websocket，并重写MyPrincipal

AuthHandshakeInterceptor

AuthHandshakeInterceptor是HandshakeInterceptor 的子类。在websocket握手前判断，判断当前用户是否已经登录。如果未登录，则不允许登录websocket

@Component
public class AuthHandshakeInterceptor implements HandshakeInterceptor {
    private static final Logger log = LoggerFactory.getLogger(AuthHandshakeInterceptor.class);


    @Override
    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Map<String, Object> attributes) throws Exception {
        HttpSession httpSession = getSession(request);
        String user = (String)httpSession.getAttribute("loginName");

        if(StringUtils.isEmpty(user)){
            log.error("未登录系统，禁止登录websocket!");
            return false;
        }
        log.info("login = " + user);

        return true;
    }

    @Override
    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Exception exception) {
    }

    // 参考 HttpSessionHandshakeInterceptor
    private HttpSession getSession(ServerHttpRequest request) {
        if (request instanceof ServletServerHttpRequest) {
            ServletServerHttpRequest serverRequest = (ServletServerHttpRequest) request;
            return serverRequest.getServletRequest().getSession(false);
        }
        return null;
    }
}

MyPrincipalHandshakeHandler

MyPrincipalHandshakeHandler是DefaultHandshakeHandler 的子类，处理websocket请求，这里我们只重写determineUser方法，生成我们自己的Principal ，这里我们使用loginName标记登录用户，而不是默认值

@Component
public class MyPrincipalHandshakeHandler extends DefaultHandshakeHandler {
    private static final Logger log = LoggerFactory.getLogger(MyPrincipalHandshakeHandler.class);

    @Override
    protected Principal determineUser(ServerHttpRequest request, WebSocketHandler wsHandler, Map<String, Object> attributes) {

        HttpSession httpSession = getSession(request);
        String user = (String)httpSession.getAttribute("loginName");

        if(StringUtils.isEmpty(user)){
            log.error("未登录系统，禁止登录websocket!");
            return null;
        }
        log.info(" MyDefaultHandshakeHandler login = " + user);
        return new MyPrincipal(user);
    }

    private HttpSession getSession(ServerHttpRequest request) {
        if (request instanceof ServletServerHttpRequest) {
            ServletServerHttpRequest serverRequest = (ServletServerHttpRequest) request;
            return serverRequest.getServletRequest().getSession(false);
        }
        return null;
    }
}

MyPrincipal

定义自己的Principal

public class MyPrincipal implements Principal {
    private String loginName;

    public MyPrincipal(String loginName){
        this.loginName = loginName;
    }
    @Override
    public String getName() {
        return loginName;
    }
}

配置websocket

在registerStompEndpoints中将我们MyPrincipalHandshakeHandler 和AuthHandshakeInterceptor 配置到服务中

configureMessageBroker方法配置rabbitmq信息，这里略

@Configuration
// 此注解开使用STOMP协议来传输基于消息代理的消息，此时可以在@Controller类中使用@MessageMapping
@EnableWebSocketMessageBroker
public class WebSocketRabbitMQMessageBrokerConfigurer extends AbstractWebSocketMessageBrokerConfigurer {

    @Autowired
    private MyPrincipalHandshakeHandler myDefaultHandshakeHandler;
    @Autowired
    private AuthHandshakeInterceptor sessionAuthHandshakeInterceptor;

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {

        registry.addEndpoint("/ws/icc/websocket")
                .addInterceptors(sessionAuthHandshakeInterceptor)
                .setHandshakeHandler(myDefaultHandshakeHandler)
                .withSockJS();
    }
    …. 
}

实现向特定用户发送消息的功能

TestMQCtl：

登录到模拟发送页面：send.jsp

我们使用SimpMessagingTemplate 对象的convertAndSendToUser向指定用户的/topic/demo发送消息

@Autowired
    private SimpMessagingTemplate template;

    /**
     * 发送页面
     */
    @RequestMapping(value = "send")
    public String sendMq2UserPage(String msg, String userName){
        return "websocket/sendtouser/send";
    }
    /**
     * 向执行用户发送请求
     */
    @RequestMapping(value = "send2user")
    @ResponseBody
    public int sendMq2User(String msg, String name){
        System.out.println("===========" + msg + "=======" + name);
        RequestMessage demoMQ = new RequestMessage();
        demoMQ.setName(msg);
        template.convertAndSendToUser(name, "/topic/demo", JSON.toJSONString(demoMQ));
        return 0;
    }

send.jsp

模拟发送页面

<form action="login" method="post">
        接收者用户:<input type="text" id="name" name="name" value="<%=session.getAttribute("loginName") %>" />
    <p>
        消息内容:<input type="text" id="msg" name="msg" />
    <p>
        <input type="button" id="send" value="发送" />
</form>


<script src="/websocket/jquery.js"></script>
<script type=text/javascript>

    $("#send").click(function(){
        $.post("send2user",
            {
                name: $('#name').val(),
                msg: $('#msg').val()
            },
            function(data, status){
                alert("Data: " + data + "nStatus: " + status);
            });
    });
</script>

测试

测试一：

点击连接，如果连接变灰色，则登录websocket成功

登录模拟发送页面http://127.0.0.1:8080/ws/send，向xiaoming发送test-msg

此时页面收到信息：

在模拟界面，如果我们向其它用户发送信息，则此界面不会收到信息

测试二：

打开两个不同的浏览器，分别使用xiaoming1,xiaoming2登录系统，

使用模拟界面向xiaoming1发送消息，则只有xiaoming1收到

使用模拟界面向xiaoming2发送消息，则只有xiaoming2收到

结论：

我们已经实现向特定的用户发送消息的功能