首先要配置Redis，我这里就以最简单的配置配了一下
pom.xml
<!--redis-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<!-- websocket -->
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
配置文件application.yml
server:
port: 10001
#spring:
# main:
# log-startup-info: false
spring:
redis:
port: 6379
host: localhost
password:
database: 0
配置RedisTemplate
package com.studytest.study.common.config;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.Topic;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import java.util.*;
@Configuration
public class RedisConfig {
@Bean
public RedisTemplate<String,Object> redisTemplate(RedisConnectionFactory redisConnectionFactory){
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<>(Object.class);
ObjectMapper om = new ObjectMapper();
om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
jackson2JsonRedisSerializer.setObjectMapper(om);
redisTemplate.setConnectionFactory(redisConnectionFactory);
redisTemplate.setKeySerializer(new StringRedisSerializer());
redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
redisTemplate.setHashKeySerializer(new StringRedisSerializer());
redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);
return redisTemplate;
}
/**
* 监听订阅的消息
* @param redisConnectionFactory
* @param myMessageListener
* @return
*/
@Bean
public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory redisConnectionFactory,MyMessageListener myMessageListener){
RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();
redisMessageListenerContainer.setConnectionFactory(redisConnectionFactory);
//设置消息监听器 和 Topic名称，默认就设置了一个MYCHANEL
Map<MessageListener, Collection<? extends Topic>> map = new HashMap<>();
List<ChannelTopic> channelTopicList = new ArrayList<>();
ChannelTopic channelTopic = new ChannelTopic("MYCHANEL");
channelTopicList.add(channelTopic);
map.put(myMessageListener,channelTopicList);
redisMessageListenerContainer.setMessageListeners(map);
return redisMessageListenerContainer;
}
}
自定义消息监听器
package com.studytest.study.common.config;
import com.studytest.study.controller.WebSocketService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.stereotype.Component;
@Component
public class MyMessageListener implements MessageListener{
@Autowired
private RedisTemplate<String, Object> redisTemplate;
@Autowired
private WebSocketService webSocketService;
@Override
public void onMessage(Message message, byte[] bytes) {
Object deserialize = redisTemplate.getValueSerializer().deserialize(message.getBody());
String channel =new StringRedisSerializer().deserialize(message.getChannel());
System.out.println(channel);
System.out.println(deserialize);
webSocketService.sendMessage(deserialize.toString(),channel);
};
}
websocket 配置
package com.studytest.study.common.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
@Configuration
public class WebSocketConfig {
/**
* 注入一个ServerEndpointExporter,该Bean会自动注册使用@ServerEndpoint注解申明的websocket endpoint
*/
@Bean
public ServerEndpointExporter serverEndpointExporter() {
return new ServerEndpointExporter();
}
}
发布消息工具类
package com.studytest.study.common.config;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Component;
@Component
public class PublishUtils {
@Autowired
private RedisTemplate<String,Object> redisTemplate;
public void publish(String channel,String message){
RedisSerializer<Object> valueSerializer = (RedisSerializer<Object>) redisTemplate.getValueSerializer();
byte[] smessage = valueSerializer.serialize(message);
redisTemplate.executePipelined(new RedisCallback<Object>() {
@Override
public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
redisConnection.publish(channel.getBytes(),smessage);
return null;
}
});
}
}
由于@ServerEndpoint 修饰websocket 的类中不能正常用@Autowired引入bean，所以我们用实现ApplicationContextAware接口的方式来创建一个工具类，以便于在websocket类中进行调用
package com.studytest.study.common.config;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
@Component
public class SpringContextUtil<T> implements ApplicationContextAware {
private static ApplicationContext ac;
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
ac = applicationContext;
}
public static <T>T getBean(Class<?> clazz){
return (T) ac.getBean(clazz);
}
}
websocket服务
package com.studytest.study.controller;
import com.studytest.study.common.config.PublishUtils;
import com.studytest.study.common.config.SpringContextUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@ServerEndpoint("/websocket/{subscribeKey}")
@Component
public class WebSocketService {
Logger log = LoggerFactory.getLogger(WebSocketService.class);
private int count = 1;
private static Map<String, List<WebSocketService>> webSocketMap = new ConcurrentHashMap<String, List<WebSocketService>>();
private Session session;
private String subscribeKey;
/**
* 连接建立成功调用的方法
*/
@OnOpen
public void onOpen(@PathParam("subscribeKey") String subscribeKey, Session session) {
this.session = session;
this.subscribeKey = subscribeKey;
addOnlineNum(); // 在线数加1
addWebSocketMap(subscribeKey,this);
log.info("当前在线人数为：{}", count);
}
/**
* 连接关闭调用的方法
*/
@OnClose
public void onClose(@PathParam("subscribeKey") String subscribeKey,Session session) {
leaveOnlineNum(); // 在线数减1
removeWebSocketMap(subscribeKey,this);
log.info("有一连接关闭：{}，当前在线人数为：{}", session.getId(), count);
}
/**
* 收到客户端消息后调用的方法
*
* @param message
* 客户端发送过来的消息
*/
@OnMessage
public void onMessage(@PathParam("subscribeKey") String subscribeKey,String message) {
log.info("服务端收到客户端的消息:{}", message);
String[] split = message.split("\\|");
if (split.length >= 2){
String msg = split[0] + "说：" + split[1];
System.out.println(msg);
PublishUtils publishUtils = SpringContextUtil.getBean(PublishUtils.class);
publishUtils.publish(subscribeKey,msg);
}
}
@OnError
public void onError(Throwable error) {
log.error("发生错误");
error.printStackTrace();
}
/**
* 服务端发送消息给客户端
*/
public void sendMessage(String message,String subscribeKey) {
try {
log.info("服务端给客户端,发送消息{}", message);
List<WebSocketService> webSocketServices = webSocketMap.get(subscribeKey);
if (!CollectionUtils.isEmpty(webSocketServices)){
for (WebSocketService webSocketService : webSocketServices) {
webSocketService.session.getBasicRemote().sendText(message);
}
}
} catch (Exception e) {
log.error("服务端发送消息给客户端失败：{}", e.getMessage());
}
}
public synchronized void addWebSocketMap(String subscribeKey,WebSocketService webSocketService){
if (webSocketMap.get(subscribeKey) != null){
List<WebSocketService> webSocketServices = webSocketMap.get(subscribeKey);
webSocketServices.add(webSocketService);
}else {
List<WebSocketService> webSocketServices = new ArrayList<>();
webSocketServices.add(webSocketService);
webSocketMap.put(subscribeKey,webSocketServices);
}
}
public synchronized void removeWebSocketMap(String subscribeKey,WebSocketService webSocketService){
if (webSocketMap.get(subscribeKey) != null){
List<WebSocketService> webSocketServices = webSocketMap.get(subscribeKey);
webSocketServices.remove(webSocketService);
}
}
public synchronized void addOnlineNum(){
count++;
}
public synchronized void leaveOnlineNum(){
count--;
}
}
Controller
@Controller
public class LoginController {
@GetMapping("/index")
public String index() {
return "index.html";
}
}
前端HTML index.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
<!-- <script src="https://heerey525.github.io/layui-v2.4.3/layui/layui.js"></script>-->
<script src="https://cdn.bootcss.com/jquery/3.4.1/jquery.js"></script>
</head>
<body>
<!--<button id="user1" onclick="connect()">连接</button>-->
<input id="link" type="text"/>
<input id="sendMsg" type="text"/>
<button onclick="send()">发送</button>
<div id="message">
</div>
</body>
<script type="text/javascript">
var username = "aaa";
var websocket = null;
// function connect() {
//判断当前浏览器是否支持WebSocket ,主要此处要更换为自己的地址
if('WebSocket' in window){
//后边是订阅的topic名称
websocket = new WebSocket("ws://localhost:10001/websocket/MYCHANEL");
}
else{
alert('Not support websocket')
}
//连接发生错误的回调方法
websocket.onerror = function(){
// setMessageInnerHTML("error");
};
//连接成功建立的回调方法
websocket.onopen = function(event){
console.log("连接成功！！！")
// setMessageInnerHTML("open");
$("#link").val("连接成功！！")
}
//连接关闭的回调方法
websocket.onclose = function(){
// setMessageInnerHTML("close");
}
//监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。
// window.onbeforeunload = function(){
// websocket.close();
// }
//发送消息
function send(){
//用户名 + 消息
var msg = username + "|" + $("#sendMsg").val();
websocket.send(msg);
// onmessage();
}
//接收到消息的回调方法
// function onmessage(){
websocket.onmessage = function(event){
console.log(event.data)
// setMessageInnerHTML(event.data);
$("#message").append("<h1>"+ event.data + "</h1>")
// }
}
</script>
</html>
测试 ，我们同时打开两个窗口，一个窗口进行消息发送，看另一个窗口是否也能接收到
image.png
进行消息发送
image.png
我们看另一个窗口可以正常接收到同一个topic订阅的消息
image.png
完成！！！！！
我们还可以利用Redis 来实现1对1的聊天，可以采用发布订阅或者Redis的List （[Lpush/Rpush,LPOP/RPOP] or [BLPOP/BRPOP]）的一些特性的实现。
同时在做消息的持久化的时候，可以利用Redis的Zset的特性来对历史消息进行存储。
