前言
在一些小型场景下，使用mq中间件可能会为原有项目增加不少维护成本，使用redisson实现消息的收发是个不错的选择
什么是redisson？
官网：Redisson: Easy Redis Java client with features of In-Memory Data Grid
Redisson是一个基于Redis的Java驻留内存数据网格（In-Memory Data Grid）和分布式锁框架。它提供了一系列的分布式Java对象和服务，可以帮助开发者更方便地使用Redis作为数据存储和分布式锁的解决方案。
Redisson的主要功能包括：
- 分布式集合：Redisson提供了分布式的Set、List、Queue、Deque等集合，可以在分布式环境下进行操作，实现数据共享和协作。
- 分布式映射：Redisson提供了分布式的Map、Multimap、ConcurrentMap等映射结构，可以在分布式环境下进行操作，实现数据的存储和共享。
- 分布式锁：Redisson提供了可重入锁、公平锁、读写锁等分布式锁，可以在分布式环境下实现资源的互斥访问，保证数据的一致性和并发安全。
- 分布式对象：Redisson提供了分布式的AtomicLong、CountDownLatch、Semaphore等对象，可以在分布式环境下实现共享状态和协作操作。
Redisson的使用场景包括但不限于：
分布式缓存：Redisson可以作为分布式缓存的解决方案，将数据存储在Redis中，提高数据的读取速度和系统的性能。
分布式锁：Redisson可以用于实现分布式锁，保证在分布式环境下对共享资源的互斥访问，避免数据的并发冲突。
分布式任务调度：Redisson可以用于实现分布式任务调度，将任务分发到不同的节点上执行，提高系统的并发处理能力。
Redisson的优点包括：
简单易用：Redisson提供了简洁的API和丰富的文档，使得使用者可以快速上手。
高性能：Redisson利用Redis的高性能特性，可以实现快速的数据读写和并发操作。
可靠性：Redisson提供了分布式锁和数据持久化等机制，保证数据的一致性和可靠性。
Redisson的缺点包括：
依赖于Redis：Redisson需要依赖Redis作为数据存储和分布式锁的后端，需要确保Redis的可用性和性能。
部署复杂性：Redisson的部署和配置相对复杂，需要对Redis和Redisson的相关参数进行调优和配置。
redisson发布订阅的基本使用
// 创建Redisson客户端
RedissonClient redisson = Redisson.create();
// 获取RTopic对象
RTopic<String> 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 容器中，让需要注入的属性，可以被动态注入。
结合自定义注解优雅实现
导入依赖
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
<version>3.14.1</version>
</dependency>
编写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<? extends MessageListener> 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<String> {
@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;
}
}
