这里演示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","你好啊");
再去看控制台，已经打印接收到的消息了，到此完成。
