redis订阅/发布

redis订阅发布实现是通过类似于Map<Key，List>这样的数据结构，使用subscribe监听某个channel（通道）也就是key，然后将自己加入对应Key的list中，当有publish命令将msg推送到对应的channel中时，遍历对应的list，发送数据。这是redis实现的原理。说说java中redis订阅发布我看到的吧！

我们需要创建消息监听器容器，然后将消息监听器与相应的消息订阅处理器绑定，其中消息监听适配器需通过反射指定对应的处理方法，在容器中消息订阅器订阅的时Key（这里我用枚举来处理），然后创建发布者和监听者对应的方法就可以了

1.创建对应的key值

/**
 * @Auther: eros
 * @Date: 2019/2/27 15:43
 * @Description:
 * 发布/订阅 key 枚举
 */
public enum WebSocketContent {

    WS_EVENT_TYPE,
    WS_EVENT_DEV,
    WS_EVENT_DEV_VERSION;

}

2.配置redis消息监听器

因为目前页面上有三个数据需要后台websocket进行传输，所以绑定了三个消息监听器和相应的消息订阅处理器，这里没有添加redisTemplate，根据自己的需要添加相对应的

    /* =================================== redis listener =================================== */

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

    /**
     * 消息监听器适配器，绑定消息处理器，利用反射技术调用消息处理器的业务方法
     * @param redisReceiver
     * @return
     */
    @Bean(name = "listenerAdapter_dev")
    public MessageListenerAdapter listenerAdapterDev(RedisReceiverService redisReceiver) {
        return new MessageListenerAdapter(redisReceiver, "receiveMessageByDev");
    }


    /**
     * @param redisReceiver
     * @return
     */
    @Bean(name = "listenerAdapter_type")
    public MessageListenerAdapter listenerAdapterType(RedisReceiverService redisReceiver) {
        return new MessageListenerAdapter(redisReceiver, "receiveMessageByType");
    }


    /**
     * @param redisReceiver
     * @return
     */
    @Bean(name = "listenerAdapter_dev_version")
    public MessageListenerAdapter listenerAdapterDevVersion(RedisReceiverService redisReceiver) {
        return new MessageListenerAdapter(redisReceiver, "receiveMessageByDevVersion");
    }

3.创建发布者，需要发布数据的地方都是用这一个接口

/**
 * @Auther: eros
 * @Date: 2019/2/27 15:37
 * @Description: 发布者
 */
@Service
public class RedisPublisherService {

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

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * @auther: eros
     * @param content 发布类型
     * @param message 发布消息
     * @date: 2019/2/27 16:04
     * @description
     * @return: void
     */
    public void sendMessage(WebSocketContent content,String message) {
        LOGGER.info("》》》》》》》》》》》》》》》》》》》》》》》
");
        LOGGER.info("发布信息");
        try {
            redisTemplate.convertAndSend(content.name(), message);
            LOGGER.info("发布成功");
            LOGGER.info("《《《《《《《《《《《《《《《《《《《《《《《
");
        } catch (Exception e) {
            e.printStackTrace();
            LOGGER.info("发布失败");
            LOGGER.info("《《《《《《《《《《《《《《《《《《《《《《《
");
        }
    }

}

4.创建对应的处理器，这里与上次写的websocket配套使用

/**
 * @Auther: eros
 * @Date: 2019/2/27 16:01
 * @Description:
 */
@Service
public class RedisReceiverService{

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

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * @param message
     */
    public void receiveMessageByDev(String message) {
        message = message.substring(message.indexOf("["));
        LOGGER.info("收到的mq消息 %s - %s" ,WebSocketContent.WS_EVENT_DEV,message);
        sendMessage(WebSocketContent.WS_EVENT_DEV +":" + JSON.toJSONString(message));
    }

    /**
     * @param message
     */
    public void receiveMessageByType(String message) {
        message = message.substring(message.indexOf("["));
        LOGGER.info("收到的mq消息 %s - %s" ,WebSocketContent.WS_EVENT_TYPE,message);
        sendMessage(WebSocketContent.WS_EVENT_TYPE +":" + JSON.toJSONString(message));
    }

    /**
     * @param message
     */
    public void receiveMessageByDevVersion(String message) {
        message = message.substring(message.indexOf("["));
        LOGGER.info("收到的mq消息 %s - %s" ,WebSocketContent.WS_EVENT_DEV_VERSION,message);
        sendMessage(WebSocketContent.WS_EVENT_DEV_VERSION +":" + JSON.toJSONString(message));
    }

    /**
     * @auther: eros
     * @param message 1
     * @date: 2019/2/27 16:22
     * @description 发送信息
     * @return: void
     */
    private void sendMessage(String message){
        Map<String,ChatRoomServerEndpoint> sessionMap = ChatRoomServerEndpoint.getSessionMap();
        for(Map.Entry<String,ChatRoomServerEndpoint> entries : sessionMap.entrySet()){
            LOGGER.info("content:ip - {} 获取",entries.getKey());
            ChatRoomServerEndpoint webSocketServer = entries.getValue();
            try {
                webSocketServer.getSession().getBasicRemote().sendText(message);
            } catch (IOException e) {
                LOGGER.error("JSON转换有问题");
                e.printStackTrace();
            }
        }
    }

}