JAVA 实现 Redis 发布订阅-Toy模板网

这篇具有很好参考价值的文章主要介绍了JAVA 实现 Redis 发布订阅。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

Redis 发布订阅

发布订阅：消息发布者发布消息 和 消息订阅者接收消息，两者之间通过某种媒介联系起来

例如订杂志，当自己订阅了爱格杂志，每个月会发刊一本。到发布的时候派送员将杂志送到自己手上就能看到杂志内容。只有我们订阅了该杂志才会派送给我们

Redis 发布订阅（pub/sub）是一种消息通信模式：发送者（pub）发送消息，订阅者（sub）接收消息。

Redis 客户端可以订阅任意数量的频道。

订阅 / 发布消息图：

java redis 订阅,redis,java,缓存

图中可以看出，所需：

消息发送者、 2. 频道、 3. 消息订阅者

发布订阅机制

当一个客户端通过 PUBLISH 命令向订阅者发布消息的时候，称这个客户端为发布者publisher
当一个客户端通过subscribe 或者 PSUBSCRIBE 接收消息时，称这个客户端为订阅者 subscriber
为了解耦发布者和订阅者之间的关系，Redis 使用了频道channel(频道)作为两者之间的中介，发布者直接把消息发送给 channel，而 channel 负责把消息发送给订阅者，发布者和订阅者之间没有直接的联系，都不知道对方的存在

订阅者 1，2，3 订阅了频道 channel，当有消息发布给频道时，这个消息就会被发送到三个订阅者客户端

java redis 订阅,redis,java,缓存

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

Listener模式

创建一个监听容器

@Configuration
public class CatListenerConfig extends CachingConfigurerSupport {

    /**
     * 消息监听容器
     *
     * @param factory
     * @return
     */
    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory factory){
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(factory);
        //订阅一个通道 该处的通道名是发布消息时的名称
       container.setConnectionFactory(connectionFactory);
        //订阅了一个叫cat 的通道
        container.addMessageListener(catAdapter, new PatternTopic("cat"));
        container.addMessageListener(fishAdapter, new PatternTopic("fish"));
        return container;
    }

    

}

@Component
public class CatListener implements MessageListener {
    @Override
    public void onMessage(Message message, byte[] bytes) {
        System.out.println("我是监听者,我监听到的消息是 " + 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  implements MessageListener  {

    @Autowired
    RedisTemplate redisTemplate;

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

建测试类，测试发布监听

@RestController
public class TestController {

    @Resource
    StringRedisTemplate stringRedisTemplate;

    @PostMapping("/cat")
    public void test2(){
        stringRedisTemplate.convertAndSend("cat","测试：消息发布者发布消息");
    }
  
   @PostMapping("/fish")
    public void fish(){
        stringRedisTemplate.convertAndSend("fish","测试：消息发布者发布消息");
    }
}

测试结果

java redis 订阅,redis,java,缓存

Adapter模式

2.1 配置文章来源地址https://www.toymoban.com/news/detail-634209.html

    /**
     * MessageListenerAdapter 模式
     * 该处topic的 key为bean的name
     * @param connectionFactory
     * @param adapterMap
     * @return
     */
    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,Map<String, MessageListenerAdapter> adapterMap) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        adapterMap.keySet().forEach(topic-> container.addMessageListener(adapterMap.get(topic),new PatternTopic(topic)));
        return container;
    }

监听1

package com.sst.loan.risk.listener;

import com.sst.loan.risk.manager.RuleLoadManager;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * 监听适配器
 *
 * @author 蔡定努
 * @date 2023/06/13 10:26
 */
@Slf4j
@Component("ruleRefreshAdapter")
public class RuleRefreshAdapter  extends MessageListenerAdapter {
  


    @Override
    public void onMessage(Message message, byte[] bytes) {
        log.info(">>>>>>> 消息适配器收到刷新规则的请求 <<<<<<<<<<<<<<");

    }
}

监听2

package com.sst.loan.risk.listener;

import com.sst.loan.risk.manager.RuleLoadManager;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * 监听适配器
 *
 * @author 蔡定努
 * @date 2023/06/13 10:26
 */
@Slf4j
@Component("ruleRefreshAdapter2")
public class RuleRefreshAdapter2  extends MessageListenerAdapter {
  


    @Override
    public void onMessage(Message message, byte[] bytes) {
        log.info(">>>>>>> 消息适配器收到刷新规则的请求 <<<<<<<<<<<<<<");
        
    }
}

测试


    @Resource
    private StringRedisTemplate stringRedisTemplate;


    /**
     * 
     * @author 蔡定努
     */
    @GetMapping("refresh")
    public Object refresh() {
        stringRedisTemplate.convertAndSend("ruleRefreshAdapter","refresh");
        return Result.success();
    }
    
    
    /**
     * 
     * @author 蔡定努
     */
    @GetMapping("refresh")
    public Object refresh() {
        stringRedisTemplate.convertAndSend("ruleRefreshAdapter2","refresh");
        return Result.success();
    }