假设有这样一个需求，前端需要实时提醒数据库进入的数据，或者监听数据库里面的数据里面有什么变动，需要后端及时的提醒前端消息，这里可以使用websocket实现后端主动推送数据给前端，可以用redis保存每个用户的session，然后分别给不同用户发送不同的消息，这个例子我用的是redis中的发布订阅的功能（subscribe,publish命令）redis中可以用subscribe …(例如 subscribe channer) 订阅了channer这个频道，换句话说就是创建了这个频道，哪个用户订阅了channer这个频道，就可以接收到里面的数据，可以用publish…（例如：sublish channer “往channer频道存入数据”）往频道中发送数据这里是看别人的代码拉取下来学习看的，这里做一个笔记下面先看代码先把项目结构贴出来
pom.xml文件的依赖
org.springframework.boot spring-boot-starter-web org.springframework.boot spring-boot-starter-test test org.springframework.boot spring-boot-starter-data-redis org.springframework.boot spring-boot-starter-websocket 1.3.5.RELEASE
项目启动的时候就加载，将订阅的频道给创建出来，就是相当于我已经订阅 了这三个频道了（订阅哪几个频道可以自己进行设置）
package com.demo.redisandwebsocket.config;
import com.demo.redisandwebsocket.listener.SubscribeListener;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
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.Topic;
import java.util.ArrayList;
import java.util.List;
@Configuration //相当于xml中的beans
public class RedisConfig {
@Bean
public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
//订阅频道 的通道
// 原来的写法 订阅了ZHOU这个频道
// container.addMessageListener(new MessageListener(){
// @Override
// public void onMessage(Message message, byte[] pattern) {
// String msg = new String(message.getBody());
// System.out.println(new String(pattern) + "主题发布：" + msg);
// }
// }, new PatternTopic("ZHOU"));
// 改成lambda表达式的写法，更加的简洁 订阅了ZHOU这个频道
// container.addMessageListener((a,b) -> {
// String msg = new String(a.getBody());
// System.out.println(new String(b) + "主题发布：" + msg);
// },new PatternTopic("ZHOU"));
List list = new ArrayList<>();
list.add(new PatternTopic("ZHOU"));
list.add(new PatternTopic("DA"));
list.add(new PatternTopic("TOU"));
container.addMessageListener(new SubscribeListener(),list);
return container;
}
}
然后自定义SubscribeListener类，实现MessageListener,这个类是监听redis中的频道是否有数据进来了，如果有数据进来这个监听器就会监听到，并且触发里面onMessage方法
package com.demo.redisandwebsocket.listener;
import com.demo.redisandwebsocket.web.WebSocketServer;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import javax.annotation.Resource;
import javax.websocket.Session;
import java.io.IOException;
public class SubscribeListener implements MessageListener {
@Resource
private WebSocketServer webSocketServer;
private Session session;
public Session getSession() { return session; }
public void setSession(Session session) {
this.session = session;
}
@Override
public void onMessage(Message message, byte[] pattern) {
String msg = new String(message.getBody());
System.out.println(new String(pattern) + "主题发布：" + msg);
if (null != session && session.isOpen()) {
try {
session.getBasicRemote().sendText(msg);
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
然后是WebSocketConfig类，这个类是初始化websocket的配置，也是初始化的时候将他对象注入到sping容器当中，因为websocket连接是基于http协议连接的，连接上以后就跟http协议没有关系了，然后就基于websocket进行连接了
package com.demo.redisandwebsocket.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 {
@Bean
public ServerEndpointExporter serverEndpointExporter() {
return new ServerEndpointExporter();
}
}
springUtils工具类，目前没有能力解释，希望有人帮我解释一下
package com.demo.redisandwebsocket.util;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.stereotype.Component;
@Component
public final class SpringUtils implements BeanFactoryPostProcessor {
private static ConfigurableListableBeanFactory beanFactory; // Spring应用上下文环境
@Override
public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
SpringUtils.beanFactory = beanFactory;
}
public static ConfigurableListableBeanFactory getBeanFactory() {
return beanFactory;
}
@SuppressWarnings("unchecked")
public static T getBean(String name) throws BeansException {
return (T) getBeanFactory().getBean(name);
}
public static T getBean(Class clz) throws BeansException {
T result = (T) getBeanFactory().getBean(clz);
return result;
}
public static boolean containsBean(String name) {
return getBeanFactory().containsBean(name);
}
public static boolean isSingleton(String name) throws NoSuchBeanDefinitionException {
return getBeanFactory().isSingleton(name);
}
public static Class getType(String name) throws NoSuchBeanDefinitionException {
return getBeanFactory().getType(name);
}
public static String[] getAliases(String name) throws NoSuchBeanDefinitionException {
return getBeanFactory().getAliases(name);
}
}
这里就是将消息发送到前端的websocket的类了，里面有四个方法（四个注解）
package com.demo.redisandwebsocket.web;
import com.demo.redisandwebsocket.listener.SubscribeListener;
import com.demo.redisandwebsocket.util.SpringUtils;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.Topic;
import org.springframework.stereotype.Component;
import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicInteger;
@Component
@ServerEndpoint("/websocket/server")
public class WebSocketServer {
private RedisMessageListenerContainer redisMessageListenerContainer = SpringUtils.getBean(RedisMessageListenerContainer.class);
//静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
private static AtomicInteger onlineCount=new AtomicInteger(0);
//concurrent包的线程安全Set，用来存放每个客户端对应的webSocket对象。若要实现服务端与单一客户端通信的话，可以使用Map来存放，其中Key可以为用户标识
private static CopyOnWriteArraySet webSocketSet = new CopyOnWriteArraySet();
//与某个客户端的连接会话，需要通过它来给客户端发送数据
private Session session;
private SubscribeListener subscribeListener;
@OnOpen
public void onOpen(Session session){
this.session = session;
webSocketSet.add(this); //加入set中
addOnlineCount(); //在线数加1
System.out.println("有新连接加入！当前在线人数为" + getOnlineCount());
subscribeListener = new SubscribeListener();
subscribeListener.setSession(session);
List list = new ArrayList<>();
list.add(new PatternTopic("ZHOU"));
list.add(new PatternTopic("DA"));
list.add(new PatternTopic("TOU"));
//设置订阅topic
redisMessageListenerContainer.addMessageListener(subscribeListener, list);
}
@OnClose
public void onClose() throws IOException {
webSocketSet.remove(this); //从set中删除
subOnlineCount(); //在线数减1
redisMessageListenerContainer.removeMessageListener(subscribeListener);
System.out.println("有一连接关闭！当前在线人数为" + getOnlineCount());
}
@OnMessage
public void onMessage(String message, Session session) {
System.out.println("来自客户端的消息:" + message);
//群发消息
for(WebSocketServer item: webSocketSet){
try {
item.sendMessage(message);
} catch (IOException e) {
e.printStackTrace();
continue;
}
}
}
@OnError
public void onError(Session session, Throwable error){
System.out.println("发生错误");
error.printStackTrace();
}
public void sendMessage(String message) throws IOException {
this.session.getBasicRemote().sendText(message);
}
public int getOnlineCount() {
return onlineCount.get();
}
public void addOnlineCount() {
WebSocketServer.onlineCount.getAndIncrement();
}
public void subOnlineCount() {
WebSocketServer.onlineCount.getAndDecrement();
}
}
把前端也贴出来
在这里插入代码片
websocket
使用redis订阅消息和websocket实现消息推送
收到的订阅消息：