脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 软件编程 → java → SpringBoot Redisson消息队列

SpringBoot集成Redisson实现消息队列的示例代码

 更新时间：2024年10月10日 09:12:37   作者：入秋的大橘  
本文介绍了如何在SpringBoot中通过集成Redisson来实现消息队列的功能,包括RedisQueue、RedisQueueInit、RedisQueueListener、RedisQueueService等相关组件的实现和测试,感兴趣的可以了解一下

包含组件内容

  • RedisQueue：消息队列监听标识
  • RedisQueueInit：Redis队列监听器
  • RedisQueueListener：Redis消息队列监听实现
  • RedisQueueService：Redis消息队列服务工具

代码实现

RedisQueue

import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

/**
 * Redis消息队列注解
 */
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
public @interface RedisQueue {
    /**
     * 队列名
     */
    String value();
}

RedisQueueInit

import jakarta.annotation.Resource;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
import org.redisson.RedissonShutdownException;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

/**
 * 初始化Redis队列监听器
 *
 * @author 十八
 * @createTime 2024-09-09 22:49
 */
@Slf4j
@Component
public class RedisQueueInit implements ApplicationContextAware {

    public static final String REDIS_QUEUE_PREFIX = "redis-queue";
    final AtomicBoolean shutdownRequested = new AtomicBoolean(false);
    @Resource
    private RedissonClient redissonClient;
    private ExecutorService executorService;

    public static String buildQueueName(String queueName) {
        return REDIS_QUEUE_PREFIX + ":" + queueName;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        Map<String, RedisQueueListener> queueListeners = applicationContext.getBeansOfType(RedisQueueListener.class);
        if (!queueListeners.isEmpty()) {
            executorService = createThreadPool();
            for (Map.Entry<String, RedisQueueListener> entry : queueListeners.entrySet()) {
                RedisQueue redisQueue = entry.getValue().getClass().getAnnotation(RedisQueue.class);
                if (redisQueue != null) {
                    String queueName = redisQueue.value();
                    executorService.submit(() -> listenQueue(queueName, entry.getValue()));
                }
            }
        }
    }

    private ExecutorService createThreadPool() {
        return new ThreadPoolExecutor(
                Runtime.getRuntime().availableProcessors() * 2,
                Runtime.getRuntime().availableProcessors() * 4,
                60L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(100),
                new NamedThreadFactory(REDIS_QUEUE_PREFIX),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
    }

    private void listenQueue(String queueName, RedisQueueListener redisQueueListener) {
        queueName = buildQueueName(queueName);
        RBlockingQueue<?> blockingQueue = redissonClient.getBlockingQueue(queueName);
        log.info("Redis队列监听开启: {}", queueName);
        while (!shutdownRequested.get() && !redissonClient.isShutdown()) {
            try {
                Object message = blockingQueue.take();
                executorService.submit(() -> redisQueueListener.consume(message));
            } catch (RedissonShutdownException e) {
                log.info("Redis连接关闭，停止监听队列: {}", queueName);
                break;
            } catch (Exception e) {
                log.error("监听队列异常: {}", queueName, e);
            }
        }
    }

    public void shutdown() {
        if (executorService != null) {
            executorService.shutdown();
            try {
                if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
                    executorService.shutdownNow();
                }
            } catch (InterruptedException ex) {
                executorService.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
        shutdownRequested.set(true);
        if (redissonClient != null && !redissonClient.isShuttingDown()) {
            redissonClient.shutdown();
        }
    }

    private static class NamedThreadFactory implements ThreadFactory {
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        public NamedThreadFactory(String prefix) {
            this.namePrefix = prefix;
        }

        @Override
        public Thread newThread(@NotNull Runnable r) {
            return new Thread(r, namePrefix + "-" + threadNumber.getAndIncrement());
        }
    }

}

RedisQueueListener

/**
 * Redis消息队列监听实现
 *
 * @author 十八
 * @createTime 2024-09-09 22:51
 */
public interface RedisQueueListener<T> {

    /**
     * 队列消费方法
     *
     * @param content 消息内容
     */
    void consume(T content);
}

RedisQueueService

import jakarta.annotation.Resource;
import java.util.concurrent.TimeUnit;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Component;

/**
 * Redis 消息队列服务
 *
 * @author 十八
 * @createTime 2024-09-09 22:52
 */
@Component
public class RedisQueueService {

    @Resource
    private RedissonClient redissonClient;

    /**
     * 添加队列
     *
     * @param queueName 队列名称
     * @param content   消息
     * @param <T>       泛型
     */
    public <T> void send(String queueName, T content) {
        RBlockingQueue<T> blockingQueue = redissonClient.getBlockingQueue(RedisQueueInit.buildQueueName(queueName));
        blockingQueue.add(content);
    }

    /**
     * 添加延迟队列
     *
     * @param queueName 队列名称
     * @param content   消息类型
     * @param delay     延迟时间
     * @param timeUnit  单位
     * @param <T>       泛型
     */
    public <T> void sendDelay(String queueName, T content, long delay, TimeUnit timeUnit) {
        RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(RedisQueueInit.buildQueueName(queueName));
        RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
        delayedQueue.offer(content, delay, timeUnit);
    }

    /**
     * 发送延迟队列消息(单位毫秒)
     *
     * @param queueName 队列名称
     * @param content   消息类型
     * @param delay     延迟时间
     * @param <T>       泛型
     */
    public <T> void sendDelay(String queueName, T content, long delay) {
        RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(RedisQueueInit.buildQueueName(queueName));
        RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
        delayedQueue.offer(content, delay, TimeUnit.MILLISECONDS);
    }
}

测试

创建监听对象

import cn.yiyanc.infrastructure.redis.annotation.RedisQueue;
import cn.yiyanc.infrastructure.redis.queue.RedisQueueListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

/**
 * @author 十八
 * @createTime 2024-09-10 00:09
 */
@Slf4j
@Component
@RedisQueue("test")
public class TestListener implements RedisQueueListener<String> {
    @Override
    public void invoke(String content) {
        log.info("队列消息接收 >>> {}", content);
    }
}

测试用例

import jakarta.annotation.Resource;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author 十八
 * @createTime 2024-09-10 00:11
 */
@RestController
@RequestMapping("queue")
public class QueueController {

    @Resource
    private RedisQueueService redisQueueService;

    @PostMapping("send")
    public void send(String message) {
        redisQueueService.send("test", message);
        redisQueueService.sendDelay("test", "delay messaege -> " + message, 1000);
    }

}

测试结果

到此这篇关于SpringBoot集成Redisson实现消息队列的示例代码的文章就介绍到这了,更多相关SpringBoot Redisson消息队列内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

相关文章

  • Java8中新特性Optional、接口中默认方法和静态方法详解

    Java8中新特性Optional、接口中默认方法和静态方法详解

    Java 8 已经发布很久了，很多报道表明Java 8 是一次重大的版本升级。下面这篇文章主要给大家介绍了关于Java8中新特性Optional、接口中默认方法和静态方法的相关资料，文中通过示例代码介绍的非常详细，需要的朋友可以参考下。
    2017-12-12
  • SpringBoot中Tomcat配置的示例代码

    SpringBoot中Tomcat配置的示例代码

    本文分享了在SpringBoot项目中配置Tomcat的一些心得和经验,包括Tomcat版本选择、调整配置参数、自定义连接器、监控和日志管理等方面,通过这些配置,可以有效提升应用的性能、响应速度和并发处理能力
    2024-11-11
  • java解析json数组方式

    java解析json数组方式

    这篇文章主要介绍了java解析json数组方式，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
    2023-06-06
  • maven package 打包报错 Failed to execute goal的解决

    maven package 打包报错 Failed to execute goal的解决

    这篇文章主要介绍了maven package 打包报错 Failed to execute goal的解决，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
    2020-11-11
  • IDEA使用Tomcat运行web项目教程分享

    IDEA使用Tomcat运行web项目教程分享

    在非Spring Boot项目中运行Nacos示例,需要手动配置Tomcat容器,本文介绍了如何在IDEA中配置Tomcat,并详细解决了配置过程中可能遇到的异常情况,步骤包括修改IDEA项目结构、添加Web模块、配置Artifacts和Tomcat Server
    2024-10-10
  • spring学习之参数传递与检验详解

    spring学习之参数传递与检验详解

    这篇文章主要给大家介绍了关于spring参数传递与检验的相关资料，文中通过示例代码介绍的非常详细，对大家的学习或者工作能带来一定的帮助，需要的朋友们下面跟着小编来一起学习学习吧。
    2017-07-07
  • 详解用Kotlin写一个基于Spring Boot的RESTful服务

    详解用Kotlin写一个基于Spring Boot的RESTful服务

    这篇文章主要介绍了详解用Kotlin写一个基于Spring Boot的RESTful服务 ，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
    2017-05-05
  • java实现对对碰小游戏

    java实现对对碰小游戏

    这篇文章主要为大家详细介绍了java实现对对碰小游戏，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
    2019-12-12
  • 简要分析Java的Hibernate框架中的自定义类型

    简要分析Java的Hibernate框架中的自定义类型

    这篇文章主要介绍了Java的Hibernate框架中的自定义类型,Hibernate是Java的SSH三大web开发框架之一,需要的朋友可以参考下
    2016-01-01
  • MyBatis-Plus 复杂查询Lambda+Wrapper 多条件功能实现

    MyBatis-Plus 复杂查询Lambda+Wrapper 多条件功能实现

    通过本文的介绍,我们深入了解了MyBatis-Plus中Lambda+Wrapper的强大功能,它不仅极大地简化了SQL的编写,还提高了代码的安全性和可维护性,感兴趣的朋友跟随小编一起看看吧
    2026-01-01

最新评论