在一些小型场景下,使用mq中间件可能会为原有项目增加不少维护成本,使用redisson实现消息的收发是个不错的选择
官网:Redisson: Easy Redis Java client with features of In-Memory Data Grid
Redisson是一个基于Redis的Java驻留内存数据网格(In-Memory Data Grid)和分布式锁框架。它提供了一系列的分布式Java对象和服务,可以帮助开发者更方便地使用Redis作为数据存储和分布式锁的解决方案。
分布式集合:Redisson提供了分布式的Set、List、Queue、Deque等集合,可以在分布式环境下进行操作,实现数据共享和协作。
分布式映射:Redisson提供了分布式的Map、Multimap、ConcurrentMap等映射结构,可以在分布式环境下进行操作,实现数据的存储和共享。
分布式锁:Redisson提供了可重入锁、公平锁、读写锁等分布式锁,可以在分布式环境下实现资源的互斥访问,保证数据的一致性和并发安全。
分布式对象:Redisson提供了分布式的AtomicLong、CountDownLatch、Semaphore等对象,可以在分布式环境下实现共享状态和协作操作。
分布式缓存:Redisson可以作为分布式缓存的解决方案,将数据存储在Redis中,提高数据的读取速度和系统的性能。
分布式锁:Redisson可以用于实现分布式锁,保证在分布式环境下对共享资源的互斥访问,避免数据的并发冲突。
分布式任务调度:Redisson可以用于实现分布式任务调度,将任务分发到不同的节点上执行,提高系统的并发处理能力。
简单易用:Redisson提供了简洁的API和丰富的文档,使得使用者可以快速上手。
高性能:Redisson利用Redis的高性能特性,可以实现快速的数据读写和并发操作。
可靠性:Redisson提供了分布式锁和数据持久化等机制,保证数据的一致性和可靠性。
依赖于Redis:Redisson需要依赖Redis作为数据存储和分布式锁的后端,需要确保Redis的可用性和性能。
部署复杂性:Redisson的部署和配置相对复杂,需要对Redis和Redisson的相关参数进行调优和配置。
// 创建Redisson客户端 RedissonClient redisson = Redisson.create(); // 获取RTopic对象 RTopictopic = redisson.getTopic("myTopic"); // 发布消息 topic.publish("Hello, Redisson!"); // 添加监听器 topic.addListener(String.class, (channel, msg) -> { System.out.println("Received message: " + msg); }); // 关闭Redisson客户端 redisson.shutdown();
org.redisson redisson-spring-boot-starter3.14.1
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
/** * @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 ""; }
/** * @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; } }
@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; } }