相关推荐recommended
Springboot中使用redisson + 自定义注解优雅的实现消息的发布订阅
作者:mmseoamin日期:2024-04-27

前言

在一些小型场景下,使用mq中间件可能会为原有项目增加不少维护成本,使用redisson实现消息的收发是个不错的选择

什么是redisson?

官网:Redisson: Easy Redis Java client with features of In-Memory Data Grid

Redisson是一个基于Redis的Java驻留内存数据网格(In-Memory Data Grid)和分布式锁框架。它提供了一系列的分布式Java对象和服务,可以帮助开发者更方便地使用Redis作为数据存储和分布式锁的解决方案。

Springboot中使用redisson + 自定义注解优雅的实现消息的发布订阅,第1张

Redisson的主要功能包括:

  1. 分布式集合:Redisson提供了分布式的Set、List、Queue、Deque等集合,可以在分布式环境下进行操作,实现数据共享和协作。

  2. 分布式映射:Redisson提供了分布式的Map、Multimap、ConcurrentMap等映射结构,可以在分布式环境下进行操作,实现数据的存储和共享。

  3. 分布式锁:Redisson提供了可重入锁、公平锁、读写锁等分布式锁,可以在分布式环境下实现资源的互斥访问,保证数据的一致性和并发安全。

  4. 分布式对象:Redisson提供了分布式的AtomicLong、CountDownLatch、Semaphore等对象,可以在分布式环境下实现共享状态和协作操作。

Redisson的使用场景包括但不限于:

  1. 分布式缓存:Redisson可以作为分布式缓存的解决方案,将数据存储在Redis中,提高数据的读取速度和系统的性能。

  2. 分布式锁:Redisson可以用于实现分布式锁,保证在分布式环境下对共享资源的互斥访问,避免数据的并发冲突。

  3. 分布式任务调度:Redisson可以用于实现分布式任务调度,将任务分发到不同的节点上执行,提高系统的并发处理能力。

Redisson的优点包括:

  1. 简单易用:Redisson提供了简洁的API和丰富的文档,使得使用者可以快速上手。

  2. 高性能:Redisson利用Redis的高性能特性,可以实现快速的数据读写和并发操作。

  3. 可靠性:Redisson提供了分布式锁和数据持久化等机制,保证数据的一致性和可靠性。

Redisson的缺点包括:

  1. 依赖于Redis:Redisson需要依赖Redis作为数据存储和分布式锁的后端,需要确保Redis的可用性和性能。

  2. 部署复杂性:Redisson的部署和配置相对复杂,需要对Redis和Redisson的相关参数进行调优和配置。

redisson发布订阅的基本使用

// 创建Redisson客户端
RedissonClient redisson = Redisson.create();
// 获取RTopic对象
RTopic topic = redisson.getTopic("myTopic");
// 发布消息
topic.publish("Hello, Redisson!");
// 添加监听器
topic.addListener(String.class, (channel, msg) -> {
    System.out.println("Received message: " + msg);
});
// 关闭Redisson客户端
redisson.shutdown();
  • 发布和订阅,是我们需要对同一个 Topic 进行发布和监听操作。但这个操作的代码是一种手动编码,但在我们实际使用中,如果所有的都是手动编码,一个是非常麻烦,再有一个是非常累人。
  • 通过自定义注解,来完成动态监听和将对象动态注入到 Spring 容器中,让需要注入的属性,可以被动态注入。

    结合自定义注解优雅实现

    导入依赖

            
                org.redisson
                redisson-spring-boot-starter
                3.14.1
            

    编写redisson初始化配置

    redis:
      sdk:
        config:
          host: localhost
          port: 6379
          pool-size: 10
          min-idle-size: 5
          idle-timeout: 30000
          connect-timeout: 5000
          retry-attempts: 3
          retry-interval: 1000
          ping-interval: 60000
          keep-alive: true

    编写redisson配置类

    /**
     * @description redis连接配置
     * @create 2023/12/17 20:48:13
     */
    @Data
    @ConfigurationProperties(prefix = "redis.sdk.config", ignoreInvalidFields = true)
    public class RedissonCientConfigProperties {
        /** host:ip */
        private String host;
        /** 端口 */
        private int port;
        /** 账密 */
        private String password;
        /** 设置连接池的大小,默认为64 */
        private int poolSize = 64;
        /** 设置连接池的最小空闲连接数,默认为10 */
        private int minIdleSize = 10;
        /** 设置连接的最大空闲时间(单位:毫秒),超过该时间的空闲连接将被关闭,默认为10000 */
        private int idleTimeout = 10000;
        /** 设置连接超时时间(单位:毫秒),默认为10000 */
        private int connectTimeout = 10000;
        /** 设置连接重试次数,默认为3 */
        private int retryAttempts = 3;
        /** 设置连接重试的间隔时间(单位:毫秒),默认为1000 */
        private int retryInterval = 1000;
        /** 设置定期检查连接是否可用的时间间隔(单位:毫秒),默认为0,表示不进行定期检查 */
        private int pingInterval = 0;
        /** 设置是否保持长连接,默认为true */
        private boolean keepAlive = true;
    }

    编写自定义主题注解,用来指定主题

    /**
     * @description redisson 消息主题注解
     * @create 2023/12/17 21:56:10
     */
    @Retention(RetentionPolicy.RUNTIME)
    @Target({ElementType.TYPE})
    @Documented
    public @interface RedisTopic {
        // 主题名称
        String topic() default "";
    }

    初始化Redisson客户端,注册主题监听

    /**
     * @description redis客户端
     * @create 2023/12/17 20:49:41
     */
    @Configuration
    @EnableConfigurationProperties(RedisCientConfigProperties.class)
    public class RedisClientConfig {
         public RedissonClient redissonClient(ConfigurableApplicationContext applicationContext, RedisCientConfigProperties properties) {
            Config config = new Config();
            config.useSingleServer()
                    .setAddress("redis://" + properties.getHost() + ":" + properties.getPort())
                   .setPassword(properties.getPassword())
                    .setConnectionPoolSize(properties.getPoolSize())
                    .setConnectionMinimumIdleSize(properties.getMinIdleSize())
                    .setIdleConnectionTimeout(properties.getIdleTimeout())
                    .setConnectTimeout(properties.getConnectTimeout())
                    .setRetryAttempts(properties.getRetryAttempts())
                    .setRetryInterval(properties.getRetryInterval())
                    .setPingConnectionInterval(properties.getPingInterval())
                    .setKeepAlive(properties.isKeepAlive())
            ;
            RedissonClient redissonClient = Redisson.create(config);
            // 注册消息发布订阅主题Topic
            // 找到所有实现了Redisson中MessageListener接口的bean名字
            String[] beanNamesForType = applicationContext.getBeanNamesForType(MessageListener.class);
            for (String beanName : beanNamesForType) {
                // 通过bean名字获取到监听bean
                MessageListener bean = applicationContext.getBean(beanName, MessageListener.class);
                Class beanClass = bean.getClass();
                // 如果bean的注解里包含我们的自定义注解RedisTopic.class,则以RedisTopic注解的值作为name将该bean注册到bean工厂,方便在别处注入
                if (beanClass.isAnnotationPresent(RedisTopic.class)) {
                    RedisTopic redisTopic = beanClass.getAnnotation(RedisTopic.class);
                    RTopic topic = redissonClient.getTopic(redisTopic.topic());
                    topic.addListener(String.class, bean);
                    ConfigurableListableBeanFactory beanFactory = applicationContext.getBeanFactory();
                    beanFactory.registerSingleton(redisTopic.topic(), topic);
                }
            }
            return redissonClient;
        }
    }

    在监听器上声明注解,如下服务即订阅了testRedisTopic02主题

    @Slf4j
    @Service
    @RedisTopic(topic = "testRedisTopic02")
    public class RedisTopicListener02 implements MessageListener {
        @Override
        public void onMessage(CharSequence channel, String msg) {
            log.info("02-监听消息(Redis 发布/订阅): {}", msg);
        }
    }

    使用,发布消息

    @Slf4j
    @Repository
    public class OrderRepository implements IOrderRepository {
        @Resource(name = "testRedisTopic02")
        private RTopic testRedisTopic02;
        @Resource(name = "testRedisTopic03")
        private RTopic testRedisTopic03;
        @Override
        public String createOrder(OrderAggregate orderAggregate) {
        
            // 向  testRedisTopic02 发布消息
            testRedisTopic02.publish(JSON.toJSONString(orderEntity));
            // 向  testRedisTopic03 发布消息
            testRedisTopic03.publish(JSON.toJSONString(orderEntity));
            return orderId;
        }    
    }