当前位置：首页 > news >正文

Kafka实现延迟消息

news 来源：原创 2025/9/17 16:40:16

Kafka 实现延迟消息

Kafka 本身不支持原生的延迟消息（不像 RocketMQ 内置了延迟队列），但可以通过多种方式来实现延迟消息。常见的方案如下：

1. 使用不同的 Topic 分区（最常见）

思路：

创建多个延迟队列 Topic，比如 delay-5s、delay-10s、delay-30s，代表不同延迟时间的队列。
生产者按延迟时间把消息发送到对应的 Topic。
消费者监听并处理这些 Topic，延迟对应的秒数之后【线程睡眠 Thread.sleep(delay)】转发到真正的业务主题 real-topic。

示例：

// 生产
String topic = "delay-10s"; // 发送到 10 秒延迟的队列
ProducerRecord<String, String> record = new ProducerRecord<>(topic, "message");
producer.send(record);// 消费while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));for (ConsumerRecord<String, String> record : records) {System.out.println("Received delayed message: " + record.value());// 3. 模拟延迟（等待 10 秒）try {Thread.sleep(10000);} catch (InterruptedException e) {e.printStackTrace();}// 4. 发送到真正的业务 TopicProducerRecord<String, String> forwardRecord = new ProducerRecord<>(REAL_TOPIC, record.key(), record.value());producer.send(forwardRecord);System.out.println("Message forwarded to real-topic: " + record.value());}}

适用于：固定延迟时间的场景，比如 5s、10s、30s。
优点：简单易用，不需要额外组件。
缺点：如果延迟时间种类很多，Topic 可能会很多，管理复杂。

2. 采用定时任务 + 数据库存储

思路：

生产者把消息存到 数据库（如 MySQL）或者 Redis，并记录目标执行时间。
定时任务（如 Quartz、XXL-JOB）轮询数据库，判断是否到时间，然后再投递到 Kafka 进行消费。

示例：

生产者：

// 先存到数据库，设置目标消费时间
saveToDatabase("message1", System.currentTimeMillis() + 10000); // 10秒后发送

定时任务：

List<Message> messages = queryReadyMessages(); // 查询到期的消息
for (Message msg : messages) {producer.send(new ProducerRecord<>("real-topic", msg.getContent()));deleteFromDatabase(msg.getId()); // 发送后删除
}

适用于：延迟时间不固定、消息量不大的情况。
优点：可以精准控制延迟时间，灵活性强。
缺点：依赖数据库或 Redis，效率受限。

3. 结合 Redis 的 ZSet（有序集合）

思路：

生产者将消息存入 Redis 的 ZSet，score 设为 目标消费时间戳。
消费者轮询 Redis，取出时间到期的消息，然后发送到 Kafka 进行消费。

示例：

生产者：

// message：消息体，delayMillis：延迟时间
public void addDelayTask(String message, long delayMillis) {long executeTime = System.currentTimeMillis() + delayMillis;redisTemplate.opsForZSet().add("delay-queue", message, executeTime);System.out.println("任务添加：" + message + "，执行时间：" + executeTime);
}

消费者（定时轮询取出到期消息）：

@Scheduled(fixedRate = 1000) // 每秒执行一次
public void consumeDelayTask() {long now = System.currentTimeMillis();// 返回一个 Set<String> 集合，包含所有 score 在 [0, now] 之间的元素。Set<String> messages = redisTemplate.opsForZSet().rangeByScore("delay-queue", 0, now);for (String msg : messages) {producer.send(new ProducerRecord<>("real-topic", msg)); // 可根据场景选择发送到目标主题 或者 直接处理消息redisTemplate.opsForZSet().remove("delay-queue", msg); // 处理完删除}
}

适用于：高并发场景，且对小量延迟消息有需求。
优点：基于内存操作，性能高。
缺点：需要额外维护 Redis，消息量过大可能导致性能问题。

4. 使用 Kafka Streams + State Store

思路：

Kafka Streams 允许我们使用时间窗口，可以让消息在流处理中等一段时间，然后再投递到目标 Topic。

示例：

KStream<String, String> stream = builder.stream("input-topic");
stream.transform(() -> new DelayTransformer(10 * 1000)) // 延迟 10 秒.to("output-topic");

适用于：流式计算场景，如订单超时处理。
优点：适合与 Kafka 生态集成，流式处理友好。
缺点：学习成本较高，对 Kafka Streams 有一定要求。

5. 结合 Flink 处理

如果你的架构里用到了 Flink，可以使用 Flink Timer 进行定时延迟处理：

接收 Kafka 消息 → Flink 设置定时器 → 到时间后发送回 Kafka。

ctx.timerService().registerProcessingTimeTimer(timestamp + 10000); // 10s后触发

适用于：大数据、实时计算场景。
优点：可以结合流处理做复杂逻辑。
缺点：需要 Flink 支持，架构要求高。

结论：如何选型？

方案	适用场景	优点	缺点
多个 Topic	固定延迟（5s、10s、30s）	简单易用	Topic 过多不好管理
数据库 + 定时任务	低吞吐、灵活时间	可靠性高	依赖数据库，效率受限
Redis ZSet	高并发、小量延迟消息	低延迟、性能高	Redis 容量受限
Kafka Streams	实时流处理	适配 Kafka 生态	学习成本较高
Flink Timer	大数据流处理	适配 Flink	需要 Flink

最佳实践

如果是常见的业务（延迟 5s、10s）：多 Topic 方案（简单易用）。
如果消息量大且时间灵活：Redis ZSet（高效）。
如果业务可靠性要求高：数据库 + 定时任务（可落地）。
如果是流式计算：Kafka Streams / Flink（强大但复杂）。

Kafka实现延迟消息

Kafka 实现延迟消息 Kafka 本身不支持原生的延迟消息（不像 RocketMQ 内置了延迟队列），但可以通过多种方式来实现延迟消息。常见的方案如下： 1. 使用不同的 Topic 分区（最常见） 思路： 创建多…...

编程日记 2025/9/17 16:40:16

大数据（7.2）Kafka万亿级数据洪流下的架构优化实战：从参数调优到集群治理

目录一、海量数据场景下的性能之殇1.1 互联网企业的数据增长曲线1.2 典型性能瓶颈分析二、生产者端极致优化2.1 批量发送黄金法则2.1.1 分区选择算法对比 2.2 序列化性能突破三、消费者端并发艺术3.1 多线程消费模式演进3.1.1 消费组Rebalance优化 3.2 位移管理高阶技巧四、…...

编程日记 2025/9/17 16:31:59

要查看 FAISS 使用的 OpenMP 版本，需根据安装方式和系统环境采用不同方法。以下是具体步骤和原理分析：

要查看 FAISS 使用的 OpenMP 版本，需根据安装方式和系统环境采用不同方法。以下是具体步骤和原理分析： 方法 1：通过库文件名称直接查看（推荐） FAISS 的 OpenMP 版本通常直接体现在其依赖的动态链接库（DLL/…...

编程日记 2025/9/17 16:32:01

AI 大模型的标准化工具箱MCP (Model Context Protocol)

MCP简介 MCP （Model Context Protocol，模型上下文协议）定义了应用程序和 AI 模型之间交换上下文信息的方式。这使得开发者能够以一致的方式将各种数据源、工具和功能连接到 AI 模型（一个中间协议层），就像 …...

编程日记 2025/9/17 16:31:06

哈希表的封装

目录引入哈希表封装修改哈希表参数修改哈希表成员修改%时使用的变量修改读取时获得的变量迭代器的实现迭代器的定义迭代器迭代器*解引用迭代器->成员访问迭代器重载和! 封装迭代器 HashTable迭代器封装非const版本 const版本 unordered_set迭…...

编程日记 2025/9/17 16:34:13

2025年认证杯数模竞赛赛题浅析-快速选题

赛题浅析认证杯作为国内最早的数学建模论坛、唯一一个全部公开参赛论文的竞赛、国内最大的数学建模竞赛之一、唯一一个对非学生群里开放的数学建模竞赛、国内唯二的支持高中生参赛的大学生数模竞赛。在数模界一直被视为国赛之前较好的练手赛，本文将初步简略得介绍…...

编程日记 2025/9/17 16:32:43

【网络安全】Linux 常见命令

未经许可，不得转载。文章目录正文系统信息查看用户与权限管理进程管理网络配置与检测文件操作日志查看与分析权限审计与安全检测正文在网络安全工作中，熟练掌握 Linux 系统中的常用命令，对于日常运维、日志分析、安全排查等工作至关重要。以下为常用命令汇总，供参考。…...

编程日记 2025/9/17 16:34:13

电脑卡顿严重怎么办电脑卡顿的处理指南

电脑突然卡顿比较严重，这是很多用户都曾经遇到过的问题，鼠标一直转圈圈，无法进行任何操作。电脑卡顿，电脑卡顿不仅会降低工作效率，还可能导致数据丢失，数据无法保存。很多用户解决电脑卡顿的方法就是直接一…...

编程日记 2025/9/17 16:40:13

山东大学软件学院创新项目实训开发日志（9）之测试前后端连接

在正式开始前后端功能开发前，在队友的帮助下，成功完成了前后端测试连接： 首先在后端编写一个测试相应程序： 然后在前端创建vue 并且在index.js中添加一下元素： 然后进行测试，测试成功： 后续可…...

编程日记 2025/9/17 16:28:32

H.264 NVMPI解码性能优化策略

H.264 NVMPI解码性能优化策略‌ ‌1. 硬件与驱动配置‌ ‌JetPack版本匹配‌：确保NVIDIA Jetson设备的JetPack SDK版本与CUDA驱动兼容，避免因驱动不匹配导致硬件解码性能下降‌8。‌显存分配优化‌：调整FFmpeg的-hwaccel_device参数指定GPU…...

编程日记 2025/9/17 16:31:07

汽车软件开发常用的需求管理工具汇总

目录往期推荐 DOORS（IBM ） 行业应用企业： 应用背景： 主要特点： Polarion ALM（Siemens） 行业应用企业： 应用背景： 主要特点： Codebeamer ALM&#x…...

编程日记 2025/9/17 16:34:13

如何从零构建一个自己的 CentOS 基础镜像

如何从零构建一个自己的 CentOS 基础镜像从零构建一个基于 CentOS 的基础镜像是一个很好的实践，可以帮助你理解 Docker 镜像的底层原理。以下是以 CentOS 为例，从零开始（不依赖现有镜像）构建基础镜像的详细步骤。我们将使用 yum…...

编程日记 2025/9/9 0:22:09

mongodb和clickhouse比较

好问题——MongoDB 也能处理这种高写入定期删除的时间序列场景，尤其从 MongoDB 5.0 开始支持了专门的 Time Series Collections（时间序列集合），对你的需求其实挺对口的。不过它有些优点和局限，需要具体分析下你场景…...

编程日记 2025/9/17 16:37:31

C#容器源码分析 --- List

List是一个非常常用的泛型集合类，它位于 System.Collections.Generic 命名空间下，本质上是一个动态数组，它提供了一系列方便的方法来管理和操作元素，例如添加、删除、查找等。与传统的数组相比，List可以根据需要动态调…...

编程日记 2025/9/17 16:35:55

以太坊区块大小的决定因素：深入解析区块 Gas 限制及其影响

以太坊（Ethereum）作为全球领先的区块链平台，其区块大小并非固定的物理尺寸，而是由区块 Gas 限制（Block Gas Limit）所决定。理解区块 Gas 限制及其影响因素，对于深入掌握以太坊网络的运行机制至…...

编程日记 2025/9/17 16:40:16

利用DeepFlow解决APISIX故障诊断中的方向偏差问题

概要：随着APISIX作为IT应用系统入口的普及，其故障定位能力的不足导致了在业务故障诊断中，APISIX常常成为首要的“嫌疑对象”。这不仅导致了“兴师动众”式的资源投入，还可能使诊断方向“背道而驰”，从而导致业务故障“…...

编程日记 2025/9/17 16:38:54

智慧养老实训基地建设方案：如何以科技赋能养老实操培训

在人口老龄化加剧的当下，智慧养老产业蓬勃发展，对专业技能型人才的需求愈发迫切。智慧养老实训基地建设意义非凡，它为培育具备实操能力与创新思维的养老人才搭建关键平台，有助于填补行业人才缺口，推动养老服务从传统模…...

编程日记 2025/9/17 16:35:08

基于AI的Web应用防火墙（AppWall）实战：漏洞拦截与威胁情报集成

摘要：针对Web应用面临的OWASP、CVE等漏洞攻击，本文结合群联AI云防护系统的AppWall模块，详解AI规则双引擎的防御原理，并提供漏洞拦截配置与威胁情报集成代码示例。一、Web应用安全挑战与AppWall优势传统WAF依赖规则库更新滞后&a…...

编程日记 2025/9/17 16:34:12

什么是采购管理？如何做好采购管理的持续优化？

你是不是也遇到过这种情况： 公司采购部刚换了新供应商，结果原材料质量忽高忽低，生产线上三天两头出状况；行政采购的办公用品，月初买回来月底就堆在仓库吃灰；财务部天天追着问采购成本怎么又超支了... 这些…...

编程日记 2025/9/12 16:01:00

根据鼠标位于屏幕的区域，设置弹窗锚点以及位置 public static void TipsPos(Transform tf) {//获取ui相机var uiCamera GetUICamera();var popup tf.GetComponent<RectTransform>();//获取鼠标位置Vector2 mousePos Input.mousePosition;float screenWidt…...

编程日记 2025/9/5 22:21:57

区块链知识点5-Solidity编程基础

1. 全局变量名具体描述 msg.sender 返回当前调用函数的调用者的地址 msg.value 当前消息所附带的以太币，单位为wei 2.变量的用法默认存储位置修饰符函数的返回值 memory 函数内部的局…...

编程日记 2025/9/6 22:45:18

OLAP与OLTP架构设计原理对比

OLAP与OLTP架构设计原理对比一、核心区别维度OLTPOLAP设计目标支持高并发、低延迟的事务操作（增删改查）支持复杂分析查询（聚合、多维度统计）数据模型规范化模型（3NF），减少冗余维度模型&…...

编程日记 2025/9/10 18:09:35

ubuntu20.04在mid360部署direct_lidar_odometry(DLO)

editor：1034Robotics-yy time：2025.4.10 1.下载DLO，mid360需要的一些...: 1.1 在工作空间/src下下载DLO： git clone https://github.com/vectr-ucla/direct_lidar_odometry 1.2 在工作空间/src下下载livox_ros_driver2&…...

编程日记 2025/9/5 20:34:37

检索增强生成（RAG）架构深度解析：突破大模型边界的工程实践

一、RAG技术架构设计哲学 1.1 范式演进：从静态模型到动态知识系统 graph LR A[传统LLM架构] -->|问题| B[依赖预训练参数] B --> C[知识固化风险] C --> D[领域适配困难]A -->|解决方案| E[RAG增强架构] E --> F[实时知识检索] F --> G[动态上下…...

编程日记 2025/9/15 1:09:46

线代第四课：行列式的性质

行列式性质转置行列式把行列式的第一行转置成第一列，使用表示如果在转置一下： 性质一： 行列地位相同，对行性质，对列性质性质二： 交换D的两行（列），D值变符号性…...

编程日记 2025/9/11 13:09:56

【语音识别】vLLM 部署 Whisper 语音识别模型指南

目录 1. 模型下载 2. 环境安装 3. 部署脚本 4. 服务测试语音识别技术在现代人工智能应用中扮演着重要角色，OpenAI开源的Whisper模型以其出色的识别准确率和多语言支持能力成为当前最先进的语音识别解决方案之一。本文将详细介绍如何使用vLLM（一个高…...

编程日记 2025/9/16 16:44:57

Python | kelvin波的水平空间结构

写在前面简单记录一下之前想画的一个图： 思路整体比较简单，两个子图，本质上就是一个带有投影，一个不带投影，通常用在EOF的空间模态和时间序列的绘制中，可以看看之前的几个详细的画法。 Python | El Ni…...

编程日记 2025/9/16 0:41:51

什么叫行列式

《行列式：数学中的重要概念及其应用》行列式是数学中的一个重要概念，主要用于描述线性方程组、向量空间等方面的性质。以下是关于它的详细介绍： 定义行列式是由排成正方形的一组数（称为元素）按照特定的规则计算得…...

编程日记 2025/9/13 22:48:13

构建高可用大数据平台：Hadoop与Spark分布式集群搭建指南

想象一下，你手握海量数据，却因为测试环境不稳定，频频遭遇宕机和数据丢失的噩梦。Hadoop和Spark作为大数据处理的“黄金搭档”，如何在分布式高可用（HA）环境下稳如磐石地运行？答案就在于一个精心构…...

编程日记 2025/9/13 3:16:32

[leetcode]211. 添加与搜索单词(Trie+DFS)

题目链接题意实现词典类 WordDictionary ： WordDictionary() 初始化词典对象void addWord(word) 将 word 添加到数据结构中，之后可以对它进行匹配bool search(word) 如果数据结构中存在字符串与 word 匹配，则返回 true ；否则…...

编程日记 2025/9/16 4:53:38

AI | 字节跳动 AI 中文IDE编辑器 Trae 初体验

Trae 简介与安装 🔦 什么是 Trae Trae 是大厂字节跳动出品的国内首个 AI IDE，深度理解中文开发场景。AI 高度集成于 IDE 环境之中，为你带来比 AI 插件更加流畅、准确、优质的开发体验。说是能够不用写代码，全靠一张嘴跟 AI 聊天…...

编程日记 2025/9/11 13:02:53

【开发经验】结合实际问题解决详述HTTPS通信过程

最近的开发调试过程中涉及到了HTTPS发送与接收，遇到实际问题才发现对这部分尚属于一知半解。结合实际问题的解决过程来详细整理以下HTTPS通信过程。需要调试的功能为BMC作为客户端向搭建好的Web服务器发送HTTPS请求，Web服务器负责接收处理发送过来的HT…...

编程日记 2025/9/9 19:56:30

灵霄破茧：仙途启幕 - 灵霄门新篇-(4)

重建之路，风云再起灵霄门内一片萧瑟，残垣断壁间弥漫着悲伤与凝重。弟子们忙碌地清理着战场，救治伤员，每个人的脸上都带着劫后余生的疲惫。陈霄日夜守在玄风真人的榻前，眼中满是自责与担忧。玄风真人的伤势极重&#…...

编程日记 2025/9/15 17:25:45

微信小程序事件绑定基本语法

微信小程序使用 bind 或 catch 前缀绑定事件，语法如下： <组件 bind事件名"处理函数" catch事件名"处理函数"></组件> bind：事件绑定，允许事件冒泡（向父组件传递）。 catc…...

编程日记 2025/9/8 13:06:28

vscode 连不上 Ubuntu 18 server 的解决方案

下载 vscode 历史版本 18.5（windows请装在系统盘 C 盘） 打开 vdcode，将自动更新设置为 None （很关键，否则容易前功尽弃） 重命名（删除） 服务器上的 .vscode-server 文件夹重新…...

编程日记 2025/9/12 13:17:30

OSPF接口的网络类型和不规则区域

网络类型(数据链路层所使用的协议所构建的二层网络类型) 1、MA --- 多点接入网络 BMA --- 支持广播的多点接入网络 NBMA --- 不支持广播的多点接入网络 2、P2P --- 点到点网络以太网 --- 以太网最主要的特点是需要基于MAC地址进行物理寻址，主要是因为以太网接口所连…...

编程日记 2025/9/15 17:58:58

基于Flask的勒索病毒应急响应平台架构设计与实践

基于Flask的勒索病毒应急响应平台架构设计与实践序言：安全工程师的防御视角作为从业十年的网络安全工程师，我深刻理解勒索病毒防御的黄金时间法则——应急响应速度每提升1分钟，数据恢复成功率将提高17%。本文介绍的应急响应平台&#xff…...

编程日记 2025/8/29 3:51:57

0410 | 软考高项笔记：项目管理概述

以下是不同组织结构中项目经理的角色、工作特点以及快速记忆的方法： 不同组织结构中项目经理的角色和工作特点组织结构项目经理的角色工作特点职能型组织项目协调者、辅助管理者权力有限，主要负责协调部门间的工作，项目成员向部门经理汇报…...

编程日记 2025/9/4 15:29:00

基于 Qt 的图片处理工具开发（一）：拖拽加载与基础图像处理功能实现

一、引言在桌面应用开发中，图片处理工具的核心挑战在于用户交互的流畅性和异常处理的健壮性。本文以 Qt为框架，深度解析如何实现一个支持拖拽加载、亮度调节、角度旋转的图片处理工具。通过严谨的文件格式校验、分层的架构设计和用户友好的交互逻辑&am…...

编程日记 2025/9/4 1:21:52

2025年4月通信科技领域周报（3.31-4.06）：6G技术加速落地与全连接生态构建

2025年4月通信科技领域周报（3.31-4.06）：6G技术加速落地与全连接生态构建目录一、本周热点回顾二、技术进展深度解析三、产业动态全景扫描四、行业生态与政策风向五、专业术语解释六、免责声明一、本周热点回顾 1. 华为发布6G全场景技术…...

编程日记 2025/9/9 10:54:45

Codeforces-CF816B-Karen and Coffee(差分/前缀和)

题目翻译： Karen 喜欢咖啡。她有 n 本食谱，第 i 本食谱包含两个数 li,ri，表示这本食谱推荐用 [li,ri] 之间的温度（包含 li.ri）来煮咖啡。 Karen 认为一个温度 a 是可接受的当且仅当有 ≥k 本食谱推荐用 …...

编程日记 2025/9/15 0:24:05

4.DJI-PSDK云台x-port控制：

DJI-PSDK云台x-port控制： X-Port 功能控制，即控制 X-Port 云台，头文件为 dji_xport.h 使用PSDK 的“云台控制”功能，开发者需要先设计负载设备的云台并开发出控制云台的程序，将云台的控制函数注册到PSDK 指定的接口后…...

编程日记 2025/9/9 5:01:44

大语言模型中的幻觉现象深度解析

一、幻觉的定义及出现的原因 1. 基本定义幻觉(Hallucination) 指大语言模型在自然语言处理过程中产生的与客观事实或既定输入相悖的响应，主要表现为信息失准与逻辑矛盾。 2. 幻觉类型与机制 2.1 事实性幻觉定义：生成内容与可验证…...

编程日记 2025/9/15 5:21:15

衣橱管理助手系统(衣服推荐系统)(springboot+ssm+vue+mysql)含运行文档

衣橱管理助手系统(衣服推荐系统)(springbootssmvuemysql)含运行文档该系统名为衣橱管理助手，是一个衣物搭配管理系统，主要功能包括衣物档案管理、衣物搭配推荐、搭配收藏以及套装智能推荐。用户可以通过系统进行衣物的搭配和收藏管理，系统提…...

编程日记 2025/9/13 5:53:26

视觉对象 - 数据可视化解读

Power BI 提供了丰富的视觉对象（Visuals），帮助用户以直观的方式呈现和分析数据。以下是 32 个常用视觉对象的解读及案例分享，涵盖核心功能、适用场景和注意事项。内容基于实际应用场景整理，便于快速理解。一、数据比较类视觉对象这类视觉对象主要用于比较不同类别、组别…...

编程日记 2025/9/9 21:16:47

使用物联网卡的烟感（NB-IoT/4G烟感）详解

基于物联网卡（NB-IoT/4G）的智能烟感是一种无线联网型火灾报警设备，相比传统烟感，它能够实时上报火警信息，适用于无人值守场所、智慧消防、远程监控等场景。 1. 物联网卡烟感的核心功能功能说明实时报警探测到烟雾后&…...

编程日记 2025/9/17 13:25:44

（2）网络学习之堡垒机

堡垒机和防火墙的区别： 1.功能定位防火墙主要负责抵御外部攻击，就像一道坚固的城墙，防止黑客进入内部网络。堡垒机则专注于内部管理，监控和记录运维人员的操作行为，确保内部网络的安全。 2.部署位置与作用范围防…...

编程日记 2025/9/9 11:41:34

FlinkSQL的常用语言

FlinkSQL 常用语言指南 FlinkSQL 是 Apache Flink 提供的 SQL 接口，允许用户使用标准 SQL 或扩展的 SQL 语法来处理流式和批式数据。以下是 FlinkSQL 的常用语言元素和操作： 基本查询 -- 选择查询 SELECT * FROM table_name;-- 带条件的查询 SELECT c…...

编程日记 2025/9/14 1:32:10