likes
comments
collection
share

基于Redission高级应用9-RQueue RDeque 实战应用

作者站长头像
站长
· 阅读数 25

RQueue/RDeque 实现原理Redisson

的 RQueue 和 RDeque 是基于 Redis 的列表（list）数据结构实现的分布式 Java 队列和双端队列（deque）。

RQueue

  • 实现原理：RQueue 使用 Redis 的 LPUSH 命令来添加元素到队列的尾部，使用 RPOP 命令从队列的头部移除元素。
  • 优点：
    • 分布式：由于基于 Redis，它天然支持分布式环境。
    • 持久化：Redis 提供的持久化机制保证了队列数据不会因服务器重启而丢失。
    • 性能：Redis 提供了高性能的操作，可以快速响应队列命令。
    • 原子操作：队列操作是原子的，保证了在并发环境下的数据一致性。
  • 缺点：
    • 内存限制：由于数据存储在内存中，队列的大小受到物理内存的限制。
    • 成本：与在JVM内存中的队列相比，Redis 需要单独的服务器和网络通信，可能会增加成本。

一个基本的消息队列处理流程示例：

生产者RQueue消费者发送消息存储消息拉取消息处理消息确认消息处理完成生产者RQueue消费者

RDeque

  • 实现原理：RDeque 除了支持 RQueue 的操作外，还可以使用 RPUSH 将元素添加到队列的头部，使用 LPOP 从队列的尾部移除元素。

  • 优点：

    • 双端操作：可以从两端插入或删除元素，提供了更高的灵活性。
    • 其他优点：与 RQueue 相同，包括分布式支持、持久化、高性能和原子操作。

  • 缺点：

    • 内存限制：与 RQueue 相同，受限于物理内存。
    • 成本：可能比在JVM内存中的双端队列增加更多成本。

    绘制 RDeque（Redis 双端队列）的操作流程图，我们可以考虑几个基本的操作，比如添加元素到队列头部或尾部，以及从队列头部或尾部移除元素:

    用户RDeque队列头部添加元素A队列尾部添加元素B查看队列头部元素移除队列尾部元素移除队列头部元素addFirst(元素A)addLast(元素B)peekFirst()得到元素ApollLast()得到元素BpollFirst()得到元素A用户RDeque

以下是使用 Mermaid 语法绘制的 RDeque 在 Redisson 中的简化类图，展示了它的一些基本操作和继承关系：

RQueue
+offer(E e)
+poll()
+peek()
RDeque
+addFirst(E e)
+addLast(E e)
+pollFirst()
+pollLast()
+peekFirst()
+peekLast()
RBlockingQueue
+put(E e)
+take()
RBlockingDeque
+putFirst(E e)
+putLast(E e)
+takeFirst()
+takeLast()

工具类示例

下面是一个使用 Spring 和 Redisson 实现的工具类示例，用于操作分布式队列和双端队列：

RDeque :

import org.redisson.api.RDeque;
import org.redisson.api.RFuture;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.ExecutionException;
/**
 * @Author derek_smart
 * @Date 202/4/25 17:25
 * @Description RDeque 工具类
 * <p>
 */
@Component
public class RedissonDQueueUtils {

    private final RedissonClient redissonClient;

    @Autowired
    public RedissonDQueueUtils(RedissonClient redissonClient) {
        this.redissonClient = redissonClient;
    }

    // 添加元素到队列
    public <T> void enqueue(String queueName, T element) {
        RDeque<T> deque = redissonClient.getDeque(queueName);
        deque.addLast(element);
    }

    // 从队列头部移除元素
    public <T> T dequeue(String queueName) {
        RDeque<T> deque = redissonClient.getDeque(queueName);
        return deque.pollFirst();
    }

    // 添加元素到队列头部（对于双端队列）
    public <T> void push(String dequeName, T element) {
        RDeque<T> deque = redissonClient.getDeque(dequeName);
        deque.addFirst(element);
    }

    // 从队列尾部移除元素（对于双端队列）
    public <T> T pop(String dequeName) {
        RDeque<T> deque = redissonClient.getDeque(dequeName);
        return deque.pollLast();
    }

    // 获取队列所有元素
    public <T> Collection<T> getAllElements(String queueName) {
        RDeque<T> deque = redissonClient.getDeque(queueName);
        return deque.readAll();
    }

    // 获取队列大小
    public int getSize(String queueName) {
        RDeque<?> deque = redissonClient.getDeque(queueName);
        return deque.size();
    }

    // 异步添加元素到队列
    public <T> RFuture<Boolean> enqueueAsync(String queueName, T element) {
        RDeque<T> deque = redissonClient.getDeque(queueName);
        return deque.offerLastAsync(element);
    }

    // 异步从队列头部移除元素
    public <T> RFuture<T> dequeueAsync(String queueName) {
        RDeque<T> deque = redissonClient.getDeque(queueName);
        return deque.pollFirstAsync();
    }

    // 批量添加元素到队列
    public <T> boolean enqueueAll(String queueName, Collection<? extends T> elements) {
        RDeque<T> deque = redissonClient.getDeque(queueName);
        return deque.addAll(elements);
    }

    // 批量从队列移除元素，并返回移除的元素集合
    public <T> List<T> dequeueBatch(String queueName, int batchSize) {
        RDeque<T> deque = redissonClient.getDeque(queueName);
        return deque.pollFirst(batchSize);
    }

    // 清空队列
    public void clearQueue(String queueName) {
        RDeque<?> deque = redissonClient.getDeque(queueName);
        deque.clear();
    }

    // 异步执行的通用方法，用于处理异步结果
    private <T> T executeAsync(RFuture<T> future) {
        try {
            return future.get();
        } catch (InterruptedException | ExecutionException e) {
            // 日志记录异常，可以根据实际情况进行异常处理
            // log.error("Error executing async operation", e);
            Thread.currentThread().interrupt();
            return null;
        }
    }
}

基于Redission高级应用9-RQueue RDeque 实战应用

在这个工具类中，我们定义了几个基本的队列操作方法: 比如入队（enqueue）、出队（dequeue）、双端队列的头部入队（push）和尾部出队（pop）。我们还包括了获取队列所有元素的方法和获取队列大小的方法。

RQueue:

import org.redisson.api.RQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Collection;
import java.util.Iterator;
/**
 * @Author derek_smart
 * @Date 202/4/25 18:25
 * @Description Deque 工具类
 * <p>
 */
@Component
public class RedissonQueueUtils {

    private final RedissonClient redissonClient;

    @Autowired
    public RedissonQueueUtils(RedissonClient redissonClient) {
        this.redissonClient = redissonClient;
    }

    // 添加元素到队列
    public <T> boolean enqueue(String queueName, T element) {
        RQueue<T> queue = redissonClient.getQueue(queueName);
        return queue.offer(element);
    }

    // 从队列头部移除元素
    public <T> T dequeue(String queueName) {
        RQueue<T> queue = redissonClient.getQueue(queueName);
        return queue.poll();
    }
// 版本过低
/*    // 阻塞式从队列头部移除元素
    public <T> T take(String queueName) throws InterruptedException {
        RQueue<T> queue = redissonClient.getQueue(queueName);
        return queue.take();
    }*/

    // 查看队列头部元素但不移除
    public <T> T peek(String queueName) {
        RQueue<T> queue = redissonClient.getQueue(queueName);
        return queue.peek();
    }
// 版本过低
/*    // 添加元素到队列并设置超时时间
    public <T> boolean enqueue(String queueName, T element, long timeout, TimeUnit unit) {
        RQueue<T> queue = redissonClient.getQueue(queueName);
        return queue.offer(element, timeout, unit);
    }*/

    // 批量添加元素到队列
    public <T> boolean enqueueAll(String queueName, Collection<? extends T> elements) {
        RQueue<T> queue = redissonClient.getQueue(queueName);
        return queue.addAll(elements);
    }

    // 获取队列大小
    public int getSize(String queueName) {
        RQueue<?> queue = redissonClient.getQueue(queueName);
        return queue.size();
    }

    // 获取队列的迭代器
    public <T> Iterator<T> iterator(String queueName) {
        RQueue<T> queue = redissonClient.getQueue(queueName);
        return queue.iterator();
    }

    // 根据条件移除队列中的元素
    public <T> boolean removeIf(String queueName, java.util.function.Predicate<?super T> filter) {
        RQueue<T> queue = redissonClient.getQueue(queueName);
        return queue.removeIf(filter);
    }

    // 清空队列
    public void clearQueue(String queueName) {
        RQueue<?> queue = redissonClient.getQueue(queueName);
        queue.clear();
    }


//版本低
/*    // 阻塞式出队，带超时
    public <T> T dequeue(String queueName, long timeout, TimeUnit unit) throws InterruptedException {
        RQueue<T> queue = redissonClient.getQueue(queueName);
        return queue.poll(timeout, unit);
    }*/


//版本低
/*    // 获取并移除队列尾部元素
    public <T> T removeLast(String queueName) {
        RQueue<T> queue = redissonClient.getQueue(queueName);
        return queue.pollLast();
    }*/

    // 检查队列是否包含某个元素
    public <T> boolean contains(String queueName, Object element) {
        RQueue<T> queue = redissonClient.getQueue(queueName);
        return queue.contains(element);
    }

/*    // 从队列中读取元素（不移除）
    public <T> T get(String queueName, int index) {
        RQueue<T> queue = redissonClient.getQueue(queueName);
        return queue.get(index);
    }*/

    // 转移元素到另一个队列
    public <T> T transferTo(String sourceQueueName, String destinationQueueName) {
        RQueue<T> sourceQueue = redissonClient.getQueue(sourceQueueName);
        RQueue<T> destinationQueue = redissonClient.getQueue(destinationQueueName);
        return sourceQueue.pollLastAndOfferFirstTo(destinationQueue.getName());
    }

/*
    // 设置队列的最大容量
    public <T> void setCapacity(String queueName, int capacity) {
        RQueue<T> queue = redissonClient.getQueue(queueName);
        queue.trySetMaxSize(capacity);
    }
*/

}

基于Redission高级应用9-RQueue RDeque 实战应用

使用用例： 提供一些使用 RedissonQueueUtils 工具类的示例用例，以及相应的代码实现。这些示例将展示如何在实际应用中调用这个工具类的方法。

示例用例 1: 添加元素到队列

场景: 将一个新的任务添加到任务队列中。

// 假设我们有一个任务对象
class Task {
    private String description;

    // 构造函数、getter 和 setter
    public Task(String description) {
        this.description = description;
    }

    public String getDescription() {
        return description;
    }

    public void setDescription(String description) {
        this.description = description;
    }
}

// 在服务中使用 RedissonQueueUtils 工具类
@Service
public class TaskService {

    @Autowired
    private RedissonQueueUtils queueUtils;

    public void addTaskToQueue(Task task) {
        queueUtils.enqueue("taskQueue", task);
    }
}

示例用例 2: 阻塞式从队列中获取任务

场景: 从任务队列中获取任务进行处理，如果队列为空，则等待直到任务可用或超时。

@Service
public class TaskProcessorService {

    @Autowired
    private RedissonQueueUtils queueUtils;

    public Task getTaskFromQueue() {
        try {
            return queueUtils.dequeue("taskQueue", 10, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return null;
        }
    }
}

示例用例 3: 检查队列中是否包含某个特定任务

场景: 验证任务队列中是否已经存在一个特定的任务。

@Service
public class TaskVerificationService {

    @Autowired
    private RedissonQueueUtils queueUtils;

    public boolean isTaskInQueue(Task task) {
        return queueUtils.contains("taskQueue", task);
    }
}

示例用例 4: 转移任务到另一个队列

场景: 将任务从主任务队列转移到高优先级任务队列。

@Service
public class TaskTransferService {

    @Autowired
    private RedissonQueueUtils queueUtils;

    public void transferTaskToPriorityQueue() {
        queueUtils.transferTo("taskQueue", "priorityTaskQueue");
    }
}

示例用例 5: 设置队列的最大容量

场景: 为了避免队列无限增长，设置队列的最大容量。

@Service
public class TaskQueueSetupService {

    @Autowired
    private RedissonQueueUtils queueUtils;

    public void setMaxSizeForTaskQueue(int capacity) {
        queueUtils.setCapacity("taskQueue", capacity);
    }
}

这些示例展示了如何在服务中注入 RedissonQueueUtils 工具类，并调用其方法来执行队列相关操作。在实际应用中，你可能需要结合业务逻辑来决定何时和如何使用这些方法。此外，错误处理和异常处理应该根据实际的应用需求来定制。

转载自:https://juejin.cn/post/7363138988012240911
评论
请登录