基于Redis发布订阅和websocket实现聊天室功能

首先要配置Redis，我这里就以最简单的配置配了一下pom.xml<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></depe

m0_67505824

1400人浏览 · 2022-03-15 10:54:17

m0_67505824 · 2022-03-15 10:54:17 发布

首先要配置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>