JAVA 實作 Redis 釋出訂閱

Redis 釋出訂閱

釋出訂閱：消息釋出者釋出消息 和 消息訂閱者接收消息，兩者之間通過某種媒介聯系起來

Redis 釋出訂閱（pub/sub）是一種 消息通信模式 ：發送者（pub）發送消息，訂閱者（sub）接收消息。

Redis 用戶端可以訂閱任意數量的頻道。

訂閱 / 釋出消息圖：

圖中可以看出，所需：

1. 消息發送者 、 2. 頻道 、 3. 消息訂閱者

釋出訂閱機制

1. 當一個用戶端通過 ​

   ​PUBLISH​

   ​​ 指令向訂閱者釋出消息的時候，稱這個用戶端為釋出者​

   ​publisher​

   ​
2. 當一個用戶端通過​

   ​subscribe​

   ​​ 或者 ​

   ​PSUBSCRIBE​

   ​​ 接收消息時，稱這個用戶端為 訂閱者 ​

   ​subscriber​

   ​
3. 為了解耦釋出者和訂閱者之間的關系，Redis 使用了頻道​

   ​channel(頻道)​

   ​作為兩者之間的中介，釋出者直接把消息發送給 channel，而 channel 負責把消息發送給訂閱者，釋出者和訂閱者之間沒有直接的聯系，都不知道對方的存在

訂閱者 1，2，3 訂閱了頻道 channel，當有消息釋出給頻道時，這個消息就會被發送到三個訂閱者用戶端

demo 實作

​​學習參考連結​​

引入依賴

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>      

添加配置檔案

spring:
  redis:
    host: 127.0.0.1
    database: 5
    password:
    port: 6379      

建立一個監聽類，來監聽消息

package com.maoxs.listener;

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.stereotype.Component;

/**
 * 監聽發送的消息
 */
@Component
public class CatListener extends MessageListenerAdapter implements MessageListener  {

    @Autowired
    RedisTemplate redisTemplate;

    @Override
    public void onMessage(Message message, byte[] bytes) {
        System.out.println("我是Cat監聽" + message.toString());
    }
}      

package com.maoxs.listener;

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;

/**
 * 監聽發送的消息
 */
public class FishListener extends MessageListenerAdapter implements MessageListener  {

    @Autowired
    RedisTemplate redisTemplate;

    @Override
    public void onMessage(Message message, byte[] bytes) {
        System.out.println("我是Fish監聽" + message.toString());
    }
}      

建立一個監聽容器

package com.maoxs.redis;

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.maoxs.pojo.MessageReceiver;
import org.springframework.cache.CacheManager;
import org.springframework.cache.annotation.CachingConfigurerSupport;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.cache.RedisCacheManager;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;

import java.util.Arrays;

@Configuration
public class RedisMessageConfig {


    /**
     * redis消息監聽器容器
     * 可以添加多個監聽不同話題的redis監聽器，隻需要把消息監聽器和相應的消息訂閱處理器綁定，該消息監聽器
     * 通過反射技術調用消息訂閱處理器的相關方法進行一些業務處理
     * @param connectionFactory
     * @param listenerAdapter
     * @return
     */
    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
                                            CatListener catAdapter, FishListener fishAdapter) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        //訂閱了一個叫chat 的通道
        container.addMessageListener(catAdapter, new PatternTopic("cat"));
        container.addMessageListener(fishAdapter, new PatternTopic("fish"));
        //這個container 可以添加多個 messageListener
        return container;
    }

    /**
     * 消息監聽器擴充卡，綁定消息處理器
     *
     * @param receiver
     * @return
     */
   // @Bean
   // MessageListenerAdapter CatAdapter() {
   //     return new MessageListenerAdapter(new CatListener());
   // }

    /**
     * 消息監聽器擴充卡，綁定消息處理器
     *
     * @param receiver
     * @return
     */
  //  @Bean
  //  MessageListenerAdapter FishAdapter() {
   //     return new MessageListenerAdapter(new FishListener());
   // }

    /**
     * redis 讀取内容的template
     */
    @Bean
    StringRedisTemplate stringRedisTemplate(RedisConnectionFactory connectionFactory) {
        return new StringRedisTemplate(connectionFactory);
    }

    @Bean
    public RedisTemplate<String, String> redisTemplate(RedisConnectionFactory factory) {
        StringRedisTemplate template = new StringRedisTemplate(factory);

        //定義value的序列化方式
        Jackson2JsonRedisSerializer 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);


        template.setValueSerializer(jackson2JsonRedisSerializer);
        template.setHashKeySerializer(jackson2JsonRedisSerializer);
        template.afterPropertiesSet();
        return template;
    }
}      

@RestController
public class TestController {
    
    @Resource
    StringRedisTemplate stringRedisTemplate;
 


    @GetMapping("cat")
    public void sendCatMessage(){
        stringRedisTemplate.convertAndSend("cat","貓");
    }

    @GetMapping("fish")
    public void sendFishMessage(){
        stringRedisTemplate.convertAndSend("fish","魚");
    }