Spring boot + Web Socket + Redis

使用到的技术：

① redis消息订阅；
② spring boot 集成web socket;

需求：

想要实现用户提交审核，后端能够及时收到消息来进行处理。

整体思路：

1.应用程序端：用来处理app数据的服务；
2.后端：用来处理审核的服务；
3.客户端：浏览器。

使用redis 消息订阅功能，应用程序端发出消息，后端接收，保存在redis HashMap中，并通过web socket发送消息到客户端中。客户端获取消息进行增删处理。

应用程序端代码：

@Service
public class BacklogServiceImpl implements BacklogService {

    @Autowired
    RedisAssistant redisAssistant;

    @Override
    public void addBacklog(Backlog backlog) {
        redisAssistant.sendMsg(BacklogConstants.CHANNEL_NAME, backlog);
    }
}

后台端代码：

redis配置订阅消息：

  

 @Configuration
 @EnableCaching
 public class RedisConfig extends CachingConfigurerSupport {

    //省略其他配置...


    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
                                            MessageListenerAdapter backlogListenerAdapter) {

        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        //订阅了一个叫chat 的通道
        container.addMessageListener(backlogListenerAdapter, new PatternTopic(BacklogConstants.CHANNEL_NAME));
        //这个container 可以添加多个 messageListener
        return container;
    }

    @Bean
    MessageListenerAdapter backlogListenerAdapter(BacklogMsgReceiver receiver) {
        //这个地方 是给messageListenerAdapter 传入一个消息接受的处理器，利用反射的方法调用“receiveMessage”
        //也有好几个重载方法，这边默认调用处理器的方法 叫handleMessage 可以自己到源码里面看
        return new MessageListenerAdapter(receiver, "receiveMsg");
    }
}

redis接收订阅的消息：

@Component
public class BacklogMsgReceiver {

    @Autowired
    RedisAssistant redisAssistant;
    @Autowired
    BacklogService backlogService;

    /**
     * 接受待处理事项的消息，存储到redis中，并同时发送到ws客户端。
     */
    public void receiveMsg(String msg) {
        Backlog obj = (Backlog) redisAssistant.deserializeValue(msg.getBytes());
        backlogService.addBacklog(obj);
    }

}

web socket配置：

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    @Autowired
    WebSocketInterceptor webSocketInterceptor;

    /***
     * 注册 Stomp的端点
     * addEndpoint：添加STOMP协议的端点。提供WebSocket或SockJS客户端访问的地址
     * withSockJS：使用SockJS协议
     */
    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("stomp")
                .setAllowedOrigins("*")//添加允许跨域访问
                .withSockJS() ;
    }

    /**
     * 1、设置拦截器
     * 2、首次连接的时候，获取其Header信息，利用Header里面的信息进行权限认证
     * 3、通过认证的用户，使用 accessor.setUser(user); 方法，将登陆信息绑定在该 StompHeaderAccessor 上，在Controller方法上可以获取 StompHeaderAccessor 的相关信息
     */
    @Override
    public void configureClientInboundChannel(ChannelRegistration registration) {
        registration.interceptors(webSocketInterceptor);
    }

}

service处理逻辑：

@Service
public class BacklogServiceImpl implements BacklogService {

    @Autowired
    RedisAssistant redisAssistant;
    @Autowired
    SimpMessagingTemplate messagingTemplate;


    @Override
    public void addBacklog(Backlog backlog) {
        String key = CacheKeyConstants.BACKLOG_KEY + backlog.getGroup();
        redisAssistant.hset(key, ""+backlog.getId(), backlog);
        messagingTemplate.convertAndSend("/topic/backlog", JSONUtils.toJson(backlog));//发给客户端新增消息
    }

    @Override
    public void removeBacklog(Backlog backlog) {
        String key = CacheKeyConstants.BACKLOG_KEY + backlog.getGroup();
        redisAssistant.hdel(key, ""+backlog.getId());
        backlog.setIsDelete(1);
        messagingTemplate.convertAndSend("/topic/backlog",JSONUtils.toJson(backlog));//发给客户端删除消息

    }

    @Override
    public List<Backlog> getBacklogs(Integer group) {
        String key = CacheKeyConstants.BACKLOG_KEY + group;
        Map m = redisAssistant.hmget(key);
        if (m != null && !m.isEmpty()) {
            return (List<Backlog>) m.values().stream().collect(Collectors.toList());
        }
        return null;
    }

    @Override
    public int countBacklogs(Integer group) {
        String key = CacheKeyConstants.BACKLOG_KEY + group;
        Long size = redisAssistant.hsize(key);
        return size == null ? 0 : size.intValue();
    }

}