springboot集成Redisson做分布式消息队列

这里演示Redisson做分布式消息队列。首先引入 Redisson依赖，官方github

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>3.16.0</version>
</dependency>

首先创建一个自定义注解RedissonTopic.java，用于指定消息的路由key

package com.zyq.annotation;

import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import java.util.concurrent.TimeUnit;

/** Redissson消息队列注解
 * author xiaochi
 * date 2024/10/23
 */
@Inherited
@Documented
@Target({ElementType.TYPE,ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
public @interface RedissonTopic {

    /**
     * topic名称
     * @return
     */
    String key();

    /**
     * 是否队列发送消息
     * @return
     */
    boolean queue() default false;

    /**
     * 队列容量
     * @return
     */
    int queueSize() default 100;

    /** queue为true时生效
     * 延迟发送时间(大于0默认延迟，延迟队列可设置大于0)
     * @return
     */
    int delayTime() default 0;

    /** queue为true时生效
     * 时间单位(默认毫秒)
     * @return
     */
    TimeUnit timeUnit() default TimeUnit.MILLISECONDS;
}

继续创建消息监听器 RedissonTopicMessageListener.java，具体内容如下：

package com.zyq.listener;

/** Redisson消息监听接口
 * author xiaochi
 * date 2024/10/23
 */
public interface RedissonTopicMessageListener{

    /**
     * 接收的消息处理
     * @param message
     */
    void message(Object message);

    /**
     * 发送失败(队列已满时会调用)
     * @param message
     */
    void sendFail(Object message);

    /**
     * 异常
     * @param ex
     */
    void exception(Exception ex);
}

接下来就是最重要的Redisson配置，内容如下：

/**
 * Redisson 配置
 * @return
 */
@Bean(destroyMethod="shutdown")
public RedissonClient redissonClient(ConfigurableApplicationContext applicationContext){
    Config config = new Config();
    config.useSingleServer().setPassword("123456")
            .setDatabase(0)
            .setConnectionPoolSize(24) // 连接池大小，默认64
            .setConnectionMinimumIdleSize(3) // 最小空闲连接数，默认32
            .setRetryAttempts(3) // 命令失败重试次数 3
            .setRetryInterval(1500) // 命令重试发送时间间隔(毫秒) 默认1500
            .setTimeout(10000) // 命令等待超时(毫秒) 默认10000
            .setConnectTimeout(10000) // 连接空闲超时(毫秒) 默认10000
            .setIdleConnectionTimeout(10000) // 连接空闲超时(毫秒) 默认10000
            .setSubscriptionConnectionMinimumIdleSize(3) // 发布和订阅连接的最小空闲连接数
            .setSubscriptionConnectionPoolSize(50) // 发布和订阅连接池大小 默认50
            .setDnsMonitoringInterval(10000) // DNS监测时间间隔(毫秒)，默认5000
            .setAddress("redis://127.0.0.1:6379");
    //config.setThreads(Runtime.getRuntime().availableProcessors());// 默认 16
    RedissonClient redissonClient = Redisson.create(config);
    StringBuilder msg = new StringBuilder();
    msg.append("redisson topic register[");
    String[] beanNames = applicationContext.getBeanNamesForType(RedissonTopicMessageListener.class);
    for (String beanName : beanNames) {
        RedissonTopicMessageListener topicMessageListener = applicationContext.getBean(beanName, RedissonTopicMessageListener.class);
        if (topicMessageListener.getClass().isAnnotationPresent(RedissonTopic.class)){
            RedissonTopic redissonTopic = topicMessageListener.getClass().getAnnotation(RedissonTopic.class);
            if (redissonTopic.queue()){
                RBoundedBlockingQueue<Object> boundedBlockingQueue = redissonClient.getBoundedBlockingQueue(redissonTopic.key());
                boundedBlockingQueue.trySetCapacity(redissonTopic.queueSize());
                RDelayedQueue<Object> delayedQueue = null;
                if (0 != redissonTopic.delayTime()){
                    delayedQueue = redissonClient.getDelayedQueue(boundedBlockingQueue);
                }
                RTopic topic = redissonClient.getTopic(redissonTopic.key());
                RDelayedQueue<Object> finalDelayedQueue = delayedQueue;
                topic.addListener(Object.class, (channel, message) -> {
                    if (finalDelayedQueue != null){
                        try {
                            finalDelayedQueue.offer(message,redissonTopic.delayTime(), redissonTopic.timeUnit());
                        }catch (Exception e){
                            topicMessageListener.exception(e);
                        }
                    }else {
                        try {
                            if (!boundedBlockingQueue.offer(message)){
                                topicMessageListener.sendFail(message);
                            }
                        }catch (Exception e){
                            topicMessageListener.exception(e);
                        }
                    }
                });
                // 为了不阻塞主线程，放在新线程中运行
                AsyncUtil.run(() -> {
                    while (!Thread.currentThread().isInterrupted() && !redissonClient.isShutdown()){
                        try {
                            Object take = boundedBlockingQueue.take();
                            if (!"".equals(take)){
                                topicMessageListener.message(take);
                            }
                        } catch (Exception e) {
                            topicMessageListener.exception(e);
                            log.info("redisson.{}队列监测异常,{}",redissonTopic.key(),e);
                        }
                    }
                    if (Thread.currentThread().isInterrupted() || redissonClient.isShutdown()){
                        log.info("redisson service shutdown");
                    }
                });
            }else {
                RTopic topic = redissonClient.getTopic(redissonTopic.key());
                topic.addListener(Object.class, (channel,message) -> {
                    try {
                        topicMessageListener.message(message);
                    }catch (Exception e){
                        topicMessageListener.exception(e);
                    }
                });
            }
            msg.append(redissonTopic.key()).append(".");
        }
    }
    msg.append("]").append("finish.");
    log.info(msg.toString());
    return redissonClient;
}

到此基本就完成了，接下来就是创建消息监听类进行消费消息了TopicMessageListener.java 去实现消息监听器接口RedissonTopicMessageListener.java

package com.zyq.listener;

import com.zyq.annotation.RedissonTopic;
import org.springframework.stereotype.Component;

/** 消息监听类
 * author xiaochi
 * date 2024/10/23
 */
@Component
@RedissonTopic(key = "testTopic",queue = true,delayTime = 5000)
public class TopicMessageListener implements RedissonTopicMessageListener {

    @Override
    public void message(Object message) {
        System.out.println("testTopic监听器延迟队列收到消息," + message);
    }

    @Override
    public void sendFail(Object message) {
        System.out.println("延迟队列 TopicMessageListener testTopic消息发送失败");
    }

    @Override
    public void exception(Exception ex) {
        System.out.println("延迟队列 TopicMessageListener testTopic消息异常,{}",ex);
    }
}

现在可以起2个springboot项目进行消息交流了。封装一个消息发送工具 RedissonMessageUtil.java，内容如下：

package com.demo3.util;

import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/** Redisson消息发送工具
 * author xiaochi
 * date 2024/10/24
 */
@Component
public class RedissonMessageUtil {

    private static RedissonClient redissonClient;

    @Autowired
    public void setRedissonClient(RedissonClient redissonClient) {
        RedissonMessageUtil.redissonClient = redissonClient;
    }

    /**
     * 发送消息
     * @param key
     * @param message
     * @return 返回接收消息的客户端数量
     */
    public static long send(String key,Object message){
        RTopic topic = redissonClient.getTopic(key);
        return topic.publish(message);
    }
}

在业务处使用

RedissonMessageUtil.send("testTopic","你好啊");

再去看控制台，已经打印接收到的消息了，到此完成。

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

推荐阅读更多精彩内容