Redis学习（九）SpringBoot实现（Pub/Sub）发布订阅

news2024/11/28 8:49:56

目录

    • 一、背景
    • 二、Redis的发布订阅
      • 2.1 订阅单个频道
        • 常用命令
      • 2.2 按规则（Pattern）订阅频道
      • 2.3 不推荐使用的原因
    • 三、SpringBoot实现发布订阅
      • 3.1 RedisUtil.java 发布类
        • 1）MessageDTO.java 实体类
        • 2）发布测试
      • 3.2 订阅实现方式一：实现MessageListener接口
        • 1）RedisConfig.java 配置类
        • 2）RedisMessageListener.java 监听类
        • 3）执行结果
      • 3.2 订阅实现方式二：注入MessageListenerAdapter实例，反射调用
        • 1）RedisConfig.java 配置类
        • 2）RedisMessageReceiver.java 监听类
        • 3）执行结果

一、背景

Redis 的 List 数据类型可以通过 rpush 和 blpop 命令实现消息队列（先进后出），没有任何元素可以弹出的时候，连续会被阻塞。

但是基于 List 实现的消息队列，不支持一对多的消息广播，相当于只有一个消费者。

如果要实现一对多的消息广播，怎么办？

二、Redis的发布订阅

Redis 2.8 及以上版本实现了发布订阅的功能。

2.1 订阅单个频道

首先思考一个问题：如果消息的生产者和消费者是不同的客户端，连接到同一个 Redis。通过什么对象把生产者和消费者关联起来呢？

在 RabbitMQ 里面叫 Queue，在 Kafka 里面叫 Topic，在 Redis 里面叫 channel（频道）。

订阅者可以订阅一个或多个 channel。 消息的发布者可以给指定的 channel 发布消息。只要消息到达了 channel，所有订阅了这个 channel 的订阅者都会收到这条消息。

在这里插入图片描述

常用命令

订阅命令，一次可以订阅多个频道：

subscribe channel-1 channel-2

发布消息，一次只能在一个频道发布：

publish channel-1 hello

取消订阅（不能再订阅状态下使用）：

unsubscribe channel-1

测试-

2.2 按规则（Pattern）订阅频道

支持 ? 和 * 占位符：

  • ?：代表一个字符。
  • *：代表 0 个或多个字符。

例如：现在有三个新闻频道:

  • 运动新闻（news-sport）
  • 音乐新闻（news-music）
  • 天气新闻（news-weather）

在这里插入图片描述

消费端1，订阅运动新闻：

psubscribe *sport

消费端2，订阅所有新闻：

psubscribe news*

消费端3，订阅天气新闻：

psubscribe new-weather

生产者，向三个频道分别发布三条消息，对应的订阅者能收到消息：

publish news-sport kobe
publish news-music jaychou
publish news-weather sunny

2.3 不推荐使用的原因

  • 消息丢失： Redis 的 Pub/Sub 模式不会对消息进行持久化，如果订阅者在消息发布之前未连接到 Redis 服务器，它们将无法接收到之前发布的消息。这意味着如果订阅者在消息发布之前断开连接或重新启动，它们将错过这些消息。
  • 内存占用： 由于 Redis 将所有订阅者的订阅信息存储在内存中，当订阅者数量非常大时，可能会导致 Redis 服务器的内存占用过高。这会对 Redis 的性能和可伸缩性产生负面影响。
  • 阻塞问题： 当订阅者在执行阻塞操作（例如阻塞式读取）时，它们将无法处理其他的 Redis 命令。这可能会导致性能问题，特别是在高并发环境中。
  • 无法保证消息传递顺序： 在 Pub/Sub 模式中，消息的传递是异步的，并且无法保证消息的传递书匈奴。如果应用程序需要处理有序的消息，Pub/Sub 模式可能不适合。

一般来说，考虑到性能和持久化等因素，不建议使用 Redis 的发布订阅功能来实现 MQ。Redis 的一些内部机制用到了发布订阅功能。

三、SpringBoot实现发布订阅

3.1 RedisUtil.java 发布类

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * <p> @Title RedisUtil
 * <p> @Description Redis工具类
 *
 * @author ACGkaka
 * @date 2021/6/16 16:32
 */
@Slf4j
@Component
public class RedisUtil {

    @Qualifier("redisTemplate")
    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    /**
     * 向通道发布消息
     * @param channel   通道
     * @param message   消息
     * @return true成功 false失败
     */
    public boolean publish(String channel, Object message) {
        if (!StringUtils.hasText(channel)) {
            return false;
        }
        try {
            redisTemplate.convertAndSend(channel, message);
            log.info("发送消息成功，channel:{}, message:{}", channel, message);
            return true;
        } catch (Exception e) {
            log.error("发送消息失败，channel:{}, message:{}", channel, message, e);
        }
        return false;
    }

}
1）MessageDTO.java 实体类
package com.demo.redis.listener;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.io.Serializable;
import java.time.LocalDateTime;

@Data
@NoArgsConstructor
@AllArgsConstructor
public class MessageDTO implements Serializable {

    /**
     * 消息标题
     */
    private String title;

    /**
     * 消息内容
     */
    private String content;

    /**
     * 消息内容
     */
    private LocalDateTime createTime;
}
2）发布测试
import com.demo.redis.listener.MessageDTO;
import com.demo.util.RedisUtil;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.time.LocalDateTime;

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringBootRedisApplicationTests {

    @Autowired
    private RedisUtil redisUtil;

    @Test
    public void test1() {
        // 订阅主题
        final String TOPIC_NAME_1 = "TEST_TOPIC_1";
        final String TOPIC_NAME_2 = "TEST_TOPIC_2";
        // 发布消息
        MessageDTO dto = new MessageDTO("测试标题", "测试内容", LocalDateTime.now());
        redisUtil.publish(TOPIC_NAME_1, dto);
    }
}

3.2 订阅实现方式一：实现MessageListener接口

1）RedisConfig.java 配置类
import com.demo.redis.RedisCustomizeProperties;
import com.demo.redis.listener.RedisMessageListener;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.io.Serializable;

@Configuration
@EnableConfigurationProperties(RedisCustomizeProperties.class)
public class RedisConfig {

    /**
     * 配置RedisTemplate
     *
     * @param redisConnectionFactory 连接工厂
     * @return RedisTemplate
     */
    @Bean
    public RedisTemplate<String, Serializable> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate<String, Serializable> redisTemplate = new RedisTemplate<>();
        //设置key的存储方式为字符串
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        //设置为value的存储方式为JDK二进制序列化方式，还有jackson序列化方式（Jackson2JsonRedisSerialize）
        redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
        //设置连接工厂
        redisTemplate.setConnectionFactory(redisConnectionFactory);
        return redisTemplate;
    }

    /**
     * Redis消息监听器容器（实现方式一）
     *
     * @param redisConnectionFactory    连接工厂
     * @param listener                  消息监听器
     * @return Redis消息监听容器
     */
    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory redisConnectionFactory,
                                                   RedisMessageListener listener) {
        // 订阅主题
        final String TOPIC_NAME_1 = "TEST_TOPIC_1";
        final String TOPIC_NAME_2 = "TEST_TOPIC_2";
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        // 设置连接工厂
        container.setConnectionFactory(redisConnectionFactory);
        // 订阅频道（可以添加多个）
        container.addMessageListener(listener, new PatternTopic(TOPIC_NAME_1));
        container.addMessageListener(listener, new PatternTopic((TOPIC_NAME_2)));

        return container;
    }
}
2）RedisMessageListener.java 监听类
package com.demo.redis.listener;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
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;

import javax.annotation.Resource;

@Slf4j
@Component
public class RedisMessageListener implements MessageListener {

    @Qualifier("redisTemplate")
    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Override
    public void onMessage(Message message, byte[] pattern) {
        // 打印渠道
        log.info(">>>>>>>>>> 【INFO】订阅的channel：{}", new String(pattern));

        // 获取消息
        byte[] messageBody = message.getBody();
        // 序列化对象
        MessageDTO messageDTO = (MessageDTO) redisTemplate.getValueSerializer().deserialize(messageBody);

        // 打印消息
        log.info(">>>>>>>>>> 【INFO】收到的message：{}", messageDTO);
    }
}
3）执行结果

执行 3.1 的发布测试，结果如下：

在这里插入图片描述

3.2 订阅实现方式二：注入MessageListenerAdapter实例，反射调用

1）RedisConfig.java 配置类
import com.demo.redis.listener.RedisMessageReceiver;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
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.JdkSerializationRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.io.Serializable;

@Configuration
public class RedisConfig {

    /**
     * 配置RedisTemplate
     *
     * @param redisConnectionFactory 连接工厂
     * @return RedisTemplate
     */
    @Bean
    public RedisTemplate<String, Serializable> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate<String, Serializable> redisTemplate = new RedisTemplate<>();
        //设置key的存储方式为字符串
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        //设置为value的存储方式为JDK二进制序列化方式，还有jackson序列化方式（Jackson2JsonRedisSerialize）
        redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
        //设置连接工厂
        redisTemplate.setConnectionFactory(redisConnectionFactory);
        return redisTemplate;
    }

    /**
     * Redis消息监听器容器（实现方式二）
     *
     * @param redisConnectionFactory    连接工厂
     * @param adapter                   消息监听器
     * @return Redis消息监听容器
     */
    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory redisConnectionFactory,
                                                    MessageListenerAdapter adapter) {
        // 订阅主题
        final String TOPIC_NAME_1 = "TEST_TOPIC_1";
        final String TOPIC_NAME_2 = "TEST_TOPIC_2";
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        // 设置连接工厂
        container.setConnectionFactory(redisConnectionFactory);
        // 订阅频道（可以添加多个）
        container.addMessageListener(adapter, new PatternTopic(TOPIC_NAME_1));
        container.addMessageListener(adapter, new PatternTopic((TOPIC_NAME_2)));

        return container;
    }

    /**
     * 用于接收消息的消息接收器
     * @param receiver
     * @return
     */
    @Bean
    public MessageListenerAdapter listenerAdapter(RedisMessageReceiver receiver) {
        // receiveMessage 为反射调用，用于接收消息的方法名
        MessageListenerAdapter receiveMessage = new MessageListenerAdapter(receiver, "receiveMessage");
        receiveMessage.setSerializer(new JdkSerializationRedisSerializer());
        return receiveMessage;
    }

}
2）RedisMessageReceiver.java 监听类
package com.demo.redis.listener;

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

/**
 * <p> @Title RedisMessageReceiver
 * <p> @Description Redis消息接收器（实现方式二）
 *
 * @author ACGkaka
 * @date 2023/10/7 18:28
 */
@Slf4j
@Component
public class RedisMessageReceiver {

    /**
     * 接收消息（在 RedisConfig.java 中反射调用）
     */
    public void receiveMessage(MessageDTO messageDTO, String channel) {
        // 打印渠道
        log.info(">>>>>>>>>> 【INFO】订阅的channel：{}", channel);

        // 打印消息
        log.info(">>>>>>>>>> 【INFO】收到的message：{}", messageDTO);
    }
}
3）执行结果

执行 3.1 的发布测试，结果如下：

在这里插入图片描述

整理完毕，完结撒花~ 🌻





参考地址：

1.Spring boot整合Redis实现发布订阅（超详细），https://blog.csdn.net/BBQ__ZXB/article/details/124980860

2.springboot中使用redis发布订阅，https://blog.csdn.net/H900302/article/details/113914979

3.Redis的Pub/Sub为何不建议进行消息订阅，https://www.jianshu.com/p/3eff7425429a

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/1070007.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

day10.8ubentu流水灯

day10.8ubentu流水灯

流水灯 .text .global _start _start: 1.设置GPIOE寄存器的时钟使能 RCC_MP_AHB4ENSETR[4]->1 0x50000a28LDR R0,0X50000A28LDR R1,[R0] 从r0为起始地址的4字节数据取出放在R1ORR R1,R1,#(0x1<<4) 第4位设置为1STR R1,[R0] 写回2.设置PE10管脚为输出模式 G…
阅读更多...
C#LINQ

C#LINQ

LINQ&#xff08;Language Integrated Query )语言集成查询&#xff0c;是一组用于C#和VB语言的拓展&#xff0c;它允许VB或者C#代码以操作内存数据的方式&#xff0c;查询数据库。 LINQ使用的优点&#xff1a; 无需复杂学习过程即可上手。编写更少代码即可创建完整应用。更快…
阅读更多...
okhttp4.11源码分析

okhttp4.11源码分析

目录 一&#xff0c;OKHTTP时序图 二&#xff0c;OKHTTP类图 三&#xff0c;OKHTTP流程图 一&#xff0c;OKHTTP时序图 上图是整个okhttp一次完整的请求过程&#xff0c;时序图里面有些部分为了方便采用了简单的描述&#xff0c;描述了主要的流程&#xff0c;细节的话&#…
阅读更多...
数据结构之堆，栈的实现

数据结构之堆，栈的实现

首先我们分析由于只需要尾进尾出&#xff0c;用数组模拟更简单。 实现的功能如上图。 top可以表示栈中元素个数。 capacity表示栈的容量。 首先是堆的初始化 再就是栈的插入和删除 然后实现显示栈顶元素 大小和检测是否为空的实现 销毁栈的实现&#xff08;防止内存泄露&…
阅读更多...
【无标题】Delayed延迟队列不工作

【无标题】Delayed延迟队列不工作

背景 项目中使用java 自带的延迟队列Delayed&#xff0c;只有添加进队列的消息&#xff0c;并没有被消费到 版本 jdk1.8 问题原因 上一个消费队列出现异常并且没有捕获&#xff0c;下一个队列就没有进行消费 复现代码 没有抛异常的情况下 package com.ccb.core.config.…
阅读更多...
10.8c++作业

10.8c++作业

#include <iostream>using namespace std; class Rect {int width; //宽int height; //高 public://初始化函数void init(int w,int h){widthw;heighth;}//更改宽度void set_w(int w){widthw;}//更改高度void set_h(int h){heighth;}//输出矩形周长和面积void show(){co…
阅读更多...
2023年铷铁硼行业分析：低端供应过剩，高性能材料供应不足[图]

2023年铷铁硼行业分析：低端供应过剩，高性能材料供应不足[图]

铷铁硼材料是一种Fe基磁性材料&#xff0c;主要由钕铁硼按一定比例组成的四方晶体结构&#xff0c;其中Fe元素约占总质量的三分之二&#xff0c;Nd元素约占总量的三分之一&#xff0c;而B等含量最少&#xff0c;约占1%。铷铁硼是现今磁性最强的永久磁铁&#xff0c;也是最常使用…
阅读更多...
波奇学C++：用红黑树模拟实现map和set

波奇学C++：用红黑树模拟实现map和set

用同一个树的类模板封装map(key/value)和set(key) 红黑树的Node template<class T> struct RBTreeNode {RBTreeNode<T>* _left;RBTreeNode<T>* _right;RBTreeNode<T>* _parent;T _data;Colour _col;RBTreeNode(const T& data):_left(nullptr),_r…
阅读更多...
python工具-内存采集展示

python工具-内存采集展示

1. 查看某个进程的内存占用 1. 查看某个进程的内存占用 1.1. 采集1.2. 分析 1.1. 采集 下边内存保存为 cat-memory.sh 脚本文件&#xff0c;赋予可执行权限执行 ./cat-memory.sh pid 会生成 pid.txt #!/bin/bashprocess$1 out$1.txt pid$1echo 时间 内存(KB) >> $ou…
阅读更多...
复旦大学EMBA：揭秘科创企业，领略未来战略！

复旦大学EMBA：揭秘科创企业，领略未来战略！

智能制造&#xff0c;国之重器。作为制造强国建设的主攻方向&#xff0c;智能制造的发展水平关系到我国未来制造业在全球的地位与影响力。发展智能制造&#xff0c;是加快建设现代化产业体系的重要手段&#xff0c;提升供给体系适配性的有力抓手&#xff0c;也是建设数字中国的…
阅读更多...
E. Monsters

E. Monsters

Problem - 1810E - Codeforces 思路&#xff1a;我们总结一下题意&#xff0c;能够得到这个题其实就是让我们从某个0开始搜索&#xff0c;然后看看是否可以遍历所有得节点&#xff0c;那么如果采用暴力得话那就是n^2logn&#xff0c;因为我们遍历一次使用优先队列得话是nlogn的…
阅读更多...
Stm32_标准库_8_ADC_光敏传感器_测量具体光照强度

Stm32_标准库_8_ADC_光敏传感器_测量具体光照强度

ADC简介 测量方式 采用二分法比较数据 IO通道 ADC基本结构及配置路线 获取数字变量需要用到用到光敏电阻的AO口&#xff0c;AO端口接在PA0引脚即可 测得的模拟数据与实际光照强度之间的关系为 光照强度 100 - 模拟量 / 40;代码&#xff1a; 完整朴素代码&#xff1a; #in…
阅读更多...
Mysql存储-EAV模式

Mysql存储-EAV模式

Mysql存储-EAV模式 最近又又又搞一点新东西&#xff0c;要整合不同业务进行存储和查询&#xff0c;一波学习过后总结了一下可扩展性MAX的eav模式存储。 在eav这里的数据结构设计尤为关键&#xff0c;需要充分考虑你需要使用的字段、使用场景&#xff0c;当数据结构设计完成后便…
阅读更多...
skywalking功能介绍

skywalking功能介绍

服务 服务信息 请求接口后查看skywalking&#xff0c;可以看到有一个请求&#xff0c;响应时间为1852ms&#xff0c;性能指数Apdex为0.5。 详细表盘 点进应用可以看到表盘 可以看到显示有一个slow endpoints&#xff0c;就是我请求的这个接口。 JVM信息 也可以看到JVM信息。…
阅读更多...
点餐小程序实战教程06-首页开发

点餐小程序实战教程06-首页开发

用户注册功能开发好了之后&#xff0c;我们就要开发小程序&#xff0c;首先我们是规划小程序的功能模块&#xff0c;我们一共是四个模块&#xff0c;分别是首页、订单、消息和我的。 首页我们主要是点餐的功能&#xff0c;可以选择菜品&#xff0c;加入到购物车&#xff0c;然…
阅读更多...
deckGL自定义图层学习笔记

deckGL自定义图层学习笔记

1.自定义图层 当使用DeckGL提供的图层还无法满足需求时&#xff08;https://deck.gl/docs/api-reference/layers&#xff09;&#xff0c;可能就需要自定义图层了。在DeckGL中有常见的三种自定义图层的方式 创建复合层&#xff08;composite layers.&#xff09;——复合层是一…
阅读更多...
ffmpeg从一个视频中提取音频

ffmpeg从一个视频中提取音频

ffmpeg -i ~/video/video.mp4 -vn -acodec copy ~/video/audioFile.m4a 从video.mp4中提取音频到文件audioFile.m4a中 查看提取的音频文件 ffprobe ~/video/audioFile.m4a
阅读更多...
OneDrive下的OneNote扩容方法，及查看OneDrive容量的方法（详细图文教程）

OneDrive下的OneNote扩容方法，及查看OneDrive容量的方法（详细图文教程）

目录 一、内存不足的问题二、土豪续费扩容法三、X宝扩容法3.1 购买链接3.2 登录接口3.3 详细图文操作过程3.3.1 获取链接&#xff1a;3.3.2 用订单号和获取链接扩容&#xff1a; 3.4 扩容后的容量 四、查看自己OneDrive的容量五、总结 一、内存不足的问题 一直都在用OneNote记…
阅读更多...
STM32H723 CubeMX 三路FDCAN 代码

STM32H723 CubeMX 三路FDCAN 代码

时钟频率 FDCAN1 设置250kbit/s FDCAN2 设置500kbit/s FDCAN3 设置500kbit/s fdcan.c /* USER CODE BEGIN Header */ /********************************************************************************* file fdcan.c* brief This file provides code fo…
阅读更多...
【刷题篇】回溯算法（深度优先搜索（二））

【刷题篇】回溯算法（深度优先搜索（二））

文章目录 岛屿数量电话号码的字母组合组合总和活字印刷 岛屿数量 给你一个由 ‘1’&#xff08;陆地&#xff09;和 ‘0’&#xff08;水&#xff09;组成的的二维网格&#xff0c;请你计算网格中岛屿的数量。 岛屿总是被水包围&#xff0c;并且每座岛屿只能由水平方向和/或竖直…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023