Redis实现消息队列有三种主流方式：List结构实现、Pub/Sub发布订阅、Stream流实现，分别适用于简单队列、实时广播和高可靠场景。
用 while 循环实现一个轻量级内存消息队列处理器，核心在于构造一个持续运行的消费者线程，从阻塞队列中不断拉取消息并处理。它不依赖外部中间件，适合低并发、可容忍丢失、无需持久化的场景（比如日志异步落盘、内部通知、开发调试等）。
基于 BlockingQueue 的 while 消费循环
Java 的 BlockingQueue（如 LinkedBlockingQueue）天然支持线程安全的“取-阻塞-再取”语义，配合 while 循环就能构成稳定的消息消费主干：
- 使用
queue.take()：该方法会阻塞当前线程，直到队列有元素可取，避免空转和忙等待 - 循环体包裹在
try-catch中：捕获InterruptedException用于优雅退出，也兜住业务处理异常 - 循环条件不写显式布尔变量，而是靠中断状态判断：
!Thread.currentThread().isInterrupted()
关键代码结构示例
以下是一个典型的 Spring Boot Service 中的消费者启动逻辑（省略注解和注入）：
private final BlockingQueue<Task> queue = new LinkedBlockingQueue<>(1000);
private final ExecutorService consumer = Executors.newSingleThreadExecutor();
@PostConstruct
public void startConsumer() {
consumer.submit(() -> {
while (!Thread.currentThread().isInterrupted()) {
try {
Task task = queue.take(); // 阻塞等待，有任务才继续
processTask(task); // 执行业务逻辑（如发邮件、记日志）
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
} catch (Exception e) {
System.err.println("处理任务失败: " + e.getMessage());
}
}
});
}
@PreDestroy
public void shutdown() {
consumer.shutdown();
try {
if (!consumer.awaitTermination(5, TimeUnit.SECONDS)) {
consumer.shutdownNow();
}
} catch (InterruptedException e) {
consumer.shutdownNow();
Thread.currentThread().interrupt();
}
}
如何保证消息不被漏处理或重复处理
纯内存队列本身不具备 ACK 机制，所以需按场景做取舍：
- 若允许丢失：直接用
take()或poll()，处理完即丢弃，无需额外状态管理 - 若需至少一次（at-least-once）语义：可在
processTask()内部加重试逻辑（如最多 3 次），失败后记录告警，但不回滚队列 - 若要求精确一次（exactly-once）：仅靠内存队列无法满足，必须引入外部存储（如 Redis 记录已处理 ID）或改用 Redisson 等带确认机制的方案
配套的任务提交与队列控制
生产者端建议用非阻塞方式提交，避免因队列满导致主线程卡死：
立即学习“Java免费学习笔记（深入）”；
- 使用
queue.offer(task)：成功返回 true，失败立即返回 false（可配合日志或降级策略） - 避免用
put()：它会无限阻塞，可能拖垮上游业务线程 - 可设置容量上限（如 1000），并在 offer 失败时触发熔断，例如跳过发送、写入本地文件暂存或抛出警告
不复杂但容易忽略的是线程生命周期与 Spring 容器的协同——务必在 @PreDestroy 中关闭线程池，否则应用停机时线程残留会导致 JVM 无法正常退出。
