WebSocket多人在线聊天室（SpringBoot+WebSocket）

1.前言

本文中搭建了一个简易的多人聊天室，使用了WebSocket的基础特性。

源代码来自老外的一篇好文：

https://www.callicoder.com/spring-boot-websocket-chat-example/

2.本文内容摘要：

初步理解WebSocket的前后端交互逻辑手把手使用 SpringBoot + WebSocket 搭建一个多人聊天室Demo 代码源码及其解释前端展示页面

此外，在下一篇文章中，我们将做到：

对该WebSocket聊天室进行分布式改造，同时部署多台机器来作为集群，支撑高并发。保存用户session，并且在集群上实现session同步，比如实时展示当前在线的用户！

3.正文

WebSocket多人在线聊天室本文工程源代码：

https://github.com/qqxx6661/springboot-websocket-demo

新建工程

我们新建一个SpringBoot2的项目工程，在默认依赖中，添加websocket依赖：

```xml org.springframework.boot spring-boot-starter-websocket

```

WebSocket 配置

我们先来设置websocket的配置，新建config文件夹，在里面新建类WebSocketConfig

```java import org.springframework.context.annotation.Configuration; import org.springframework.messaging.simp.config.MessageBrokerRegistry; import org.springframework.web.socket.config.annotation.*;

@Configuration @EnableWebSocketMessageBroker public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

@Override
public void registerStompEndpoints(StompEndpointRegistry registry) {
    registry.addEndpoint("/ws").withSockJS();
}

@Override
public void configureMessageBroker(MessageBrokerRegistry registry) {
    registry.setApplicationDestinationPrefixes("/app");
    registry.enableSimpleBroker("/topic");
}

}

```

代码解释：

@EnableWebSocketMessageBroker用于启用我们的WebSocket服务器。

我们实现了WebSocketMessageBrokerConfigurer接口，并实现了其中的方法。

在第一种方法中，我们注册一个websocket端点，客户端将使用它连接到我们的websocket服务器。

withSockJS()是用来为不支持websocket的浏览器启用后备选项，使用了SockJS。

方法名中的STOMP是来自Spring框架STOMP实现。 STOMP代表简单文本导向的消息传递协议。它是一种消息传递协议，用于定义数据交换的格式和规则。为啥我们需要这个东西？因为WebSocket只是一种通信协议。它没有定义诸如以下内容：如何仅向订阅特定主题的用户发送消息，或者如何向特定用户发送消息。我们需要STOMP来实现这些功能。

在configureMessageBroker方法中，我们配置一个消息代理，用于将消息从一个客户端路由到另一个客户端。

第一行定义了以“/app”开头的消息应该路由到消息处理方法（之后会定义这个方法）。

第二行定义了以“/topic”开头的消息应该路由到消息代理。消息代理向订阅特定主题的所有连接客户端广播消息。

在上面的示例中，我们使用的是内存中的消息代理。

之后也可以使用RabbitMQ或ActiveMQ等其他消息代理。

创建 ChatMessage 实体

ChatMessage用来在客户端和服务端中交互

我们新建model文件夹，创建实体类ChatMessage。

```java public class ChatMessage { private MessageType type; private String content; private String sender;

public enum MessageType {
    CHAT,
    JOIN,
    LEAVE
}

public MessageType getType() {
    return type;
}

public void setType(MessageType type) {
    this.type = type;
}

public String getContent() {
    return content;
}

public void setContent(String content) {
    this.content = content;
}

public String getSender() {
    return sender;
}

public void setSender(String sender) {
    this.sender = sender;
}

}

```

实体中，有三个字段：

type:消息类型
content：消息内容
sender：发送

类型有三种：

CHAT: 消息
JOIN：加入
LEAVE：离开

创建Controller来接收和发送消息

创建controller文件夹，在controller文件夹添加类ChatController

```java import com.example.websocketdemo.model.ChatMessage; import org.springframework.messaging.handler.annotation.MessageMapping; import org.springframework.messaging.handler.annotation.Payload; import org.springframework.messaging.handler.annotation.SendTo; import org.springframework.messaging.simp.SimpMessageHeaderAccessor; import org.springframework.stereotype.Controller;

@Controller public class ChatController {

@MessageMapping("/chat.sendMessage")
@SendTo("/topic/public")
public ChatMessage sendMessage(@Payload ChatMessage chatMessage) {
    return chatMessage;
}

@MessageMapping("/chat.addUser")
@SendTo("/topic/public")
public ChatMessage addUser(@Payload ChatMessage chatMessage, 
                           SimpMessageHeaderAccessor headerAccessor) {
    // Add username in web socket session
    headerAccessor.getSessionAttributes().put("username", chatMessage.getSender());
    return chatMessage;
}

}

```

代码解释：

我们在websocket配置中，从目的地以/app开头的客户端发送的所有消息都将路由到这些使用@MessageMapping注释的消息处理方法。

例如，具有目标/app/chat.sendMessage的消息将路由到sendMessage（）方法，并且具有目标/app/chat.addUser的消息将路由到addUser（）方法

添加WebSocket事件监听

完成了上述代码后，我们还需要对socket的连接和断连事件进行监听，这样我们才能广播用户进来和出去等操作。

创建listener文件夹，新建WebSocketEventListener类

```java import com.example.websocketdemo.model.ChatMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.event.EventListener; import org.springframework.messaging.simp.SimpMessageSendingOperations; import org.springframework.messaging.simp.stomp.StompHeaderAccessor; import org.springframework.stereotype.Component; import org.springframework.web.socket.messaging.SessionConnectedEvent; import org.springframework.web.socket.messaging.SessionDisconnectEvent;

@Component public class WebSocketEventListener {

private static final Logger logger = LoggerFactory.getLogger(WebSocketEventListener.class);

@Autowired
private SimpMessageSendingOperations messagingTemplate;

@EventListener
public void handleWebSocketConnectListener(SessionConnectedEvent event) {
    logger.info("Received a new web socket connection");
}

@EventListener
public void handleWebSocketDisconnectListener(SessionDisconnectEvent event) {
    StompHeaderAccessor headerAccessor = StompHeaderAccessor.wrap(event.getMessage());

    String username = (String) headerAccessor.getSessionAttributes().get("username");
    if(username != null) {
        logger.info("User Disconnected : " + username);

        ChatMessage chatMessage = new ChatMessage();
        chatMessage.setType(ChatMessage.MessageType.LEAVE);
        chatMessage.setSender(username);

        messagingTemplate.convertAndSend("/topic/public", chatMessage);
    }
}

}

```

代码解释：

我们已经在ChatController中定义的addUser（）方法中广播了用户加入事件。因此，我们不需要在SessionConnected事件中执行任何操作。

在SessionDisconnect事件中，编写代码用来从websocket会话中提取用户名，并向所有连接的客户端广播用户离开事件。

创建前端聊天室页面

我们在src/main/resources文件下创建前端文件，结构类似这样：

```html static └── css └── main.css └── js └── main.js └── index.html

```

HTML文件 index.html

HTML文件包含用于显示聊天消息的用户界面。它包括sockjs和stomp 两个js库。

SockJS是一个WebSocket客户端，它尝试使用本机WebSockets，并为不支持WebSocket的旧浏览器提供支持。 STOMP JS是javascript的stomp客户端。

笔者在文件里使用了国内的CDN源

```html

Spring Boot WebSocket Chat Application

Type your username

```

2. JavaScript main.js

添加连接到websocket端点以及发送和接收消息所需的javascript。

```js 'use strict';

var usernamePage = document.querySelector('#username-page'); var chatPage = document.querySelector('#chat-page'); var usernameForm = document.querySelector('#usernameForm'); var messageForm = document.querySelector('#messageForm'); var messageInput = document.querySelector('#message'); var messageArea = document.querySelector('#messageArea'); var connectingElement = document.querySelector('.connecting');

var stompClient = null; var username = null;

var colors = [ '#2196F3', '#32c787', '#00BCD4', '#ff5652', '#ffc107', '#ff85af', '#FF9800', '#39bbb0' ];

function connect(event) { username = document.querySelector('#name').value.trim();

if(username) {
    usernamePage.classList.add('hidden');
    chatPage.classList.remove('hidden');

    var socket = new SockJS('/ws');
    stompClient = Stomp.over(socket);

    stompClient.connect({}, onConnected, onError);
}
event.preventDefault();

}

function onConnected() { // Subscribe to the Public Topic stompClient.subscribe('/topic/public', onMessageReceived);

// Tell your username to the server
stompClient.send("/app/chat.addUser",
    {},
    JSON.stringify({sender: username, type: 'JOIN'})
)

connectingElement.classList.add('hidden');

}

function onError(error) { connectingElement.textContent = 'Could not connect to WebSocket server. Please refresh this page to try again!'; connectingElement.style.color = 'red'; }

function onMessageReceived(payload) { var message = JSON.parse(payload.body);

var messageElement = document.createElement('li');

if(message.type === 'JOIN') {
    messageElement.classList.add('event-message');
    message.content = message.sender + ' joined!';
} else if (message.type === 'LEAVE') {
    messageElement.classList.add('event-message');
    message.content = message.sender + ' left!';
} else {
    messageElement.classList.add('chat-message');

    var avatarElement = document.createElement('i');
    var avatarText = document.createTextNode(message.sender[0]);
    avatarElement.appendChild(avatarText);
    avatarElement.style['background-color'] = getAvatarColor(message.sender);

    messageElement.appendChild(avatarElement);

    var usernameElement = document.createElement('span');
    var usernameText = document.createTextNode(message.sender);
    usernameElement.appendChild(usernameText);
    messageElement.appendChild(usernameElement);
}

var textElement = document.createElement('p');
var messageText = document.createTextNode(message.content);
textElement.appendChild(messageText);

messageElement.appendChild(textElement);

messageArea.appendChild(messageElement);
messageArea.scrollTop = messageArea.scrollHeight;

}

function getAvatarColor(messageSender) { var hash = 0; for (var i = 0; i < messageSender.length; i++) { hash = 31 * hash + messageSender.charCodeAt(i); } var index = Math.abs(hash % colors.length); return colors[index]; }

usernameForm.addEventListener('submit', connect, true) messageForm.addEventListener('submit', sendMessage, true)

```

代码解释：

connect（）函数使用SockJS和stomp客户端连接到我们在Spring Boot中配置的/ws端点。

成功连接后，客户端订阅/topic/public，并通过向/app/chat.addUser目的地发送消息将该用户的名称告知服务器。

stompClient.subscribe()函数采用一种回调方法，只要消息到达订阅主题，就会调用该方法。

其它的代码用于在屏幕上显示和格式化消息。

CSS main.css

```css * { -webkit-box-sizing: border-box; -moz-box-sizing: border-box; box-sizing: border-box; }

html,body { height: 100%; overflow: hidden; }

body { margin: 0; padding: 0; font-weight: 400; font-family: "Helvetica Neue", Helvetica, Arial, sans-serif; font-size: 1rem; line-height: 1.58; color: #333; background-color: #f4f4f4; height: 100%; }

body:before { height: 50%; width: 100%; position: absolute; top: 0; left: 0; background: #128ff2; content: ""; z-index: 0; }

.clearfix:after { display: block; content: ""; clear: both; }

.hidden { display: none; }

.form-control { width: 100%; min-height: 38px; font-size: 15px; border: 1px solid #c8c8c8; }

.form-group { margin-bottom: 15px; }

input { padding-left: 10px; outline: none; }

h1, h2, h3, h4, h5, h6 { margin-top: 20px; margin-bottom: 20px; }

h1 { font-size: 1.7em; }

a { color: #128ff2; }

button { box-shadow: none; border: 1px solid transparent; font-size: 14px; outline: none; line-height: 100%; white-space: nowrap; vertical-align: middle; padding: 0.6rem 1rem; border-radius: 2px; transition: all 0.2s ease-in-out; cursor: pointer; min-height: 38px; }

button.default { background-color: #e8e8e8; color: #333; box-shadow: 0 2px 2px 0 rgba(0, 0, 0, 0.12); }

button.primary { background-color: #128ff2; box-shadow: 0 2px 2px 0 rgba(0, 0, 0, 0.12); color: #fff; }

button.accent { background-color: #ff4743; box-shadow: 0 2px 2px 0 rgba(0, 0, 0, 0.12); color: #fff; }

username-page {

text-align: center;

}

.username-page-container { background: #fff; box-shadow: 0 1px 11px rgba(0, 0, 0, 0.27); border-radius: 2px; width: 100%; max-width: 500px; display: inline-block; margin-top: 42px; vertical-align: middle; position: relative; padding: 35px 55px 35px; min-height: 250px; position: absolute; top: 50%; left: 0; right: 0; margin: 0 auto; margin-top: -160px; }

.username-page-container .username-submit { margin-top: 10px; }

chat-page {

position: relative;
height: 100%;

}

.chat-container { max-width: 700px; margin-left: auto; margin-right: auto; background-color: #fff; box-shadow: 0 1px 11px rgba(0, 0, 0, 0.27); margin-top: 30px; height: calc(100% - 60px); max-height: 600px; position: relative; }

chat-page ul {

list-style-type: none;
background-color: #FFF;
margin: 0;
overflow: auto;
overflow-y: scroll;
padding: 0 20px 0px 20px;
height: calc(100% - 150px);

}

chat-page #messageForm {

padding: 20px;

}

chat-page ul li {

line-height: 1.5rem;
padding: 10px 20px;
margin: 0;
border-bottom: 1px solid #f4f4f4;

}

chat-page ul li p {

margin: 0;

}

chat-page .event-message {

width: 100%;
text-align: center;
clear: both;

}

chat-page .event-message p {

color: #777;
font-size: 14px;
word-wrap: break-word;

}

chat-page .chat-message {

padding-left: 68px;
position: relative;

}

chat-page .chat-message i {

position: absolute;
width: 42px;
height: 42px;
overflow: hidden;
left: 10px;
display: inline-block;
vertical-align: middle;
font-size: 18px;
line-height: 42px;
color: #fff;
text-align: center;
border-radius: 50%;
font-style: normal;
text-transform: uppercase;

}

chat-page .chat-message span {

color: #333;
font-weight: 600;

}

chat-page .chat-message p {

color: #43464b;

}

messageForm .input-group input {

float: left;
width: calc(100% - 85px);

}

messageForm .input-group button {

float: left;
width: 80px;
height: 38px;
margin-left: 5px;

}

.chat-header { text-align: center; padding: 15px; border-bottom: 1px solid #ececec; }

.chat-header h2 { margin: 0; font-weight: 500; }

.connecting { padding-top: 5px; text-align: center; color: #777; position: absolute; top: 65px; width: 100%; }

@media screen and (max-width: 730px) {

.chat-container {
    margin-left: 10px;
    margin-right: 10px;
    margin-top: 10px;
}

}

@media screen and (max-width: 480px) { .chat-container { height: calc(100% - 30px); }

.username-page-container {
    width: auto;
    margin-left: 15px;
    margin-right: 15px;
    padding: 25px;
}

#chat-page ul {
    height: calc(100% - 120px);
}

#messageForm .input-group button {
    width: 65px;
}

#messageForm .input-group input {
    width: calc(100% - 70px);
}

.chat-header {
    padding: 10px;
}

.connecting {
    top: 60px;
}

.chat-header h2 {
    font-size: 1.1em;
}

}

```

整个项目结构如下：

启动

启动SpringBoot项目

效果入下：

补充：使用RabbitMQ代替内存作为消息代理

添加依赖：

```xml

org.springframework.boot spring-boot-starter-amqp

org.springframework.boot spring-boot-starter-reactor-netty

```

然后将WebSocketConfig类中configureMessageBroker方法改为使用RabbitMq，完成！

```java public void configureMessageBroker(MessageBrokerRegistry registry) { registry.setApplicationDestinationPrefixes("/app");

// Use this for enabling a Full featured broker like RabbitMQ
registry.enableStompBrokerRelay("/topic")
        .setRelayHost("localhost")
        .setRelayPort(61613)
        .setClientLogin("guest")
        .setClientPasscode("guest");

}

```

如此一来，便可以通过RabbitMq进行消息的订阅。

4.改造为分布式集群

分布式就是为了解决单点故障问题，想象一下，如果一个服务器承载了1000个大佬同时聊天，服务器突然挂了，1000个大佬瞬间全部掉线，大概明天你就被大佬们吊起来打了。

当聊天室改为集群后，就算服务器A挂了，服务器B上聊天的大佬们还可以愉快的聊天，并且在前端还能通过代码，让连接A的大佬们快速重连至存活的服务器B，继续和大家愉快的聊天，岂不美哉！

总结一下：实现了分布式WebSocket后，我们可以将流量负载均衡到不同的服务器上并提供一种通信机制让各个服务器能进行消息同步（不然用户A连上服务器A，用户B脸上服务器B，它们发消息的时候对方都没法收到）。

如何改造为分布式集群当我们要实现分布式的时候，我们则需要在各个机器上共享这些信息，所以我们需要一个Publish/Subscribe的中间件。我们现在使用Redis作为我们的解决方案。

用户在聊天室集群如何发消息

假设我们的聊天室集群有服务器A和B，用户Alice连接在A上，Bob连接在B上、

Alice向聊天室的服务器A发送消息，A服务器必须要将收到的消息转发到Redis，才能保证聊天室集群的所有服务器（也就是A和B）能够拿到消息。否则，只有Alice在的服务器A能够读到消息，用户Bob在的服务器B并不能收到消息，A和B也就无法聊天了。

用户在聊天室集群如何接收消息说完了发送消息，那么如何保证Alice发的消息，其他所有人都能收到呢，前面我们知道了Alice发送的消息已经被传到了Redis的频道，那么所有服务器都必须订阅这个Redis频道，然后把这个频道的消息转发到自己的用户那里，这样自己服务器所管辖的用户就能收到消息。

补充知识点：STOMP 简介

上期我们搭建了个websocket聊天室demo，并且使用了STOMP协议，但是我并没有介绍到底什么是STOMP协议，同学们会有疑惑，这里对于STOMP有很好地总结：

当直接使用WebSocket时（或SockJS）就很类似于使用TCP套接字来编写Web应用。因为没有高层级的线路协议（wire protocol），因此就需要我们定义应用之间所发送消息的语义，还需要确保连接的两端都能遵循这些语义。

就像HTTP在TCP套接字之上添加了请求-响应模型层一样，STOMP在WebSocket之上提供了一个基于帧的线路格式（frame-based wire format）层，用来定义消息的语义。

与HTTP请求和响应类似，STOMP帧由命令、一个或多个头信息以及负载所组成。例如，如下就是发送数据的一个STOMP帧：

```sh

SEND transaction:tx-0 destination:/app/marco content-length:20

{"message":"Marco!"}

```

好了，介绍完了概念，让我们开始动手改造！

功能一：向聊天室集群中的全体用户发消息——Redis的订阅/发布如果你不熟悉Redis的sub/pub（订阅/发布）功能，请看这里进行简单了解它的用法，很简单：

https://redisbook.readthedocs.io/en/latest/feature/pubsub.html

在我们上篇文章的Demo基础上，我们进行集群改造。上一篇文章的源码见下方：

https://github.com/qqxx6661/springboot-websocket-demo/releases/tag/%E5%8D%95%E6%9C%BA%E7%89%88

添加Redis依赖pom

```xml

org.springframework.boot spring-boot-starter-data-redis

```

application.properties新增redis配置

当然首先要确保你安装了Redis，windows下安装redis比较麻烦，你可以搜索redis-for-windows下载安装。

```properties

redis 连接配置

spring.redis.database=0 spring.redis.host=127.0.0.1 spring.redis.password= spring.redis.port=6379 spring.redis.ssl=false

空闲连接最大数

spring.redis.jedis.pool.max-idle=10

获取连接最大等待时间(s)

spring.redis.jedis.pool.max-wait=60000

```

在application.properties添加频道名定义

```properties

Redis定义

redis.channel.msgToAll = websocket.msgToAll ```

新建redis/RedisListenerBean

```java package cn.monitor4all.springbootwebsocketdemo.redis;

import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.listener.PatternTopic; import org.springframework.data.redis.listener.RedisMessageListenerContainer; import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; import org.springframework.stereotype.Component;

import java.net.Inet4Address; import java.net.InetAddress;

/* * Redis订阅频道属性类 * @author yangzhendong01 / @Component public class RedisListenerBean {

private static final Logger LOGGER = LoggerFactory.getLogger(RedisListenerBean.class);

@Value("${server.port}")
private String serverPort;

@Value("${redis.channel.msgToAll}")
private String msgToAll;

/**
 * redis消息监听器容器
 * 可以添加多个监听不同话题的redis监听器，只需要把消息监听器和相应的消息订阅处理器绑定，该消息监听器
 * 通过反射技术调用消息订阅处理器的相关方法进行一些业务处理
 * @param connectionFactory
 * @param listenerAdapter
 * @return
 */
@Bean
RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {
    RedisMessageListenerContainer container = new RedisMessageListenerContainer();
    container.setConnectionFactory(connectionFactory);

    // 监听msgToAll
    container.addMessageListener(listenerAdapter, new PatternTopic(msgToAll));
    LOGGER.info("Subscribed Redis channel: " + msgToAll);
    return container;
}

}

```

聊天室集群：发消息改造

我们单机聊天室的发送消息Controller是这样的：

```java @MessageMapping("/chat.sendMessage") @SendTo("/topic/public") public ChatMessage sendMessage(@Payload ChatMessage chatMessage) { return chatMessage;

```

我们前端发给我们消息后，直接给/topic/public转发这个消息，让其他用户收到。

在集群中，我们需要把消息转发给Redis，并且不转发给前端，而是让服务端监听Redis消息，在进行消息发送。

将Controller改为：

```java @Value("${redis.channel.msgToAll}") private String msgToAll;

@Autowired private RedisTemplate redisTemplate;

@MessageMapping("/chat.sendMessage") public void sendMessage(@Payload ChatMessage chatMessage) { try { redisTemplate.convertAndSend(msgToAll, JsonUtil.parseObjToJson(chatMessage)); } catch (Exception e) { LOGGER.error(e.getMessage(), e); } }

```

你会发现我们在代码中使用了JsonUtil将实体类ChatMessage转为了Json发送给了Redis，这个Json工具类需要使用到FaskJson依赖：

pom添加FastJson依赖

```xml

com.alibaba fastjson 1.2.58

```

添加Json解析工具类JsonUtil，提供对象转Json，Json转对象的能力

```java package cn.monitor4all.springbootwebsocketdemo.util;

import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject; import org.slf4j.Logger; import org.slf4j.LoggerFactory;

/* * JSON 转换 / public final class JsonUtil {

private static final Logger LOGGER = LoggerFactory.getLogger(JsonUtil.class);

/**
 * 把Java对象转换成json字符串
 *
 * @param object 待转化为JSON字符串的Java对象
 * @return json 串 or null
 */
public static String parseObjToJson(Object object) {
    String string = null;
    try {
        string = JSONObject.toJSONString(object);
    } catch (Exception e) {
        LOGGER.error(e.getMessage());
    }
    return string;
}

/**
 * 将Json字符串信息转换成对应的Java对象
 *
 * @param json json字符串对象
 * @param c    对应的类型
 */
public static <T> T parseJsonToObj(String json, Class<T> c) {
    try {
        JSONObject jsonObject = JSON.parseObject(json);
        return JSON.toJavaObject(jsonObject, c);
    } catch (Exception e) {
        LOGGER.error(e.getMessage());
    }
    return null;
}

}

```

这样，我们接收到用户发送消息的请求时，就将消息转发给了redis的频道websocket.msgToAll

聊天室集群：接收消息改造

单机的聊天室，我们接收消息是通过Controller直接把消息转发到所有人的频道上，这样就能在所有人的聊天框显示。

在集群中，我们需要服务器把消息从Redis中拿出来，并且推送到自己管的用户那边，我们在Service层实现消息的推送。

在处理消息之后发送消息：正如前面看到的那样，使用 @MessageMapping 或者 @SubscribeMapping 注解可以处理客户端发送过来的消息，并选择方法是否有返回值。如果 @MessageMapping注解的控制器方法有返回值的话，返回值会被发送到消息代理，只不过会添加上"/topic"前缀。可以使用@SendTo 重写消息目的地；如果 @SubscribeMapping注解的控制器方法有返回值的话，返回值会直接发送到客户端，不经过代理。如果加上@SendTo 注解的话，则要经过消息代理。
在应用的任意地方发送消息： spring-websocket 定义了一个 SimpMessageSendingOperations 接口（或者使用SimpMessagingTemplate ），可以实现自由的向任意目的地发送消息，并且订阅此目的地的所有用户都能收到消息。我们在service实现发送，需要使用上述第二种方法。

新建类service/ChatService：

```java package cn.monitor4all.springbootwebsocketdemo.service;

import cn.monitor4all.springbootwebsocketdemo.model.ChatMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.messaging.handler.annotation.Payload; import org.springframework.messaging.simp.SimpMessageSendingOperations; import org.springframework.stereotype.Service;

@Service public class ChatService {

private static final Logger LOGGER = LoggerFactory.getLogger(ChatService.class);

@Autowired
private SimpMessageSendingOperations simpMessageSendingOperations;

public void sendMsg(@Payload ChatMessage chatMessage) {
    LOGGER.info("Send msg by simpMessageSendingOperations:" + chatMessage.toString());
    simpMessageSendingOperations.convertAndSend("/topic/public", chatMessage);
}

}

```

我们在哪里调用这个service呢，我们需要在监听到消息后调用，所以我们就要有下面的Redis监听消息处理专用类

新建类redis/RedisListenerHandle：

```java package cn.monitor4all.springbootwebsocketdemo.redis;

import cn.monitor4all.springbootwebsocketdemo.model.ChatMessage; import cn.monitor4all.springbootwebsocketdemo.service.ChatService; import cn.monitor4all.springbootwebsocketdemo.util.JsonUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.data.redis.connection.Message; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; import org.springframework.stereotype.Component;

/* * Redis订阅频道处理类 * @author yangzhendong01 / @Component public class RedisListenerHandle extends MessageListenerAdapter {

private static final Logger LOGGER = LoggerFactory.getLogger(RedisListenerHandle.class);

@Value("${redis.channel.msgToAll}")
private String msgToAll;

@Value("${server.port}")
private String serverPort;

@Autowired
private RedisTemplate<String, String> redisTemplate;

@Autowired
private ChatService chatService;

/**
 * 收到监听消息
 * @param message
 * @param bytes
 */
@Override
public void onMessage(Message message, byte[] bytes) {
    byte[] body = message.getBody();
    byte[] channel = message.getChannel();
    String rawMsg;
    String topic;
    try {
        rawMsg = redisTemplate.getStringSerializer().deserialize(body);
        topic = redisTemplate.getStringSerializer().deserialize(channel);
        LOGGER.info("Received raw message from topic:" + topic + ", raw message content：" + rawMsg);
    } catch (Exception e) {
        LOGGER.error(e.getMessage(), e);
        return;
    }


    if (msgToAll.equals(topic)) {
        LOGGER.info("Send message to all users:" + rawMsg);
        ChatMessage chatMessage = JsonUtil.parseJsonToObj(rawMsg, ChatMessage.class);
        // 发送消息给所有在线Cid
        chatService.sendMsg(chatMessage);
    } else {
        LOGGER.warn("No further operation with this topic!");
    }
}

}

```

看看效果

这样，我们的改造就基本完成了！我们看一下效果

我们将服务器运行在8080上，然后打开localhost:8080，起名Alice进入聊天室

随后，我们在application.properties中将端口server.port=8081

再次运行程序（别忘了开启IDEA的“允许启动多个并行服务”设置，不然会覆盖掉你的8080服务，如下图），在8081启动一个聊天室，起名Bob进入聊天室。

如下两图，我们已经可以在不同端口的两个聊天室，互相聊天了！（注意看url）

在互相发送消息是，我们还可以使用命令行监听下Redis的频道websocket.msgToAll，可以看到双方传送的消息。如下图：

我们还可以打开Chrome的F12控制台，查看前端的控制台发送消息的log，如下图：

大功告成了吗？

功能实现了，但是并不完美！你会发现，Bob的加入并没有提醒Bob进入了聊天室（在单机版是有的），这是因为我们在“加入聊天室”的代码还没有修改，在加入时，只有Bob的服务器B里的其他用户知道Bob加入了聊天室。我们还能再进一步！

功能二/功能三：集群用户上下线通知，集群用户信息存储我们需要弥补上面的不足，将用户上线下线的广播发送到所有服务器上。

此外，我还希望以后能够查询集群中所有的在线用户，我们在redis中添加一个set，来保存用户名，这样就可以随时得到在线用户的数量和名称。

在application.properties添加频道名定义

```properties

Redis定义

redis.channel.userStatus = websocket.userStatus redis.set.onlineUsers = websocket.onlineUsers

```

我们增加两个定义

第一个是新增redis频道websocket.userStatus用来广播用户上下线消息

第二个是redis的set，用来保存在线用户信息

在RedisListenerBean添加新频道监听

container.addMessageListener(listenerAdapter, new PatternTopic(userStatus));

在ChatService中添加

```java public void alertUserStatus(@Payload ChatMessage chatMessage) { LOGGER.info("Alert user online by simpMessageSendingOperations:" + chatMessage.toString()); simpMessageSendingOperations.convertAndSend("/topic/public", chatMessage); }

```

在service中我们向本服务器的用户广播消息，用户上线或者下线的消息都通过这里传达。

修改ChatController中的addUser方法

```java @MessageMapping("/chat.addUser") public void addUser(@Payload ChatMessage chatMessage, SimpMessageHeaderAccessor headerAccessor) {

    LOGGER.info("User added in Chatroom:" + chatMessage.getSender());
    try {
        headerAccessor.getSessionAttributes().put("username", chatMessage.getSender());
        redisTemplate.opsForSet().add(onlineUsers, chatMessage.getSender());
        redisTemplate.convertAndSend(userStatus, JsonUtil.parseObjToJson(chatMessage));
    } catch (Exception e) {
        LOGGER.error(e.getMessage(), e);
    }
}

```

我们修改了addUser方法，在这里往redis中广播用户上线的消息，并把用户名username写入redis的set中（websocket.onlineUsers）

修改WebSocketEventListener中的handleWebSocketDisconnectListener方法

```java @EventListener public void handleWebSocketDisconnectListener(SessionDisconnectEvent event) {

    StompHeaderAccessor headerAccessor = StompHeaderAccessor.wrap(event.getMessage());

    String username = (String) headerAccessor.getSessionAttributes().get("username");

    if(username != null) {
        LOGGER.info("User Disconnected : " + username);
        ChatMessage chatMessage = new ChatMessage();
        chatMessage.setType(ChatMessage.MessageType.LEAVE);
        chatMessage.setSender(username);
        try {
            redisTemplate.opsForSet().remove(onlineUsers, username);
            redisTemplate.convertAndSend(userStatus, JsonUtil.parseObjToJson(chatMessage));
        } catch (Exception e) {
            LOGGER.error(e.getMessage(), e);
        }

    }
}

```

在用户关闭网页时，websocket会调用该方法，我们在这里需要把用户从redis的在线用户set里删除，并且向集群发送广播，说明该用户退出聊天室。

修改Redis监听类RedisListenerHandle

```java else if (userStatus.equals(topic)) { ChatMessage chatMessage = JsonUtil.parseJsonToObj(rawMsg, ChatMessage.class); if (chatMessage != null) { chatService.alertUserStatus(chatMessage); }

```

在监听类中我们接受了来自userStatus频道的消息，并调用service