业务服务:redisson

文章目录

  • 前言
  • 一、配置
    • 1. 添加依赖
    • 2. 配置文件/类
    • 3. 注入redission
    • 3. 封装工具类
  • 二、应用
    • 1. RedisUtils工具类的基本使用
  • 三、队列
    • 1. 工具类
    • 2. 普通队列
    • 3. 有界队列(限制数据量)
    • 4. 延迟队列(延迟获取数据)
    • 5. 优先队列(数据可插队)


前言

redission是一个开源的java redis的客户端,在其基础上进行了进一步扩展。这些扩展极大地丰富了Redis的应用场景,尤其是在构建分布式系统时。


一、配置

1. 添加依赖

<!--redisson-->
<dependency><groupId>org.redisson</groupId><artifactId>redisson-spring-boot-starter</artifactId><version>${redisson.version}</version><exclusions><exclusion><groupId>org.redisson</groupId><artifactId>redisson-spring-data-30</artifactId></exclusion></exclusions>
</dependency>
<dependency><groupId>org.redisson</groupId><artifactId>redisson-spring-data-27</artifactId><version>${redisson.version}</version>
</dependency>
<dependency><groupId>com.baomidou</groupId><artifactId>lock4j-redisson-spring-boot-starter</artifactId><version>${lock4j.version}</version>
</dependency>

在这里插入图片描述

2. 配置文件/类

spring:redis:# 地址host: localhost# 端口,默认为6379port: 6379# 数据库索引database: 0# 密码(如没有密码请注释掉)password: asd60787533# 连接超时时间timeout: 10s# 是否开启sslssl: falseredisson:# redis key前缀keyPrefix: demo# 线程池数量threads: 4# Netty线程池数量nettyThreads: 8# 单节点配置singleServerConfig:# 客户端名称clientName: demo# 最小空闲连接数connectionMinimumIdleSize: 8# 连接池大小connectionPoolSize: 32# 连接空闲超时,单位:毫秒idleConnectionTimeout: 10000# 命令等待超时,单位:毫秒timeout: 3000# 发布和订阅连接池大小subscriptionConnectionPoolSize: 50
@Data
@Component
@ConfigurationProperties(prefix = "redisson")
public class RedissonProperties {/*** redis缓存key前缀*/private String keyPrefix;/*** 线程池数量,默认值 = 当前处理核数量 * 2*/private int threads;/*** Netty线程池数量,默认值 = 当前处理核数量 * 2*/private int nettyThreads;/*** 单机服务配置*/private SingleServerConfig singleServerConfig;/*** 集群服务配置*/private ClusterServersConfig clusterServersConfig;@Data@NoArgsConstructorpublic static class SingleServerConfig {/*** 客户端名称*/private String clientName;/*** 最小空闲连接数*/private int connectionMinimumIdleSize;/*** 连接池大小*/private int connectionPoolSize;/*** 连接空闲超时,单位:毫秒*/private int idleConnectionTimeout;/*** 命令等待超时,单位:毫秒*/private int timeout;/*** 发布和订阅连接池大小*/private int subscriptionConnectionPoolSize;}@Data@NoArgsConstructorpublic static class ClusterServersConfig {/*** 客户端名称*/private String clientName;/*** master最小空闲连接数*/private int masterConnectionMinimumIdleSize;/*** master连接池大小*/private int masterConnectionPoolSize;/*** slave最小空闲连接数*/private int slaveConnectionMinimumIdleSize;/*** slave连接池大小*/private int slaveConnectionPoolSize;/*** 连接空闲超时,单位:毫秒*/private int idleConnectionTimeout;/*** 命令等待超时,单位:毫秒*/private int timeout;/*** 发布和订阅连接池大小*/private int subscriptionConnectionPoolSize;/*** 读取模式*/private ReadMode readMode;/*** 订阅模式*/private SubscriptionMode subscriptionMode;}}

3. 注入redission

@Slf4j
@Configuration
@EnableCaching
@EnableConfigurationProperties(RedissonProperties.class)
public class RedisConfig {@Autowiredprivate RedissonProperties redissonProperties;@Autowiredprivate ObjectMapper objectMapper;@Beanpublic RedissonAutoConfigurationCustomizer redissonCustomizer() {return config -> {config.setThreads(redissonProperties.getThreads()).setNettyThreads(redissonProperties.getNettyThreads()).setCodec(new JsonJacksonCodec(objectMapper));RedissonProperties.SingleServerConfig singleServerConfig = redissonProperties.getSingleServerConfig();if (ObjectUtil.isNotNull(singleServerConfig)) {// 使用单机模式config.useSingleServer()//设置redis key前缀.setNameMapper(new KeyPrefixHandler(redissonProperties.getKeyPrefix())).setTimeout(singleServerConfig.getTimeout()).setClientName(singleServerConfig.getClientName()).setIdleConnectionTimeout(singleServerConfig.getIdleConnectionTimeout()).setSubscriptionConnectionPoolSize(singleServerConfig.getSubscriptionConnectionPoolSize()).setConnectionMinimumIdleSize(singleServerConfig.getConnectionMinimumIdleSize()).setConnectionPoolSize(singleServerConfig.getConnectionPoolSize());}// 集群配置方式 参考下方注释RedissonProperties.ClusterServersConfig clusterServersConfig = redissonProperties.getClusterServersConfig();if (ObjectUtil.isNotNull(clusterServersConfig)) {config.useClusterServers()//设置redis key前缀.setNameMapper(new KeyPrefixHandler(redissonProperties.getKeyPrefix())).setTimeout(clusterServersConfig.getTimeout()).setClientName(clusterServersConfig.getClientName()).setIdleConnectionTimeout(clusterServersConfig.getIdleConnectionTimeout()).setSubscriptionConnectionPoolSize(clusterServersConfig.getSubscriptionConnectionPoolSize()).setMasterConnectionMinimumIdleSize(clusterServersConfig.getMasterConnectionMinimumIdleSize()).setMasterConnectionPoolSize(clusterServersConfig.getMasterConnectionPoolSize()).setSlaveConnectionMinimumIdleSize(clusterServersConfig.getSlaveConnectionMinimumIdleSize()).setSlaveConnectionPoolSize(clusterServersConfig.getSlaveConnectionPoolSize()).setReadMode(clusterServersConfig.getReadMode()).setSubscriptionMode(clusterServersConfig.getSubscriptionMode());}log.info("初始化 redis 配置");};}/*** redis集群配置 yml** --- # redis 集群配置(单机与集群只能开启一个另一个需要注释掉)* spring:*   redis:*     cluster:*       nodes:*         - 192.168.0.100:6379*         - 192.168.0.101:6379*         - 192.168.0.102:6379*     # 密码*     password:*     # 连接超时时间*     timeout: 10s*     # 是否开启ssl*     ssl: false** redisson:*   # 线程池数量*   threads: 16*   # Netty线程池数量*   nettyThreads: 32*   # 集群配置*   clusterServersConfig:*     # 客户端名称*     clientName: ${ruoyi.name}*     # master最小空闲连接数*     masterConnectionMinimumIdleSize: 32*     # master连接池大小*     masterConnectionPoolSize: 64*     # slave最小空闲连接数*     slaveConnectionMinimumIdleSize: 32*     # slave连接池大小*     slaveConnectionPoolSize: 64*     # 连接空闲超时,单位:毫秒*     idleConnectionTimeout: 10000*     # 命令等待超时,单位:毫秒*     timeout: 3000*     # 发布和订阅连接池大小*     subscriptionConnectionPoolSize: 50*     # 读取模式*     readMode: "SLAVE"*     # 订阅模式*     subscriptionMode: "MASTER"*/}
public class KeyPrefixHandler implements NameMapper {private final String keyPrefix;public KeyPrefixHandler(String keyPrefix) {//前缀为空 则返回空前缀this.keyPrefix = StringUtils.isBlank(keyPrefix) ? "" : keyPrefix + ":";}/*** 增加前缀*/@Overridepublic String map(String name) {if (StringUtils.isBlank(name)) {return null;}if (StringUtils.isNotBlank(keyPrefix) && !name.startsWith(keyPrefix)) {return keyPrefix + name;}return name;}/*** 去除前缀*/@Overridepublic String unmap(String name) {if (StringUtils.isBlank(name)) {return null;}if (StringUtils.isNotBlank(keyPrefix) && name.startsWith(keyPrefix)) {return name.substring(keyPrefix.length());}return name;}
}

3. 封装工具类

@NoArgsConstructor(access = AccessLevel.PRIVATE)
@SuppressWarnings(value = {"unchecked", "rawtypes"})
public class RedisUtils {private static final RedissonClient CLIENT = SpringUtils.getBean(RedissonClient.class);/*** 限流** @param key          限流key* @param rateType     限流类型* @param rate         速率* @param rateInterval 速率间隔* @return -1 表示失败*/public static long rateLimiter(String key, RateType rateType, int rate, int rateInterval) {RRateLimiter rateLimiter = CLIENT.getRateLimiter(key);rateLimiter.trySetRate(rateType, rate, rateInterval, RateIntervalUnit.SECONDS);if (rateLimiter.tryAcquire()) {return rateLimiter.availablePermits();} else {return -1L;}}/*** 获取客户端实例*/public static RedissonClient getClient() {return CLIENT;}/*** 发布通道消息** @param channelKey 通道key* @param msg        发送数据* @param consumer   自定义处理*/public static <T> void publish(String channelKey, T msg, Consumer<T> consumer) {RTopic topic = CLIENT.getTopic(channelKey);topic.publish(msg);consumer.accept(msg);}public static <T> void publish(String channelKey, T msg) {RTopic topic = CLIENT.getTopic(channelKey);topic.publish(msg);}/*** 订阅通道接收消息** @param channelKey 通道key* @param clazz      消息类型* @param consumer   自定义处理*/public static <T> void subscribe(String channelKey, Class<T> clazz, Consumer<T> consumer) {RTopic topic = CLIENT.getTopic(channelKey);topic.addListener(clazz, (channel, msg) -> consumer.accept(msg));}/*** 缓存基本的对象,Integer、String、实体类等** @param key   缓存的键值* @param value 缓存的值*/public static <T> void setCacheObject(final String key, final T value) {setCacheObject(key, value, false);}/*** 缓存基本的对象,保留当前对象 TTL 有效期** @param key       缓存的键值* @param value     缓存的值* @param isSaveTtl 是否保留TTL有效期(例如: set之前ttl剩余90 set之后还是为90)* @since Redis 6.X 以上使用 setAndKeepTTL 兼容 5.X 方案*/public static <T> void setCacheObject(final String key, final T value, final boolean isSaveTtl) {RBucket<T> bucket = CLIENT.getBucket(key);if (isSaveTtl) {try {bucket.setAndKeepTTL(value);} catch (Exception e) {long timeToLive = bucket.remainTimeToLive();setCacheObject(key, value, Duration.ofMillis(timeToLive));}} else {bucket.set(value);}}/*** 缓存基本的对象,Integer、String、实体类等** @param key      缓存的键值* @param value    缓存的值* @param duration 时间*/public static <T> void setCacheObject(final String key, final T value, final Duration duration) {RBatch batch = CLIENT.createBatch();RBucketAsync<T> bucket = batch.getBucket(key);bucket.setAsync(value);bucket.expireAsync(duration);batch.execute();}/*** 注册对象监听器* <p>* key 监听器需开启 `notify-keyspace-events` 等 redis 相关配置** @param key      缓存的键值* @param listener 监听器配置*/public static <T> void addObjectListener(final String key, final ObjectListener listener) {RBucket<T> result = CLIENT.getBucket(key);result.addListener(listener);}/*** 设置有效时间** @param key     Redis键* @param timeout 超时时间* @return true=设置成功;false=设置失败*/public static boolean expire(final String key, final long timeout) {return expire(key, Duration.ofSeconds(timeout));}/*** 设置有效时间** @param key      Redis键* @param duration 超时时间* @return true=设置成功;false=设置失败*/public static boolean expire(final String key, final Duration duration) {RBucket rBucket = CLIENT.getBucket(key);return rBucket.expire(duration);}/*** 获得缓存的基本对象。** @param key 缓存键值* @return 缓存键值对应的数据*/public static <T> T getCacheObject(final String key) {RBucket<T> rBucket = CLIENT.getBucket(key);return rBucket.get();}/*** 获得key剩余存活时间** @param key 缓存键值* @return 剩余存活时间*/public static <T> long getTimeToLive(final String key) {RBucket<T> rBucket = CLIENT.getBucket(key);return rBucket.remainTimeToLive();}/*** 删除单个对象** @param key 缓存的键值*/public static boolean deleteObject(final String key) {return CLIENT.getBucket(key).delete();}/*** 删除集合对象** @param collection 多个对象*/public static void deleteObject(final Collection collection) {RBatch batch = CLIENT.createBatch();collection.forEach(t -> {batch.getBucket(t.toString()).deleteAsync();});batch.execute();}/*** 检查缓存对象是否存在** @param key 缓存的键值*/public static boolean isExistsObject(final String key) {return CLIENT.getBucket(key).isExists();}/*** 缓存List数据** @param key      缓存的键值* @param dataList 待缓存的List数据* @return 缓存的对象*/public static <T> boolean setCacheList(final String key, final List<T> dataList) {RList<T> rList = CLIENT.getList(key);return rList.addAll(dataList);}/*** 注册List监听器* <p>* key 监听器需开启 `notify-keyspace-events` 等 redis 相关配置** @param key      缓存的键值* @param listener 监听器配置*/public static <T> void addListListener(final String key, final ObjectListener listener) {RList<T> rList = CLIENT.getList(key);rList.addListener(listener);}/*** 获得缓存的list对象** @param key 缓存的键值* @return 缓存键值对应的数据*/public static <T> List<T> getCacheList(final String key) {RList<T> rList = CLIENT.getList(key);return rList.readAll();}/*** 缓存Set** @param key     缓存键值* @param dataSet 缓存的数据* @return 缓存数据的对象*/public static <T> boolean setCacheSet(final String key, final Set<T> dataSet) {RSet<T> rSet = CLIENT.getSet(key);return rSet.addAll(dataSet);}/*** 注册Set监听器* <p>* key 监听器需开启 `notify-keyspace-events` 等 redis 相关配置** @param key      缓存的键值* @param listener 监听器配置*/public static <T> void addSetListener(final String key, final ObjectListener listener) {RSet<T> rSet = CLIENT.getSet(key);rSet.addListener(listener);}/*** 获得缓存的set** @param key 缓存的key* @return set对象*/public static <T> Set<T> getCacheSet(final String key) {RSet<T> rSet = CLIENT.getSet(key);return rSet.readAll();}/*** 缓存Map** @param key     缓存的键值* @param dataMap 缓存的数据*/public static <T> void setCacheMap(final String key, final Map<String, T> dataMap) {if (dataMap != null) {RMap<String, T> rMap = CLIENT.getMap(key);rMap.putAll(dataMap);}}/*** 注册Map监听器* <p>* key 监听器需开启 `notify-keyspace-events` 等 redis 相关配置** @param key      缓存的键值* @param listener 监听器配置*/public static <T> void addMapListener(final String key, final ObjectListener listener) {RMap<String, T> rMap = CLIENT.getMap(key);rMap.addListener(listener);}/*** 获得缓存的Map** @param key 缓存的键值* @return map对象*/public static <T> Map<String, T> getCacheMap(final String key) {RMap<String, T> rMap = CLIENT.getMap(key);return rMap.getAll(rMap.keySet());}/*** 获得缓存Map的key列表** @param key 缓存的键值* @return key列表*/public static <T> Set<String> getCacheMapKeySet(final String key) {RMap<String, T> rMap = CLIENT.getMap(key);return rMap.keySet();}/*** 往Hash中存入数据** @param key   Redis键* @param hKey  Hash键* @param value 值*/public static <T> void setCacheMapValue(final String key, final String hKey, final T value) {RMap<String, T> rMap = CLIENT.getMap(key);rMap.put(hKey, value);}/*** 获取Hash中的数据** @param key  Redis键* @param hKey Hash键* @return Hash中的对象*/public static <T> T getCacheMapValue(final String key, final String hKey) {RMap<String, T> rMap = CLIENT.getMap(key);return rMap.get(hKey);}/*** 删除Hash中的数据** @param key  Redis键* @param hKey Hash键* @return Hash中的对象*/public static <T> T delCacheMapValue(final String key, final String hKey) {RMap<String, T> rMap = CLIENT.getMap(key);return rMap.remove(hKey);}/*** 获取多个Hash中的数据** @param key   Redis键* @param hKeys Hash键集合* @return Hash对象集合*/public static <K, V> Map<K, V> getMultiCacheMapValue(final String key, final Set<K> hKeys) {RMap<K, V> rMap = CLIENT.getMap(key);return rMap.getAll(hKeys);}/*** 设置原子值** @param key   Redis键* @param value 值*/public static void setAtomicValue(String key, long value) {RAtomicLong atomic = CLIENT.getAtomicLong(key);atomic.set(value);}/*** 获取原子值** @param key Redis键* @return 当前值*/public static long getAtomicValue(String key) {RAtomicLong atomic = CLIENT.getAtomicLong(key);return atomic.get();}/*** 递增原子值** @param key Redis键* @return 当前值*/public static long incrAtomicValue(String key) {RAtomicLong atomic = CLIENT.getAtomicLong(key);return atomic.incrementAndGet();}/*** 递减原子值** @param key Redis键* @return 当前值*/public static long decrAtomicValue(String key) {RAtomicLong atomic = CLIENT.getAtomicLong(key);return atomic.decrementAndGet();}/*** 获得缓存的基本对象列表** @param pattern 字符串前缀* @return 对象列表*/public static Collection<String> keys(final String pattern) {Stream<String> stream = CLIENT.getKeys().getKeysStreamByPattern(pattern);return stream.collect(Collectors.toList());}/*** 删除缓存的基本对象列表** @param pattern 字符串前缀*/public static void deleteKeys(final String pattern) {CLIENT.getKeys().deleteByPattern(pattern);}/*** 检查redis中是否存在key** @param key 键*/public static Boolean hasKey(String key) {RKeys rKeys = CLIENT.getKeys();return rKeys.countExists(key) > 0;}
}

二、应用

1. RedisUtils工具类的基本使用

创建接口

@GetMapping("key")
public String getKey(String key){return RedisUtils.getCacheObject(key);
}@GetMapping("setKey")
public String setKey(String key,String value){RedisUtils.setCacheObject(key,value);return "success";
}

设置key

在这里插入图片描述
在这里插入图片描述

获取key对应的值

在这里插入图片描述
其他方法的作用,可以自行测试。这里就不再演示使用

三、队列

redission也支持队列,下面封装了一些队列的相关方法。可以处理了一些简单的队列任务,如果业务复杂可以选择mq

1. 工具类

@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class QueueUtils {private static final RedissonClient CLIENT = SpringUtil.getBean(RedissonClient.class);/*** 获取客户端实例*/public static RedissonClient getClient() {return CLIENT;}/*** 添加普通队列数据** @param queueName 队列名* @param data      数据*/public static <T> boolean addQueueObject(String queueName, T data) {RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);return queue.offer(data);}/*** 通用获取一个队列数据 没有数据返回 null(不支持延迟队列)** @param queueName 队列名*/public static <T> T getQueueObject(String queueName) {RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);return queue.poll();}/*** 通用删除队列数据(不支持延迟队列)*/public static <T> boolean removeQueueObject(String queueName, T data) {RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);return queue.remove(data);}/*** 通用销毁队列 所有阻塞监听 报错(不支持延迟队列)*/public static <T> boolean destroyQueue(String queueName) {RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);return queue.delete();}/*** 添加延迟队列数据 默认毫秒** @param queueName 队列名* @param data      数据* @param time      延迟时间*/public static <T> void addDelayedQueueObject(String queueName, T data, long time) {addDelayedQueueObject(queueName, data, time, TimeUnit.MILLISECONDS);}/*** 添加延迟队列数据** @param queueName 队列名* @param data      数据* @param time      延迟时间* @param timeUnit  单位*/public static <T> void addDelayedQueueObject(String queueName, T data, long time, TimeUnit timeUnit) {RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);RDelayedQueue<T> delayedQueue = CLIENT.getDelayedQueue(queue);delayedQueue.offer(data, time, timeUnit);}/*** 获取一个延迟队列数据 没有数据返回 null** @param queueName 队列名*/public static <T> T getDelayedQueueObject(String queueName) {RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);RDelayedQueue<T> delayedQueue = CLIENT.getDelayedQueue(queue);return delayedQueue.poll();}/*** 删除延迟队列数据*/public static <T> boolean removeDelayedQueueObject(String queueName, T data) {RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);RDelayedQueue<T> delayedQueue = CLIENT.getDelayedQueue(queue);return delayedQueue.remove(data);}/*** 销毁延迟队列 所有阻塞监听 报错*/public static <T> void destroyDelayedQueue(String queueName) {RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);RDelayedQueue<T> delayedQueue = CLIENT.getDelayedQueue(queue);delayedQueue.destroy();}/*** 添加优先队列数据** @param queueName 队列名* @param data      数据*/public static <T> boolean addPriorityQueueObject(String queueName, T data) {RPriorityBlockingQueue<T> priorityBlockingQueue = CLIENT.getPriorityBlockingQueue(queueName);return priorityBlockingQueue.offer(data);}/*** 优先队列获取一个队列数据 没有数据返回 null(不支持延迟队列)** @param queueName 队列名*/public static <T> T getPriorityQueueObject(String queueName) {RPriorityBlockingQueue<T> queue = CLIENT.getPriorityBlockingQueue(queueName);return queue.poll();}/*** 优先队列删除队列数据(不支持延迟队列)*/public static <T> boolean removePriorityQueueObject(String queueName, T data) {RPriorityBlockingQueue<T> queue = CLIENT.getPriorityBlockingQueue(queueName);return queue.remove(data);}/*** 优先队列销毁队列 所有阻塞监听 报错(不支持延迟队列)*/public static <T> boolean destroyPriorityQueue(String queueName) {RPriorityBlockingQueue<T> queue = CLIENT.getPriorityBlockingQueue(queueName);return queue.delete();}/*** 尝试设置 有界队列 容量 用于限制数量** @param queueName 队列名* @param capacity  容量*/public static <T> boolean trySetBoundedQueueCapacity(String queueName, int capacity) {RBoundedBlockingQueue<T> boundedBlockingQueue = CLIENT.getBoundedBlockingQueue(queueName);return boundedBlockingQueue.trySetCapacity(capacity);}/*** 尝试设置 有界队列 容量 用于限制数量** @param queueName 队列名* @param capacity  容量* @param destroy   已存在是否销毁*/public static <T> boolean trySetBoundedQueueCapacity(String queueName, int capacity, boolean destroy) {RBoundedBlockingQueue<T> boundedBlockingQueue = CLIENT.getBoundedBlockingQueue(queueName);if (boundedBlockingQueue.isExists() && destroy) {destroyQueue(queueName);}return boundedBlockingQueue.trySetCapacity(capacity);}/*** 添加有界队列数据** @param queueName 队列名* @param data      数据* @return 添加成功 true 已达到界限 false*/public static <T> boolean addBoundedQueueObject(String queueName, T data) {RBoundedBlockingQueue<T> boundedBlockingQueue = CLIENT.getBoundedBlockingQueue(queueName);return boundedBlockingQueue.offer(data);}/*** 有界队列获取一个队列数据 没有数据返回 null(不支持延迟队列)** @param queueName 队列名*/public static <T> T getBoundedQueueObject(String queueName) {RBoundedBlockingQueue<T> queue = CLIENT.getBoundedBlockingQueue(queueName);return queue.poll();}/*** 有界队列删除队列数据(不支持延迟队列)*/public static <T> boolean removeBoundedQueueObject(String queueName, T data) {RBoundedBlockingQueue<T> queue = CLIENT.getBoundedBlockingQueue(queueName);return queue.remove(data);}/*** 有界队列销毁队列 所有阻塞监听 报错(不支持延迟队列)*/public static <T> boolean destroyBoundedQueue(String queueName) {RBoundedBlockingQueue<T> queue = CLIENT.getBoundedBlockingQueue(queueName);return queue.delete();}/*** 订阅阻塞队列(可订阅所有实现类 例如: 延迟 优先 有界 等)*/public static <T> void subscribeBlockingQueue(String queueName, Consumer<T> consumer, boolean isDelayed) {RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);if (isDelayed) {// 订阅延迟队列CLIENT.getDelayedQueue(queue);}queue.subscribeOnElements(consumer);}}

2. 普通队列

添加数据到队列

@GetMapping("add")
public String add(){QueueUtils.addQueueObject("queue:simple",1);QueueUtils.addQueueObject("queue:simple",2);QueueUtils.addQueueObject("queue:simple",3);return "ok";
}

在这里插入图片描述

消费队列数据

遵循先进先出,获取数据后就会删除。如果队列中没有数据,获取到的就为null

@GetMapping("get")
public Integer get(){return QueueUtils.getQueueObject("queue:simple");
}

在这里插入图片描述

移除队列数据

@GetMapping("remove")
public String remove(){QueueUtils.removeQueueObject("queue:simple",3);return "ok";
}

在这里插入图片描述

销毁队列

@GetMapping("destroy")
public String destroy(){QueueUtils.destroyQueue("queue:simple");return "ok";
}

在这里插入图片描述

订阅队列消息

  • 订阅的消息一般在项目启动的时候使用,只能订阅一次
  • 当监听到队列新增数据的时候会立即取出来进行消费
@PostConstruct
public void sub(){QueueUtils.subscribeBlockingQueue("queue:simple",(o)->{System.out.println("接收到消息:"+o);},false);
}

我们再次调用新增

在这里插入图片描述

3. 有界队列(限制数据量)

设置队列最大容量

有界队列在使用前必须设置容量

@GetMapping("set")
public String set(){boolean b = QueueUtils.trySetBoundedQueueCapacity("queue:bound", 10);return  "ok";
}

在这里插入图片描述

新增有界队列数据

@GetMapping("add")
public String add(){QueueUtils.addBoundedQueueObject("queue:bound",1);return "ok";
}

新增完毕后我们可以发现,我们直接设置的最大容量变成来了9。每次添加数据都会查询当前最大容量是否>0,如果大于0添加成功并且减一,否则添加失败

在这里插入图片描述
在这里插入图片描述

获取有界队列数据

@GetMapping("get")
public Integer get(){return QueueUtils.getBoundedQueueObject("queue:bound");
}

我们可以看到当获取数据的时候,容量+1,数据从redis中删除

在这里插入图片描述
在这里插入图片描述

其他用法与普通队列类似,就不再演示了

4. 延迟队列(延迟获取数据)

添加延迟数据

延迟队列的实现原理是将数据添加到另一个缓存队列中,当到达指定时间才会转移到普通队列中

@GetMapping("add")
public String add(){QueueUtils.addDelayedQueueObject("queue:belay",1,10, TimeUnit.SECONDS);return "ok";
}

获取延迟数据

必须达到指定时间后才能获取

@GetMapping("get")
public Integer get(){return QueueUtils.getDelayedQueueObject("queue:belay");
}

删除延迟数据

@GetMapping("remove")
public String remove(){QueueUtils.removeQueueObject("queue:belay",3);return "ok";
}

清空延迟数据

@GetMapping("destroy")
public String destroy(){QueueUtils.destroyDelayedQueue("queue:belay");return "ok";
}

订阅消息使用方法同普通队列类似,第三个参数需要改为true

5. 优先队列(数据可插队)

插入优先队列的数据我们需要先实现比较接口

@Data
@Accessors(chain = true)
class Order implements Comparable<Order>{private Long id;@Overridepublic int compareTo(Order o) {return Long.compare(getId(), o.id);}
}

新增优先数据

@GetMapping("add")
public String add(){QueueUtils.addPriorityQueueObject("queue:priority",new Order().setId(1L));QueueUtils.addPriorityQueueObject("queue:priority",new Order().setId(6L));QueueUtils.addPriorityQueueObject("queue:priority",new Order().setId(2L));QueueUtils.addPriorityQueueObject("queue:priority",new Order().setId(5L));QueueUtils.addPriorityQueueObject("queue:priority",new Order().setId(22L));QueueUtils.addPriorityQueueObject("queue:priority",new Order().setId(3L));return "ok";
}

我们可以看到插入的数据是有序的

在这里插入图片描述

获取优先队列数据

@GetMapping("get")
public Integer get(){return QueueUtils.getPriorityQueueObject("queue:priority");
}

删除优先队列数据

@GetMapping("remove")
public String remove(){QueueUtils.removeQueueObject("queue:priority",3);return "ok";
}

清空优先队列数据

@GetMapping("destroy")
public String destroy(){QueueUtils.destroyDelayedQueue("queue:priority");return "ok";
}

订阅消息使用方法同普通队列一样

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.rhkb.cn/news/284941.html

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

【Java多线程(1)】创建线程的几种方式和Thread类及其常见方法

目录 一、Java创建线程的方式 1. 通过继承 Thread 类实现多线程 2. 通过实现 Runnable 接口实现多线程 3. 其他变形 二、Thread类及常见方法 1. Thread类的常见构造方法 2. Thread类的几个常见属性 2.1 getName() 2.2 setDaemon() & isDaemon() 2.3 isAlive() …

HTML5+CSS3+移动web——CSS进阶

系列文章目录 HTML5CSS3移动web——HTML 基础-CSDN博客https://blog.csdn.net/ymxk2876721452/article/details/136070953?spm1001.2014.3001.5502HTML5CSS3移动web——列表、表格、表单-CSDN博客https://blog.csdn.net/ymxk2876721452/article/details/136221443?spm1001.20…

复习Day3

1231. 航班时间 - AcWing题库 #include<bits/stdc.h> using namespace std; int getTime(){//得到时间 int h1,m1,s1,h2,m2,s2,d0;scanf("%d:%d:%d %d:%d:%d (%d)",&h1,&m1,&s1,&h2,&m2,&s2,&d);//补匹配直接跳过 int timed*24*3…

【Django开发】0到1美多商城项目md教程第3篇:用户注册业务实现,1. 用户注册页面绑定Vue数据【附代码文档】

美多商城完整教程&#xff08;附代码资料&#xff09;主要内容讲述&#xff1a;欢迎来到美多商城&#xff01;&#xff0c;项目准备。展示用户注册页面&#xff0c;创建用户模块子应用。用户注册业务实现&#xff0c;用户注册前端逻辑。图形验证码&#xff0c;图形验证码接口设…

《自动机理论、语言和计算导论》阅读笔记:p1-p4

《自动机理论、语言和计算导论》学习第1天&#xff0c;p1-p4&#xff0c;总计4页。这只是个人的学习记录&#xff0c;因为很多东西不懂&#xff0c;难免存在理解错误的地方。 一、技术总结 1.有限自动机(finite automata)示例 1.software for checking digital circuits。 …

IEEE TRANSACTIONS ON INTELLIGENT TRANSPORTATION SYSTEMS (T-ITS) 投稿记录,欢迎大家评论~

投稿整个流程时间点&#xff1a;Submitted: 17 October 2023 Awaiting Reviewer Assignment: From 18 October 2023 to 6 November 2023 Under review: From 6 November 2023 to 30 November 2023 Awaiting reviewer scores: From 1 December 2023 to 13 January 2024 Aw…

概率论与数理统计-条件概率题目2-3次事件依次发生问题

题目&#xff1a;设某光学仪器厂制造的透镜,第一次落下时打破的概率为1/2,若第一次落下未打破,第二次落下打破的概率为7/10,若前两次落下未打破,第三次落下打破的概率为9/10.试求透镜落下三次而未打破的概率&#xff1f; 解答&#xff1a; 设Ai表示透镜第i次下落打破&#xff…

【微服务】以模块化单体架构开发微服务应用

目录 推荐超级课程: Docker快速入门到精通Kubernetes入门到大师通关课AWS云服务快速入门实战我们知道,起初,单体应用有显著的优势:它们更容易开发和部署。从开发人员的角度来看,这种简单性是有益的。一切都是集中的,可以快速更新任何部分的业务逻辑并立即看到结果。这种开…

Redis八股

持久化 RDB AOF 数据过期策略 惰性删除 定期删除 数据淘汰策略 分布式锁 集群模型 主从同步原理 哨兵模式 分片集群 网络模型

哪本书最了解孩子?跟《米小圈上学记》一起做孩子的“引路人”!

孩子是发展中的人&#xff0c;需要家长的陪伴&#xff0c;孩子的身心发展是有规侓的&#xff0c;是处于发展过程中的人。我们要学会尊重孩子的发展&#xff0c;从兴趣出发&#xff0c;关注孩子的成长。但是&#xff0c;家长不可能无时无刻都能陪在孩子身边&#xff0c;他需要一…

day06vue2学习

day06 路由的封装抽离 问题&#xff1a;所有的路由配置都堆在main.js中不太合适么&#xff1f;不好&#xff0c;会加大代码的复杂度 目标&#xff1a;将路由模块抽离出来。好处&#xff1a;差分模块&#xff0c;利于维护。 大致的做法就是&#xff0c;将路由相关的东西都提…

使用docker-compose搭建wordpress博客

1、从远程仓库拉取worldpress镜像到本地 2、新建一个项目&#xff0c;然后在新建的项目目录里面新建一个docker-compose.yml模版文件。 3、编写docker-compose.yml文件 4、docker-compose up 运行项目。 5、在浏览器测试 使用docker-compose搭建wordpress博客实验成功。

03-SparkSQL入门

0 Shark Spark 的一个组件&#xff0c;用于大规模数据分析的 SQL 查询引擎。Shark 提供了一种基于 SQL 的交互式查询方式&#xff0c;可以让用户轻松地对大规模数据集进行查询和分析。Shark 基于 Hive 项目&#xff0c;使用 Hive 的元数据存储和查询语法&#xff0c;并基于Hiv…

基于Java中的SSM框架实现快餐店线上点餐系统项目【项目源码+论文说明】计算机毕业设计

基于Java中的SSM框架实现快餐店线上点餐系统演示 摘要 随着计算机互联网的高速发展。餐饮业的发展也加入了电子商务团队。各种网上点餐系统纷纷涌现&#xff0c;不仅增加了商户的销售量和营业额&#xff0c;而且为买家提供了极大的方便&#xff0c;足不出户&#xff0c;就能订…

arm 外部中断

main.c: #include"key_inc.h" //封装延时函数 void delay(int ms) {int i,j;for(i0;i<ms;i){for(j0;j<2000;j){}} } int main() {//按键中断的初始化key1_it_config();key2_it_config();key3_it_config();while(1){printf("in main pro\n");delay(1…

AugmentedReality之路-创建工程及主界面(1)

本文从零创建1个工程&#xff0c;并添加1个BasicMap和1个主界面&#xff0c;主界面包含Start AR和Stop AR两个按钮 1、创建移动端工程并打包运行 创建1个空工程&#xff0c;选择C&#xff0c;Mobile&#xff0c;Scalable&#xff0c;勾选StarterContent 通过Edit->Proje…

MySQL学习笔记(一)

1、什么是数据库&#xff1f;什么是数据库管理系统&#xff1f;什么是SQL&#xff1f;他们之间的关系是什么&#xff1f; 数据库&#xff1a;英文单词DataBase&#xff0c;简称DB。按照一定格式存储数据的一些文件的组合。顾名思义&#xff0c;存储数据的仓库&#xff0c;实际…

MySQL的基本操作与增删改查管理操作

一、MySQL数据库sql语句 1.1 sql 命令 database数据库table表row行column列user用户select从数据表中获取数据updata更新数据库中的数据delete从数据库中删除数据insert into 向数据表插入数据create database创建新数据库alter database修改数据库create table创建新表alter…

MySql实战--一条SQL查询语句是如何执行的?

平时我们使用数据库&#xff0c;看到的通常都是一个整体。比如&#xff0c;你有个最简单的表&#xff0c;表里只有一个ID字段&#xff0c;在执行下面这个查询语句时&#xff1a; select * from T where ID10&#xff1b; 我们看到的只是输入一条语句&#xff0c;返回一个结果…

WorkPlus一站式IM即时通讯解决方案,提升企业沟通效率与协作能力

在企业内部沟通与协作中&#xff0c;高效的即时通讯是实现团队协作与工作效率的重要保障。而WorkPlus以其稳定可靠的性能和全面的功能&#xff0c;为企业提供一站式的IM即时通讯解决方案&#xff0c;助力企业提升沟通效率与协作能力。IM即时通讯在企业中的重要性不言而喻。作为…