• Redis - 订阅发布替换 Etcd 解决方案

    为了减轻项目的中间件臃肿，由于我们项目本身就应用了 Redis，正好 Redis 的也具备订阅发布监听的特性，正好应对 Etcd 的功能，所以本次给大家讲解如何使用 Redis 消息订阅发布来替代 Etcd 的解决方案。接下来，我们先看 Redis 订阅发布的常见情景……

    Redis 订阅发布公共类

    RedisConfig.java
    1. import com.fasterxml.jackson.annotation.JsonAutoDetect;
    2. import com.fasterxml.jackson.annotation.PropertyAccessor;
    3. import com.fasterxml.jackson.databind.ObjectMapper;
    4. import org.springframework.context.annotation.Bean;
    5. import org.springframework.context.annotation.ComponentScan;
    6. import org.springframework.context.annotation.Configuration;
    7. import org.springframework.data.redis.connection.RedisConnectionFactory;
    8. import org.springframework.data.redis.core.RedisTemplate;
    9. import org.springframework.data.redis.listener.RedisMessageListenerContainer;
    10. import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
    11. import org.springframework.data.redis.serializer.StringRedisSerializer;
    12. import java.net.UnknownHostException;
    13.  
    14. @Configuration
    15. @ComponentScan({"cn.hutool.extra.spring"})
    16. public class RedisConfig {
    17.  
    18.     @Bean
    19.     RedisMessageListenerContainer container (RedisConnectionFactory redisConnectionFactory){
    20.         RedisMessageListenerContainer container = new RedisMessageListenerContainer();
    21.         container.setConnectionFactory(redisConnectionFactory);
    22.         return  container;
    23.     }
    24.     
    25.     @Bean
    26.     public RedisTemplate redisTemplate(RedisConnectionFactory redisConnectionFactory) throws UnknownHostException {
    27.         RedisTemplate template = new RedisTemplate();
    28.         // 连接工厂
    29.         template.setConnectionFactory(redisConnectionFactory);
    30.         // 序列化配置
    31.         Jackson2JsonRedisSerializer objectJackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
    32.         ObjectMapper objectMapper = new ObjectMapper();
    33.         objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
    34.         objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
    35.         objectJackson2JsonRedisSerializer.setObjectMapper(objectMapper);
    36.         StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
    37.         // 配置具体序列化
    38.         // key采用string的序列化方式
    39.         template.setKeySerializer(stringRedisSerializer);
    40.         // hash的key采用string的序列化方式
    41.         template.setHashKeySerializer(stringRedisSerializer);
    42.         // value序列化采用jackson
    43.         template.setValueSerializer(objectJackson2JsonRedisSerializer);
    44.         // hash的value序列化采用jackson
    45.         template.setHashValueSerializer(objectJackson2JsonRedisSerializer);
    46.         template.afterPropertiesSet();
    47.         return template;
    48.     }
    49. }
    RedisUtil.java
    1. import lombok.extern.slf4j.Slf4j;
    2. import org.springframework.data.redis.core.RedisTemplate;
    3. import org.springframework.stereotype.Component;
    4. import javax.annotation.Resource;
    5.  
    6. @Slf4j
    7. @Component
    8. public class RedisUtil {
    9.  
    10.     @Resource
    11.     private RedisTemplate redisTemplate;
    12.     /**
    13.      * 消息发送
    14.      * @param topic 主题
    15.      * @param message 消息
    16.      */
    17.     public void publish(String topic, String message) {
    18.         redisTemplate.convertAndSend(topic, message);
    19.     }
    20. }
    application.yml
    1. server:
    2.   port: 7077
    3. spring:
    4.   application:
    5.     name: redis-demo
    6.   redis:
    7.     host: localhost
    8.     timeout: 3000
    9.     jedis:
    10.       pool:
    11.         max-active: 300
    12.         max-idle: 100
    13.         max-wait: 10000
    14.     port: 6379
    RedisController.java
    1. import org.springframework.web.bind.annotation.*;
    2. import javax.annotation.Resource;
    3.  
    4. /**
    5.  * @author Lux Sun
    6.  * @date 2023/9/12
    7.  */
    8. @RestController
    9. @RequestMapping("/redis")
    10. public class RedisController {
    11.  
    12.     @Resource
    13.     private RedisUtil redisUtil;
    14.     
    15.     @PostMapping
    16.     public String publish(@RequestParam String topic, @RequestParam String msg) {
    17.        redisUtil.publish(topic, msg);
    18.        return "发送成功: " + topic + " - " + msg;
    19.     }
    20. }

    一、业务情景：1 个消费者监听 1 个 Topic

    教程三步走（下文业务情景类似不再描述）
    1. 实现接口 MessageListener
    2. 消息订阅，绑定业务 Topic
    3. 重写 onMessage 消费者业务方法
    1. import lombok.extern.slf4j.Slf4j;
    2. import org.springframework.context.annotation.Bean;
    3. import org.springframework.data.redis.connection.Message;
    4. import org.springframework.data.redis.connection.MessageListener;
    5. import org.springframework.data.redis.listener.PatternTopic;
    6. import org.springframework.data.redis.listener.RedisMessageListenerContainer;
    7. import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
    8. import org.springframework.stereotype.Component;
    9. import javax.annotation.Resource;
    10.  
    11. @Slf4j
    12. @Component
    13. public class RedisReceiver1 implements MessageListener {
    14.  
    15.     @Resource
    16.     private RedisMessageListenerContainer container;
    17.  
    18.     /**
    19.      * 重点关注这方法, 进行消息订阅
    20.      */
    21.     @PostConstruct
    22.     public void init() {
    23.         MessageListenerAdapter adapter = new MessageListenerAdapter(this);
    24.         // 绑定 Topic 语法为正则表达式
    25.         container.addMessageListener(adapter, new PatternTopic("topic1.*"));
    26.     }
    27.  
    28.     @Override
    29.     public void onMessage(Message message, byte[] bytes) {
    30.         String key = new String(message.getChannel());
    31.         String value = new String(message.getBody());
    32.         log.info("Key: {}", key);
    33.         log.info("Value: {}", value);
    34.     }
    35. }
    测试
    1. curl --location '127.0.0.1:7077/redis' \
    2. --header 'Content-Type: application/x-www-form-urlencoded' \
    3. --data-urlencode 'topic=topic1.msg' \
    4. --data-urlencode 'msg=我是消息1'
    结果
    1. 2023-11-15 10:22:38.445  INFO 59189 --- [    container-2] com.xxx.redis.demo.RedisReceiver1  : Key: topic1.msg
    2. 2023-11-15 10:22:38.445  INFO 59189 --- [    container-2] com.xxx.redis.demo.RedisReceiver1  : Value: "我是消息1"

    二、业务情景：1 个消费者监听 N 个 Topic

    1. import lombok.extern.slf4j.Slf4j;
    2. import org.springframework.data.redis.connection.Message;
    3. import org.springframework.data.redis.connection.MessageListener;
    4. import org.springframework.data.redis.listener.PatternTopic;
    5. import org.springframework.data.redis.listener.RedisMessageListenerContainer;
    6. import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
    7. import org.springframework.stereotype.Component;
    8. import javax.annotation.PostConstruct;
    9. import javax.annotation.Resource;
    10.  
    11. @Slf4j
    12. @Component
    13. public class RedisReceiver1 implements MessageListener {
    14.  
    15.     @Resource
    16.     private RedisMessageListenerContainer container;
    17.  
    18.     /**
    19.      * 重点关注这方法, 进行消息订阅
    20.      */
    21.     @PostConstruct
    22.     public void init() {
    23.         MessageListenerAdapter adapter = new MessageListenerAdapter(this);
    24.         // 绑定 Topic 语法为正则表达式
    25.         container.addMessageListener(adapter, new PatternTopic("topic1.*"));
    26.         // 只需再绑定业务 Topic 即可
    27.         container.addMessageListener(adapter, new PatternTopic("topic2.*"));
    28.     }
    29.  
    30.     @Override
    31.     public void onMessage(Message message, byte[] bytes) {
    32.         String key = new String(message.getChannel());
    33.         String value = new String(message.getBody());
    34.         log.info("Key: {}", key);
    35.         log.info("Value: {}", value);
    36.     }
    37. }
    测试
    1. curl --location '127.0.0.1:7077/redis' \
    2. --header 'Content-Type: application/x-www-form-urlencoded' \
    3. --data-urlencode 'topic=topic2.msg' \
    4. --data-urlencode 'msg=我是消息2'
    结果
    1. 2023-11-15 10:22:38.445  INFO 59189 --- [    container-2] com.xxx.redis.demo.RedisReceiver1  : Key: topic2.msg
    2. 2023-11-15 10:22:38.445  INFO 59189 --- [    container-2] com.xxx.redis.demo.RedisReceiver1  : Value: "我是消息2"

    三、业务情景：N 个消费者监听 1 个 Topic

    我们看一下，现在又新增一个 RedisReceiver2，按理讲测试的时候，RedisReceiver1 和 RedisReceiver2 会同时收到消息

    1. import lombok.extern.slf4j.Slf4j;
    2. import org.springframework.data.redis.connection.Message;
    3. import org.springframework.data.redis.connection.MessageListener;
    4. import org.springframework.data.redis.listener.PatternTopic;
    5. import org.springframework.data.redis.listener.RedisMessageListenerContainer;
    6. import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
    7. import org.springframework.stereotype.Component;
    8. import javax.annotation.PostConstruct;
    9. import javax.annotation.Resource;
    10.  
    11. @Slf4j
    12. @Component
    13. public class RedisReceiver2 implements MessageListener {
    14.  
    15.     @Resource
    16.     private RedisMessageListenerContainer container;
    17.  
    18.     /**
    19.      * 重点关注这方法, 进行消息订阅
    20.      */
    21.     @PostConstruct
    22.     public void init() {
    23.         MessageListenerAdapter adapter = new MessageListenerAdapter(this);
    24.         // 绑定 Topic 语法为正则表达式
    25.         container.addMessageListener(adapter, new PatternTopic("topic1.*"));
    26.     }
    27.  
    28.     @Override
    29.     public void onMessage(Message message, byte[] bytes) {
    30.         String key = new String(message.getChannel());
    31.         String value = new String(message.getBody());
    32.         log.info("Key: {}", key);
    33.         log.info("Value: {}", value);
    34.     }
    35. }
    测试
    1. curl --location '127.0.0.1:7077/redis' \
    2. --header 'Content-Type: application/x-www-form-urlencoded' \
    3. --data-urlencode 'topic=topic1.msg' \
    4. --data-urlencode 'msg=我是消息1'
    结果
    1. 2023-11-15 10:22:38.445  INFO 59189 --- [    container-2] com.xxx.redis.demo.RedisReceiver1  : Key: topic1.msg
    2. 2023-11-15 10:22:38.449  INFO 59189 --- [    container-3] com.xxx.redis.demo.RedisReceiver2  : Key: topic1.msg
    3. 2023-11-15 10:22:38.545  INFO 59189 --- [    container-2] com.xxx.redis.demo.RedisReceiver1  : Value: "我是消息1"
    4. 2023-11-15 10:22:38.645  INFO 59189 --- [    container-3] com.xxx.redis.demo.RedisReceiver2  : Value: "我是消息1"

    四、业务情景：N 个消费者监听 N 个 Topic

    都到这阶段了，应该不难理解了吧~

    1. import lombok.extern.slf4j.Slf4j;
    2. import org.springframework.data.redis.connection.Message;
    3. import org.springframework.data.redis.connection.MessageListener;
    4. import org.springframework.data.redis.listener.PatternTopic;
    5. import org.springframework.data.redis.listener.RedisMessageListenerContainer;
    6. import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
    7. import org.springframework.stereotype.Component;
    8. import javax.annotation.PostConstruct;
    9. import javax.annotation.Resource;
    10.  
    11. @Slf4j
    12. @Component
    13. public class RedisReceiver2 implements MessageListener {
    14.  
    15.     @Resource
    16.     private RedisMessageListenerContainer container;
    17.  
    18.     /**
    19.      * 重点关注这方法, 进行消息订阅
    20.      */
    21.     @PostConstruct
    22.     public void init() {
    23.         MessageListenerAdapter adapter = new MessageListenerAdapter(this);
    24.         // 绑定 Topic 语法为正则表达式
    25.         container.addMessageListener(adapter, new PatternTopic("topic1.*"));
    26.         // 只需再绑定业务 Topic 即可
    27.         container.addMessageListener(adapter, new PatternTopic("topic2.*"));
    28.     }
    29.  
    30.     @Override
    31.     public void onMessage(Message message, byte[] bytes) {
    32.         String key = new String(message.getChannel());
    33.         String value = new String(message.getBody());
    34.         log.info("Key: {}", key);
    35.         log.info("Value: {}", value);
    36.     }
    37. }
    测试
    1. curl --location '127.0.0.1:7077/redis' \
    2. --header 'Content-Type: application/x-www-form-urlencoded' \
    3. --data-urlencode 'topic=topic2.msg' \
    4. --data-urlencode 'msg=我是消息2'
    结果
    1. 2023-11-15 10:22:38.445  INFO 59189 --- [    container-2] com.xxx.redis.demo.RedisReceiver1  : Key: topic2.msg
    2. 2023-11-15 10:22:38.449  INFO 59189 --- [    container-3] com.xxx.redis.demo.RedisReceiver2  : Key: topic2.msg
    3. 2023-11-15 10:22:38.545  INFO 59189 --- [    container-2] com.xxx.redis.demo.RedisReceiver1  : Value: "我是消息2"
    4. 2023-11-15 10:22:38.645  INFO 59189 --- [    container-3] com.xxx.redis.demo.RedisReceiver2  : Value: "我是消息2"

    好了，Redis 订阅发布的教程到此为止。接下来，我们看下如何用它来替代 Etcd 的业务情景？

    这之前，我们先大概聊下 Etcd 的 2 个要点：

    1. Etcd 消息事件类型
    2. Etcd 持久层数据

    那么问题来了，Redis 虽然具备基本的消息订阅发布，但是如何契合 Etcd 的这 2 点特性，我们目前给出对应的解决方案是：

    1. 使用 Redis K-V 的 value 作为 Etcd 消息事件类型
    2. 使用 MySQL 作为 Etcd 持久层数据：字段 id 随机 UUID、字段 key 对应 Etcd key、字段 value 对应 Etcd value，这样做的一个好处是无需重构数据结构
    1. SET NAMES utf8mb4;
    2. SET FOREIGN_KEY_CHECKS = 0;
    3.  
    4.  
    5. DROP TABLE IF EXISTS `t_redis_msg`;
    6. CREATE TABLE `t_redis_msg` (
    7. `id` varchar(32) NOT NULL,
    8. `key` varchar(255) NOT NULL,
    9. `value` longtext,
    10. PRIMARY KEY (`id`)
    11. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
    12.  
    13.  
    14. SET FOREIGN_KEY_CHECKS = 1;

    所以，如果想平替 Etcd 的事件类型和持久层数据的解决方案需要 MySQL & Redis 结合，接下来直接上代码……

    Redis & MySQL 整合

    application.yml（升级）
    1. spring:
    2.   application:
    3.     name: redis-demo
    4.   datasource:
    5.     username: root
    6.     password: 123456
    7.     url: jdbc:mysql://localhost:3306/db_demo?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai
    8.     driver-class-name: com.mysql.cj.jdbc.Driver
    9.     hikari:
    10.       connection-test-query: SELECT 1
    11.       idle-timeout: 40000
    12.       max-lifetime: 1880000
    13.       connection-timeout: 40000
    14.       minimum-idle: 1
    15.       validation-timeout: 60000
    16.       maximum-pool-size: 20
    RedisMsg.java
    1. import com.baomidou.mybatisplus.annotation.IdType;
    2. import com.baomidou.mybatisplus.annotation.TableField;
    3. import com.baomidou.mybatisplus.annotation.TableId;
    4. import com.baomidou.mybatisplus.annotation.TableName;
    5. import lombok.AllArgsConstructor;
    6. import lombok.Data;
    7. import lombok.NoArgsConstructor;
    8. import lombok.experimental.SuperBuilder;
    9.  
    10. /**
    11.  * @author Lux Sun
    12.  * @date 2021/2/19
    13.  */
    14. @Data
    15. @SuperBuilder
    16. @NoArgsConstructor
    17. @AllArgsConstructor
    18. @TableName(value = "t_redis_msg", autoResultMap = true)
    19. public class RedisMsg {
    20.  
    21.     @TableId(type = IdType.ASSIGN_UUID)
    22.     private String id;
    23.     
    24.     @TableField(value = "`key`")
    25.     private String key;
    26.     
    27.     private String value;
    28. }
    RedisMsgEnum.java
    1. /**
    2.  * @author Lux Sun
    3.  * @date 2022/11/11
    4.  */
    5. public enum RedisMsgEnum {
    6.     PUT("PUT"),
    7.     DEL("DEL");
    8.  
    9.     private String code;
    10.  
    11.     RedisMsgEnum(String code) {
    12.        this.code = code;
    13.     }
    14.  
    15.     public String getCode() {
    16.        return code;
    17.     }
    18.  
    19. }
    RedisMsgService.java
    1. import com.baomidou.mybatisplus.extension.service.IService;
    2. import java.util.List;
    3. import java.util.Map;
    4.  
    5. /**
    6.  * @author Lux Sun
    7.  * @date 2020/6/16
    8.  */
    9. public interface RedisMsgService extends IService {
    10.  
    11.     /**
    12.      * 获取消息
    13.      * @param key
    14.      */
    15.     RedisMsg get(String key);
    16.  
    17.     /**
    18.      * 获取消息列表
    19.      * @param key
    20.      */
    21.     Map map(String key);
    22.  
    23.     /**
    24.      * 获取消息值
    25.      * @param key
    26.      */
    27.     String getValue(String key);
    28.  
    29.     /**
    30.      * 获取消息列表
    31.      * @param key
    32.      */
    33.     List list(String key);
    34.  
    35.     /**
    36.      * 插入消息
    37.      * @param key
    38.      * @param value
    39.      */
    40.     void put(String key, String value);
    41.  
    42.     /**
    43.      * 删除消息
    44.      * @param key
    45.      */
    46.     void del(String key);
    47. }
    RedisMsgServiceImpl.java
    1. import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
    2. import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
    3. import lombok.extern.slf4j.Slf4j;
    4. import org.springframework.stereotype.Service;
    5. import org.springframework.transaction.annotation.Propagation;
    6. import org.springframework.transaction.annotation.Transactional;
    7. import javax.annotation.Resource;
    8. import java.util.List;
    9. import java.util.Map;
    10. import java.util.stream.Collectors;
    11.  
    12. /**
    13.  * @author Lux Sun
    14.  * @date 2020/6/16
    15.  */
    16. @Slf4j
    17. @Service
    18. public class RedisMsgServiceImpl extends ServiceImpl implements RedisMsgService {
    19.  
    20.     @Resource
    21.     private RedisMsgDao redisMsgDao;
    22.  
    23.     @Resource
    24.     private RedisUtil redisUtil;
    25.  
    26.     /**
    27.      * 获取消息
    28.      *
    29.      * @param key
    30.      */
    31.     @Override
    32.     public RedisMsg get(String key) {
    33.         LambdaQueryWrapper lqw = new LambdaQueryWrapper<>();
    34.         lqw.eq(RedisMsg::getKey, key);
    35.         return redisMsgDao.selectOne(lqw);
    36.     }
    37.  
    38.     /**
    39.      * 获取消息列表
    40.      *
    41.      * @param key
    42.      */
    43.     @Override
    44.     public Map map(String key) {
    45.         List redisMsgs = this.list(key);
    46.         return redisMsgs.stream().collect(Collectors.toMap(RedisMsg::getKey, RedisMsg::getValue));
    47.     }
    48.  
    49.     /**
    50.      * 获取消息值
    51.      *
    52.      * @param key
    53.      */
    54.     @Override
    55.     public String getValue(String key) {
    56.         RedisMsg redisMsg = this.get(key);
    57.         return redisMsg.getValue();
    58.     }
    59.  
    60.     /**
    61.      * 获取消息列表
    62.      *
    63.      * @param key
    64.      */
    65.     @Override
    66.     public List list(String key) {
    67.         LambdaQueryWrapper lqw = new LambdaQueryWrapper<>();
    68.         lqw.likeRight(RedisMsg::getKey, key);
    69.         return redisMsgDao.selectList(lqw);
    70.     }
    71.  
    72.     /**
    73.      * 插入消息
    74.      *
    75.      * @param key
    76.      * @param value
    77.      */
    78.     @Override
    79.     public void put(String key, String value) {
    80.         log.info("开始添加 - key: {}，value: {}", key, value);
    81.         LambdaQueryWrapper lqw = new LambdaQueryWrapper<>();
    82.         lqw.eq(RedisMsg::getKey, key);
    83.         this.saveOrUpdate(RedisMsg.builder().key(key).value(value).build(), lqw);
    84.         redisUtil.putMsg(key);
    85.         log.info("添加成功 - key: {}，value: {}", key, value);
    86.     }
    87.  
    88.     /**
    89.      * 删除消息
    90.      *
    91.      * @param key
    92.      */
    93.     @Override
    94.     public void del(String key) {
    95.         log.info("开始删除 - key: {}", key);
    96.         LambdaQueryWrapper lqw = new LambdaQueryWrapper<>();
    97.         lqw.likeRight(RedisMsg::getKey, key);
    98.         redisMsgDao.delete(lqw);
    99.         redisUtil.delMsg(key);
    100.         log.info("删除成功 - key: {}", key);
    101.     }
    102. }
    RedisUtil.java（升级）
    1. import lombok.extern.slf4j.Slf4j;
    2. import org.springframework.data.redis.core.RedisTemplate;
    3. import org.springframework.stereotype.Component;
    4. import javax.annotation.Resource;
    5.  
    6. @Slf4j
    7. @Component
    8. public class RedisUtil {
    9.  
    10.     @Resource
    11.     private RedisTemplate redisTemplate;
    12.  
    13.     /**
    14.      * 消息发送
    15.      * @param topic 主题
    16.      * @param message 消息
    17.      */
    18.     public void publish(String topic, String message) {
    19.         redisTemplate.convertAndSend(topic, message);
    20.     }
    21.  
    22.     /**
    23.      * 消息发送 PUT
    24.      * @param topic 主题
    25.      */
    26.     public void putMsg(String topic) {
    27.         redisTemplate.convertAndSend(topic, RedisMsgEnum.PUT);
    28.     }
    29.  
    30.     /**
    31.      * 消息发送 DELETE
    32.      * @param topic 主题
    33.      */
    34.     public void delMsg(String topic) {
    35.         redisTemplate.convertAndSend(topic, RedisMsgEnum.DEL);
    36.     }
    37. }

    演示 DEMO

    RedisMsgController.java
    1. import org.springframework.web.bind.annotation.PostMapping;
    2. import org.springframework.web.bind.annotation.RequestMapping;
    3. import org.springframework.web.bind.annotation.RequestParam;
    4. import org.springframework.web.bind.annotation.RestController;
    5. import javax.annotation.Resource;
    6.  
    7. /**
    8.  * @author Lux Sun
    9.  * @date 2023/9/12
    10.  */
    11. @RestController
    12. @RequestMapping("/redisMsg")
    13. public class RedisMsgController {
    14.  
    15.     @Resource
    16.     private RedisMsgService redisMsgService;
    17.  
    18.     @PostMapping
    19.     public String publish(@RequestParam String topic, @RequestParam String msg) {
    20.        redisMsgService.put(topic, msg);
    21.        return "发送成功: " + topic + " - " + msg;
    22.     }
    23. }
    RedisMsgReceiver.java
    1. import lombok.extern.slf4j.Slf4j;
    2. import org.springframework.data.redis.connection.Message;
    3. import org.springframework.data.redis.connection.MessageListener;
    4. import org.springframework.data.redis.listener.PatternTopic;
    5. import org.springframework.data.redis.listener.RedisMessageListenerContainer;
    6. import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
    7. import org.springframework.stereotype.Component;
    8. import javax.annotation.PostConstruct;
    9. import javax.annotation.Resource;
    10.  
    11. @Slf4j
    12. @Component
    13. public class RedisMsgReceiver implements MessageListener {
    14.  
    15.     @Resource
    16.     private RedisMsgService redisMsgService;
    17.  
    18.     @Resource
    19.     private RedisMessageListenerContainer container;
    20.  
    21.     @PostConstruct
    22.     public void init() {
    23.         MessageListenerAdapter adapter = new MessageListenerAdapter(this);
    24.         container.addMessageListener(adapter, new PatternTopic("topic3.*"));
    25.     }
    26.  
    27.     @Override
    28.     public void onMessage(Message message, byte[] bytes) {
    29.         String key = new String(message.getChannel());
    30.         String event = new String(message.getBody());
    31.         String value = redisMsgService.getValue(key);
    32.         log.info("Key: {}", key);
    33.         log.info("Event: {}", event);
    34.         log.info("Value: {}", value);
    35.     }
    36. }
    测试
    1. curl --location '127.0.0.1:7077/redisMsg' \
    2. --header 'Content-Type: application/x-www-form-urlencoded' \
    3. --data-urlencode 'topic=topic3.msg' \
    4. --data-urlencode 'msg=我是消息3'
    结果
    1. 2023-11-16 10:24:35.721  INFO 43794 --- [nio-7077-exec-1] c.c.redis.demo.RedisMsgServiceImpl       : 开始添加 - key: topic3.msg，value: 我是消息3
    2. 2023-11-16 10:24:35.935  INFO 43794 --- [nio-7077-exec-1] c.c.redis.demo.RedisMsgServiceImpl       : 添加成功 - key: topic3.msg，value: 我是消息3
    3. 2023-11-16 10:24:35.950  INFO 43794 --- [    container-2] c.xxx.redis.demo.RedisMsgReceiver  : Key: topic3.msg
    4. 2023-11-16 10:24:35.950  INFO 43794 --- [    container-2] c.xxx.redis.demo.RedisMsgReceiver  : Event: "PUT"
    5. 2023-11-16 10:24:35.950  INFO 43794 --- [    container-2] c.xxx.redis.demo.RedisMsgReceiver  : Value: 我是消息3

  • 相关阅读:
    magic API构建和基础实现
    [RK3588-Android12] 双HDMI+喇叭Speak同音问题
    叛乱沙漠风暴server安装 ubuntu 22.04
    Qt之Windows Server 2012 R2不支持openssl
    IE让我首次遭受了社会的毒打
    Tkinter常用功能示例（一）
    C++数据结构补充（线性表及其顺序存储结构）
    lsblk 硬盘属性查看
    vue Router从入门到精通
    牛客day 8
  • 原文地址：https://blog.csdn.net/Dream_Weave/article/details/134436221