1. 为什么必须亲手创建Topic、亲手发消息——这不是“多此一举”而是生产环境的生存法则在某公司做消息中间件支撑的第三年我接手了一个线上告警频发的订单履约系统。表面看是消费延迟但排查三天后发现所有问题都指向一个被忽略的细节——Topic完全依赖RocketMQ自动创建机制。当某个新业务模块首次发送一条测试消息时Broker自动创建了Topic但默认配置是8个队列、本地磁盘存储、不开启Dledger高可用。而该业务实际QPS峰值超3000单机Broker扛不住消费组集体卡死。更麻烦的是自动创建的Topic无法设置消息保留时间messageRetentionTimeInMinutes日志盘三天就写满。后来我们紧急手动重建Topic调整队列数为16、启用Dledger、设置7天保留策略问题才彻底解决。这就是为什么标题里强调“手动创建Topic”和“同步异步发送”——它不是教程里的可选项而是压测报告、故障复盘、上线Checklist里白纸黑字的强制项。RocketMQ的Topic不是数据库里的表不能“用时再建”它更像高速公路的车道规划你得提前决定几条道、限速多少、应急车道在哪否则车一多必然堵死。同步发送适合强一致性场景比如支付结果回传必须等Broker确认落盘才返回成功异步发送则是高吞吐场景的命脉比如用户行为埋点允许短暂延迟但绝不能丢。很多团队踩坑不是因为不会写代码而是没想清楚这条消息到底值不值得我等它落盘如果Broker挂了我愿不愿意重试我的消费者能不能扛住突发流量这些问题的答案直接决定了你该用同步还是异步也决定了Topic的队列数、读写权限、是否开启Trace。所以这篇内容不讲“怎么调API”而是带你从运维视角、架构视角、故障视角重新理解一次Topic创建和消息发送——就像老司机教你怎么调后视镜、怎么看胎压、什么路况该降档一样全是实打实的保命经验。2. Topic手动创建不只是执行一条命令而是完成一次容量与容灾设计2.1 为什么自动创建Topic在生产环境是“定时炸弹”RocketMQ默认开启autoCreateTopicEnabletrue这在开发环境很友好写几行代码消息一发Topic自动出现。但生产环境里这个开关等于给系统埋雷。原因有三第一参数不可控。自动创建的Topic队列数defaultTopicQueueNums默认是8但8个队列意味着最大并发消费能力只有8个线程。如果你的业务每秒要处理5000条订单而消费者实例只有4台每台启动8个线程那理论最大消费速度是4×832条/秒——连零头都不到。更致命的是队列数一旦创建就无法修改想扩容只能删Topic重建而删除Topic会清空所有未消费消息线上谁敢这么干第二存储策略缺失。自动创建的Topic默认使用DEFAULT_BROKER的存储路径且不启用Dledger或主从同步。这意味着一旦Broker磁盘损坏消息永久丢失。而手动创建时你可以明确指定-c参数指向Dledger集群或通过-r参数设置读写队列分离让写入走SSD、读取走NVMe性能提升3倍以上。第三权限与治理真空。自动创建的Topic没有归属人、没有创建时间、没有业务描述。当监控告警显示某个Topic积压100万条时你得翻遍Git历史、问遍所有开发才能确定这是哪个微服务写的。而手动创建时-n参数可以填业务域如order_create_v2-o参数能加备注如“用于订单创建事件下游含风控、库存、物流三个系统”这些信息会写入NameServer元数据成为后续容量评估、下线决策的关键依据。提示某金融客户曾因自动创建Topic导致合规审计失败——监管要求所有消息主题必须登记业务负责人、数据分类分级、留存周期而自动Topic在元数据里只有AUTO_CREATE_TOPIC这个占位符根本无法满足。2.2 手动创建Topic的完整命令与参数深解手动创建Topic必须通过mqadmin工具这是RocketMQ官方提供的运维命令行。不要试图用Java SDK创建SDK的createTopic方法本质也是调用AdminExt接口但缺乏对底层存储策略的控制力。正确姿势是登录Broker所在服务器执行sh mqadmin updateTopic -n 192.168.1.100:9876 \ -b 192.168.1.101:10911 \ -t order_create_v2 \ -r 16 \ -w 16 \ -o 订单创建事件下游含风控、库存、物流 \ -c DefaultCluster \ -a READ,WRITE \ -p 604800逐参数拆解其真实含义-n 192.168.1.100:9876NameServer地址。注意不是Broker地址NameServer是元数据注册中心所有Topic信息都存在这里。如果集群有多个NameServer需用英文逗号分隔如-n 192.168.1.100:9876,192.168.1.101:9876确保元数据同步。-b 192.168.1.101:10911Broker地址。这里必须填具体的Broker IP和端口默认10911不能填localhost或127.0.0.1否则NameServer记录的Broker地址是错的消费者拉取路由时会失败。实测过一次填错IP导致整个集群路由混乱持续2小时。-t order_create_v2Topic名称。命名必须遵循[a-zA-Z0-9_-]{1,255}规则禁止中文、空格、特殊符号。建议采用业务域_功能_版本格式如user_login_v3方便按前缀批量管理。-r 16 -w 16读队列数readQueueNums和写队列数writeQueueNums。这是最关键的容量参数。计算公式是队列数 ≥ max(消费者实例数 × 每实例线程数, 生产者实例数 × 每实例线程数)。例如你有5台消费者机器每台启10个线程则最小需要50个读队列但RocketMQ要求读写队列数必须相等所以直接设为64向上取2的幂次方利于负载均衡。注意队列数不是越多越好超过128个队列会导致NameServer内存暴涨单节点建议≤256。-c DefaultCluster所属集群名。如果部署了多套RocketMQ集群如测试集群、预发集群、生产集群必须明确指定避免Topic被错误注册到其他集群。-a READ,WRITE访问权限。READ表示允许订阅WRITE表示允许发送。对于只供内部系统使用的Topic可设为READ,WRITE若需开放给第三方应单独建thirdparty_order_v1并设READ防止误发。-p 604800消息保留时间单位秒。604800秒7天。这是兜底保障即使消费者宕机一周消息也不会被自动清理。但要注意这个值不能设得过大否则磁盘空间压力剧增。我们通常按业务SLA设定核心交易类设7天日志类设3天实时计算类设1小时。注意执行后必须验证。用sh mqadmin topicList -n 192.168.1.100:9876 | grep order_create_v2确认Topic存在再用sh mqadmin topicStatus -n 192.168.1.100:9876 -t order_create_v2检查队列数、权限、存储路径是否符合预期。我见过太多人执行完命令就去喝咖啡结果发现-b参数填错了Broker端口填成10912导致Topic注册到不存在的Broker上白白浪费2小时。2.3 高级场景跨集群同步Topic与Dledger模式下的创建要点当你的架构升级到多活数据中心时Topic创建不再是单点操作。比如上海机房和北京机房各有一套RocketMQ集群需要保证order_create_v2在两地同时存在且配置一致。这时不能分别执行两次updateTopic因为NameServer之间不互通。正确方案是使用syncTopic命令sh mqadmin syncTopic -n shanghai-ns:9876 -t order_create_v2 -d beijing-ns:9876该命令会从上海NameServer拉取Topic元数据推送到北京NameServer并自动校验队列数、权限等字段。但要注意它只同步元数据不同步消息体。消息同步需依赖RocketMQ的Replica机制或自研的跨机房复制组件。更关键的是Dledger模式RocketMQ 4.5推荐的高可用方案。在Dledger集群中Broker不再以单点存在而是以broker-a、broker-b、broker-c组成一个Raft组。此时创建Topic必须指定-c参数为Dledger集群名如DledgerCluster且-b参数要填Dledger组内任意一个Broker地址。Dledger会自动将Topic元数据同步到组内所有节点。实测发现如果-b填的是非Leader节点命令会失败并提示Not a leader broker必须先用sh mqadmin brokerStatus -n ...查出当前Leader再填其地址。另一个坑是Dledger的存储路径。自动创建的Topic会存到store/commitlog下但Dledger要求所有Topic必须存到store/dledger目录。手动创建时需在Broker配置文件broker.conf中显式设置dLegerGroupbroker-a dLegerPeersn1-192.168.1.101:40911;n2-192.168.1.102:40911;n3-192.168.1.103:40911 dLegerSelfIdn1否则即使Topic创建成功Dledger也无法接管其CommitLog导致高可用失效。3. 消息发送实战同步、异步、单向的选型逻辑与代码陷阱3.1 同步发送不是“快”而是“稳”但稳的前提是你懂它的阻塞模型同步发送send()方法的典型代码长这样DefaultMQProducer producer new DefaultMQProducer(order_producer_group); producer.setNamesrvAddr(192.168.1.100:9876); producer.start(); Message msg new Message(order_create_v2, TAG_A, KEY_123, Hello RocketMQ.getBytes()); SendResult result producer.send(msg); System.out.println(发送状态 result.getSendStatus() 队列 result.getMessageQueue());表面看就三行但背后藏着三个必须直面的现实第一它是网络阻塞调用。producer.send(msg)会一直卡住直到Broker返回ACK或超时抛异常。默认超时时间是3秒sendMsgTimeout3000但如果你的Broker磁盘IO打满响应可能长达10秒。这时你的Web接口线程池会被迅速耗尽整个应用雪崩。解决方案不是调大超时而是必须设置合理的超时并捕获异常try { SendResult result producer.send(msg, 2000); // 显式设为2秒 if (result.getSendStatus() ! SendStatus.SEND_OK) { throw new RuntimeException(消息发送失败 result.getSendStatus()); } } catch (RemotingTooMuchRequestException e) { // Broker忙立即重试最多2次 Thread.sleep(100); producer.send(msg, 2000); } catch (Exception e) { // 记录日志走降级方案如写DB暂存 log.error(订单消息发送失败订单ID{}, orderId, e); fallbackToDatabase(orderId, msg); }第二它不保证消息100%不丢。即使SendStatus.SEND_OK也只是表示Broker已接收并写入PageCache还没刷盘。如果Broker进程崩溃且未开启flushDiskTypeSYNC_FLUSH消息仍会丢失。所以同步发送必须配合Broker配置flushDiskTypeSYNC_FLUSH同步刷盘brokerRoleSYNC_MASTER同步复制。这两个参数会让性能下降30%但换来的是金融级可靠性。第三它对Producer实例有状态要求。DefaultMQProducer是线程安全的但内部维护着与Broker的Netty连接池。如果你在Spring Boot里把它声明为Scope(prototype)每次new一个实例就会创建一堆无用连接最终触发Linux文件句柄耗尽。正确做法是全局单例用ServicePostConstruct初始化Service public class OrderProducer { private DefaultMQProducer producer; PostConstruct public void init() throws MQClientException { producer new DefaultMQProducer(order_producer_group); producer.setNamesrvAddr(192.168.1.100:9876); producer.setRetryTimesWhenSendFailed(2); // 发送失败重试2次 producer.setSendMsgTimeout(2000); producer.start(); } }实操心得某电商大促期间我们把sendMsgTimeout从3秒降到1秒结果发现超时率从0.02%升到0.5%但整体系统可用性反而提升——因为快速失败让线程能及时释放避免了线程池阻塞。这印证了一个原则在分布式系统里“快失败”比“慢成功”更重要。3.2 异步发送高吞吐的引擎但油门和刹车都得自己装异步发送sendAsync()是处理海量日志、埋点、通知的标配。代码结构是典型的回调模式producer.sendAsync(msg, new SendCallback() { Override public void onSuccess(SendResult sendResult) { log.info(消息发送成功队列{}, sendResult.getMessageQueue()); } Override public void onException(Throwable throwable) { log.error(消息发送失败, throwable); // 这里必须实现重试逻辑 retrySend(msg, 3); } });但新手常犯的致命错误是以为回调函数里写个log就完了。实际上onException触发的场景极多网络抖动、Broker重启、队列满、权限不足……而RocketMQ的异步发送默认不重试retryTimesWhenSendAsyncFailed0一旦失败消息就永远消失了。所以重试逻辑必须自己写且要带退避策略private void retrySend(Message msg, int maxRetries) { for (int i 0; i maxRetries; i) { try { SendResult result producer.send(msg, 2000); if (result.getSendStatus() SendStatus.SEND_OK) { return; // 成功则退出 } } catch (Exception e) { log.warn(第{}次重试失败, i 1, e); } // 指数退避100ms, 200ms, 400ms... try { Thread.sleep((long) Math.pow(2, i) * 100); } catch (InterruptedException ignored) {} } // 重试3次全失败写入本地DB待人工干预 saveToDeadLetterDB(msg); }另一个关键是异步发送的线程模型。sendAsync会把任务提交到Producer内部的asyncSenderThreadPool线程池默认大小是Runtime.getRuntime().availableProcessors()。如果业务QPS极高如每秒10万条埋点这个线程池可能成为瓶颈。我们曾遇到过线程池满导致RejectedExecutionException所有异步发送被拒绝。解决方案是显式设置线程池producer.setAsyncSenderThreadPoolQueueSize(10000); producer.setAsyncSenderThreadPoolCoreSize(32); producer.setAsyncSenderThreadPoolMaximumSize(64);最后提醒一个隐藏坑异步发送不支持事务消息。如果你需要“发消息更新DB”原子性必须用sendMessageInTransaction()它底层是同步调用。别试图在sendAsync回调里更新DB那只是应用层的巧合不是分布式事务。3.3 单向发送Oneway极致的快但快得毫无保障单向发送sendOneway()是最快的发送方式它连TCP ACK都不等发完就返回。适用场景极其有限只要求“尽力而为”且丢失可接受。比如服务器健康心跳、低优先级监控指标。// 发送心跳丢了就丢了下一秒再发 Message heartbeat new Message(sys_heartbeat, , System.currentTimeMillis() ); producer.sendOneway(heartbeat);但它有三个硬伤无任何返回值你无法知道消息是否到达Broker更别说队列、偏移量。无重试机制网络丢包、Broker宕机消息直接消失。不参与事务绝对不能用于业务核心链路。所以除非你的场景满足“丢了不影响业务”否则别碰sendOneway。我们曾有个团队用它发订单取消通知结果因网络抖动丢失了20%的消息导致下游库存系统多扣了货损失数万元。教训是快不是目的可控才是。4. 全链路验证与压测从“能发”到“稳发”的最后一公里4.1 四步验证法确保Topic和发送逻辑真正可靠光写完代码、创建好Topic还不够必须经过四层验证缺一不可第一步路由验证用sh mqadmin topicRoute -n 192.168.1.100:9876 -t order_create_v2检查Topic路由。正常输出应包含queueDatas列出所有队列如broker-a-0到broker-a-15确认数量为16brokerDatas显示Broker地址和角色如SYNC_MASTERfilterServerTable为空表示未启用过滤服务器正常。如果queueDatas为空说明Topic没创建成功或Broker未注册如果brokerDatas里角色是SLAVE说明你连到了从节点发送会失败。第二步生产者验证写一个最小化测试类只做三件事初始化Producer、发一条消息、打印SendResult。重点观察SendStatus是否为SEND_OKmessageQueue是否落在预期Broker上如broker-aqueueOffset是否递增证明消息确实写入。第三步消费者验证启动一个简单Consumer订阅该Topic打印收到的消息。关键看是否能实时收到延迟100msgetMsgId()是否唯一排除重复消费getKeys()是否匹配发送时的KEY验证消息属性传递。第四步异常注入验证这才是真功夫。用工具模拟故障用iptables禁用Broker端口iptables -A OUTPUT -p tcp --dport 10911 -j DROP验证Producer是否快速失败并走降级用dd写满Broker磁盘dd if/dev/zero of/tmp/fill bs1M count2000验证SendResult是否返回FLUSH_DISK_TIMEOUT重启NameServer观察Producer是否自动切换到备用NameServer需配置多个-n。注意某次验证时我们发现Consumer在Broker重启后有3分钟收不到消息。排查发现是pullInterval参数默认是0即Broker主动推送但推送通道断开后Consumer没及时切到长轮询。解决方案是显式设置consumer.setPullInterval(1000)强制每秒拉一次确保快速恢复。4.2 压测脚本编写用真实流量检验你的设计光靠手工验证不够必须用JMeter或自研压测工具模拟真实流量。我们用Python写了一个轻量压测脚本核心逻辑如下import time from rocketmq.client import Producer, Message def stress_test(): producer Producer(stress_group) producer.set_namesrv_addr(192.168.1.100:9876) producer.start() start_time time.time() success_count 0 fail_count 0 for i in range(10000): # 发1万条 msg Message(order_create_v2, TAG_A, fKEY_{i}, btest_data) try: result producer.send_sync(msg, timeout2000) if result.status 0: # SEND_OK success_count 1 else: fail_count 1 except Exception as e: fail_count 1 # 控制QPS为500 if i % 500 0: time.sleep(1) end_time time.time() print(f总耗时{end_time - start_time:.2f}s成功率{success_count/(success_countfail_count)*100:.2f}%)压测时重点关注三个指标指标健康阈值超标含义应对措施平均RT 20msBroker IO或网络瓶颈检查磁盘IOPS、网卡中断失败率 0.1%网络不稳定或Broker过载增加Broker节点、优化网络队列积压 1000条消费者处理慢扩容消费者、优化消费逻辑我们曾用此脚本压测发现当QPS从1000升到2000时失败率从0.05%飙升至5%。最终定位到是Broker的nettyWorkGroupThreads默认值太小8导致Netty线程处理不过来。将其调至32后问题解决。4.3 监控大盘搭建让问题在用户投诉前暴露生产环境必须有监控否则就是裸奔。我们基于PrometheusGrafana搭建了RocketMQ监控大盘核心指标包括生产侧rocketmq_producer_send_latency_milliseconds发送延迟P99、rocketmq_producer_send_failed_total发送失败数、rocketmq_producer_client_runtime客户端运行时长Broker侧rocketmq_broker_commitlog_flush_latency_milliseconds刷盘延迟、rocketmq_broker_put_message_total每秒写入数、rocketmq_broker_get_message_total每秒读取数消费侧rocketmq_consumer_offset_diff消费延迟即commitlog_max_offset - consumer_offset、rocketmq_consumer_pull_failed_total拉取消息失败数。特别重要的是offset_diff。我们设置了告警规则当max by (topic) (rocketmq_consumer_offset_diff) 10000持续5分钟立即触发企业微信告警。这个阈值是根据业务SLA定的——订单系统要求消费延迟1秒按QPS 1000计算1秒最多积压1000条所以设10000是留足缓冲。实操心得监控不是摆设。某次凌晨offset_diff突增到50万但告警没响。排查发现是Grafana的告警规则里用了sum()聚合而实际应该用max()——因为某个消费者实例卡死其他实例正常sum()把异常值平均掉了。从此我们所有告警规则都加了by (topic, group)标签确保精准定位。5. 常见问题与独家排障技巧那些文档里不会写的血泪教训5.1 “SendResult为空”先查这三个地方producer.send()返回null这是最让人抓狂的问题。别急着重装RocketMQ按顺序查第一查Producer是否start()这是90%的新手错误。DefaultMQProducer必须调用start()才能初始化网络连接。如果忘了这行send()会直接返回null且不报任何异常。解决方案在init()方法末尾加日志producer.start(); log.info(Producer started, clientID: {}, producer.getClientId());如果日志没打印说明start()根本没执行。第二查Nameserver地址是否可达用telnet 192.168.1.100 9876测试连通性。如果超时检查Nameserver进程是否在运行ps -ef | grep namesrv防火墙是否放行9876端口iptables -L -n | grep 9876DNS是否解析正确nslookup your-ns-domain。第三查Topic是否存在且权限正确用sh mqadmin topicList -n ...确认Topic存在再用sh mqadmin topicStatus -n ... -t topic_name检查perm字段。如果是6二进制110表示WRITE,-READ,SUB允许发送如果是4100表示只读发送必失败。独家技巧在Producer代码里加一行System.setProperty(rocketmq.client.logUseSlf4j, true)然后配置logback.xml输出DEBUG日志。当send()返回null时日志里会打印No route info of this topic直接定位到Topic问题。5.2 “消息重复消费”不是RocketMQ的锅是你的代码没防重RocketMQ官方文档说“不保证消息不重复”但很多人以为这是Bug。其实这是分布式系统的必然——网络分区时Producer可能收到超时但Broker其实已写入Producer重试又发一遍。真正的解法是业务层幂等。我们通用的幂等方案是“唯一键状态机”每条消息带keys如订单ID消费时先查DB如果该订单状态已是“已创建”直接return如果不存在用INSERT IGNORE插入订单记录再更新状态。-- 创建幂等表 CREATE TABLE msg_dedup ( msg_key VARCHAR(64) PRIMARY KEY, status TINYINT DEFAULT 0, -- 0:处理中, 1:成功, 2:失败 created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- 消费逻辑伪代码 if (insertIgnore(msg_dedup, orderId, 0) 0) { // 已存在跳过 return; } // 处理业务逻辑 processOrder(orderId); // 更新状态为成功 update msg_dedup set status1 where msg_keyorderId;注意不要用Redis做幂等因为Redis可能主从延迟导致重复判断。DB的强一致性才是保障。5.3 “消费延迟飙升”90%的情况是消费者线程数配错了消费延迟offset_diff突然涨到百万级第一反应不是Broker有问题而是看消费者配置线程数consumer.setConsumeThreadMin(20)和consumer.setConsumeThreadMax(20)必须设为相同值避免动态扩缩容带来的抖动拉取大小consumer.setPullBatchSize(32)太小如8导致频繁拉取网络开销大太大如128导致单次处理时间长延迟增加流控consumer.setPullThresholdForQueue(1000)限制每个队列最多缓存1000条防止内存溢出。我们曾遇到一个案例消费者线程数设为1但Topic有16个队列。结果16个队列的消息全挤在一个线程里串行处理延迟自然爆炸。改成20个线程后延迟从小时级降到毫秒级。5.4 故障速查表5分钟定位80%的问题现象可能原因快速验证命令解决方案send()返回nullProducer未startps -ef | grep java看进程加producer.start()消费者收不到消息Topic未创建或权限不足sh mqadmin topicList -n ...用updateTopic创建发送失败率高Broker磁盘IO满iostat -x 1 3看%util清理store/abort文件扩容磁盘消费延迟持续增长消费者处理逻辑阻塞jstack pid | grep CONSUME优化DB查询加索引NameServer无响应内存溢出jstat -gc pid看OU调大-Xmx如-Xmx4g最后分享一个小技巧所有RocketMQ命令都支持-h查看帮助比如sh mqadmin updateTopic -h。但更实用的是看源码里的org.apache.rocketmq.tools.admin.DefaultMQAdminExtImpl类里面每个方法的注释都写了参数含义和默认值比官网文档还准。我在实际项目中发现80%的RocketMQ问题根源不在代码而在对Topic生命周期和消息语义的理解偏差。当你亲手创建Topic时你是在画一张容量地图当你选择同步或异步发送时你是在为业务SLA投票。这些决策没有标准答案只有结合你的数据量、一致性要求、运维能力后的最优解。所以别迷信“最佳实践”多问问自己如果现在Broker挂了我的用户会看到什么这个答案比任何技术文档都重要。
1. 为什么必须亲手创建Topic、亲手发消息——这不是“多此一举”而是生产环境的生存法则在某公司做消息中间件支撑的第三年我接手了一个线上告警频发的订单履约系统。表面看是消费延迟但排查三天后发现所有问题都指向一个被忽略的细节——Topic完全依赖RocketMQ自动创建机制。当某个新业务模块首次发送一条测试消息时Broker自动创建了Topic但默认配置是8个队列、本地磁盘存储、不开启Dledger高可用。而该业务实际QPS峰值超3000单机Broker扛不住消费组集体卡死。更麻烦的是自动创建的Topic无法设置消息保留时间messageRetentionTimeInMinutes日志盘三天就写满。后来我们紧急手动重建Topic调整队列数为16、启用Dledger、设置7天保留策略问题才彻底解决。这就是为什么标题里强调“手动创建Topic”和“同步异步发送”——它不是教程里的可选项而是压测报告、故障复盘、上线Checklist里白纸黑字的强制项。RocketMQ的Topic不是数据库里的表不能“用时再建”它更像高速公路的车道规划你得提前决定几条道、限速多少、应急车道在哪否则车一多必然堵死。同步发送适合强一致性场景比如支付结果回传必须等Broker确认落盘才返回成功异步发送则是高吞吐场景的命脉比如用户行为埋点允许短暂延迟但绝不能丢。很多团队踩坑不是因为不会写代码而是没想清楚这条消息到底值不值得我等它落盘如果Broker挂了我愿不愿意重试我的消费者能不能扛住突发流量这些问题的答案直接决定了你该用同步还是异步也决定了Topic的队列数、读写权限、是否开启Trace。所以这篇内容不讲“怎么调API”而是带你从运维视角、架构视角、故障视角重新理解一次Topic创建和消息发送——就像老司机教你怎么调后视镜、怎么看胎压、什么路况该降档一样全是实打实的保命经验。2. Topic手动创建不只是执行一条命令而是完成一次容量与容灾设计2.1 为什么自动创建Topic在生产环境是“定时炸弹”RocketMQ默认开启autoCreateTopicEnabletrue这在开发环境很友好写几行代码消息一发Topic自动出现。但生产环境里这个开关等于给系统埋雷。原因有三第一参数不可控。自动创建的Topic队列数defaultTopicQueueNums默认是8但8个队列意味着最大并发消费能力只有8个线程。如果你的业务每秒要处理5000条订单而消费者实例只有4台每台启动8个线程那理论最大消费速度是4×832条/秒——连零头都不到。更致命的是队列数一旦创建就无法修改想扩容只能删Topic重建而删除Topic会清空所有未消费消息线上谁敢这么干第二存储策略缺失。自动创建的Topic默认使用DEFAULT_BROKER的存储路径且不启用Dledger或主从同步。这意味着一旦Broker磁盘损坏消息永久丢失。而手动创建时你可以明确指定-c参数指向Dledger集群或通过-r参数设置读写队列分离让写入走SSD、读取走NVMe性能提升3倍以上。第三权限与治理真空。自动创建的Topic没有归属人、没有创建时间、没有业务描述。当监控告警显示某个Topic积压100万条时你得翻遍Git历史、问遍所有开发才能确定这是哪个微服务写的。而手动创建时-n参数可以填业务域如order_create_v2-o参数能加备注如“用于订单创建事件下游含风控、库存、物流三个系统”这些信息会写入NameServer元数据成为后续容量评估、下线决策的关键依据。提示某金融客户曾因自动创建Topic导致合规审计失败——监管要求所有消息主题必须登记业务负责人、数据分类分级、留存周期而自动Topic在元数据里只有AUTO_CREATE_TOPIC这个占位符根本无法满足。2.2 手动创建Topic的完整命令与参数深解手动创建Topic必须通过mqadmin工具这是RocketMQ官方提供的运维命令行。不要试图用Java SDK创建SDK的createTopic方法本质也是调用AdminExt接口但缺乏对底层存储策略的控制力。正确姿势是登录Broker所在服务器执行sh mqadmin updateTopic -n 192.168.1.100:9876 \ -b 192.168.1.101:10911 \ -t order_create_v2 \ -r 16 \ -w 16 \ -o 订单创建事件下游含风控、库存、物流 \ -c DefaultCluster \ -a READ,WRITE \ -p 604800逐参数拆解其真实含义-n 192.168.1.100:9876NameServer地址。注意不是Broker地址NameServer是元数据注册中心所有Topic信息都存在这里。如果集群有多个NameServer需用英文逗号分隔如-n 192.168.1.100:9876,192.168.1.101:9876确保元数据同步。-b 192.168.1.101:10911Broker地址。这里必须填具体的Broker IP和端口默认10911不能填localhost或127.0.0.1否则NameServer记录的Broker地址是错的消费者拉取路由时会失败。实测过一次填错IP导致整个集群路由混乱持续2小时。-t order_create_v2Topic名称。命名必须遵循[a-zA-Z0-9_-]{1,255}规则禁止中文、空格、特殊符号。建议采用业务域_功能_版本格式如user_login_v3方便按前缀批量管理。-r 16 -w 16读队列数readQueueNums和写队列数writeQueueNums。这是最关键的容量参数。计算公式是队列数 ≥ max(消费者实例数 × 每实例线程数, 生产者实例数 × 每实例线程数)。例如你有5台消费者机器每台启10个线程则最小需要50个读队列但RocketMQ要求读写队列数必须相等所以直接设为64向上取2的幂次方利于负载均衡。注意队列数不是越多越好超过128个队列会导致NameServer内存暴涨单节点建议≤256。-c DefaultCluster所属集群名。如果部署了多套RocketMQ集群如测试集群、预发集群、生产集群必须明确指定避免Topic被错误注册到其他集群。-a READ,WRITE访问权限。READ表示允许订阅WRITE表示允许发送。对于只供内部系统使用的Topic可设为READ,WRITE若需开放给第三方应单独建thirdparty_order_v1并设READ防止误发。-p 604800消息保留时间单位秒。604800秒7天。这是兜底保障即使消费者宕机一周消息也不会被自动清理。但要注意这个值不能设得过大否则磁盘空间压力剧增。我们通常按业务SLA设定核心交易类设7天日志类设3天实时计算类设1小时。注意执行后必须验证。用sh mqadmin topicList -n 192.168.1.100:9876 | grep order_create_v2确认Topic存在再用sh mqadmin topicStatus -n 192.168.1.100:9876 -t order_create_v2检查队列数、权限、存储路径是否符合预期。我见过太多人执行完命令就去喝咖啡结果发现-b参数填错了Broker端口填成10912导致Topic注册到不存在的Broker上白白浪费2小时。2.3 高级场景跨集群同步Topic与Dledger模式下的创建要点当你的架构升级到多活数据中心时Topic创建不再是单点操作。比如上海机房和北京机房各有一套RocketMQ集群需要保证order_create_v2在两地同时存在且配置一致。这时不能分别执行两次updateTopic因为NameServer之间不互通。正确方案是使用syncTopic命令sh mqadmin syncTopic -n shanghai-ns:9876 -t order_create_v2 -d beijing-ns:9876该命令会从上海NameServer拉取Topic元数据推送到北京NameServer并自动校验队列数、权限等字段。但要注意它只同步元数据不同步消息体。消息同步需依赖RocketMQ的Replica机制或自研的跨机房复制组件。更关键的是Dledger模式RocketMQ 4.5推荐的高可用方案。在Dledger集群中Broker不再以单点存在而是以broker-a、broker-b、broker-c组成一个Raft组。此时创建Topic必须指定-c参数为Dledger集群名如DledgerCluster且-b参数要填Dledger组内任意一个Broker地址。Dledger会自动将Topic元数据同步到组内所有节点。实测发现如果-b填的是非Leader节点命令会失败并提示Not a leader broker必须先用sh mqadmin brokerStatus -n ...查出当前Leader再填其地址。另一个坑是Dledger的存储路径。自动创建的Topic会存到store/commitlog下但Dledger要求所有Topic必须存到store/dledger目录。手动创建时需在Broker配置文件broker.conf中显式设置dLegerGroupbroker-a dLegerPeersn1-192.168.1.101:40911;n2-192.168.1.102:40911;n3-192.168.1.103:40911 dLegerSelfIdn1否则即使Topic创建成功Dledger也无法接管其CommitLog导致高可用失效。3. 消息发送实战同步、异步、单向的选型逻辑与代码陷阱3.1 同步发送不是“快”而是“稳”但稳的前提是你懂它的阻塞模型同步发送send()方法的典型代码长这样DefaultMQProducer producer new DefaultMQProducer(order_producer_group); producer.setNamesrvAddr(192.168.1.100:9876); producer.start(); Message msg new Message(order_create_v2, TAG_A, KEY_123, Hello RocketMQ.getBytes()); SendResult result producer.send(msg); System.out.println(发送状态 result.getSendStatus() 队列 result.getMessageQueue());表面看就三行但背后藏着三个必须直面的现实第一它是网络阻塞调用。producer.send(msg)会一直卡住直到Broker返回ACK或超时抛异常。默认超时时间是3秒sendMsgTimeout3000但如果你的Broker磁盘IO打满响应可能长达10秒。这时你的Web接口线程池会被迅速耗尽整个应用雪崩。解决方案不是调大超时而是必须设置合理的超时并捕获异常try { SendResult result producer.send(msg, 2000); // 显式设为2秒 if (result.getSendStatus() ! SendStatus.SEND_OK) { throw new RuntimeException(消息发送失败 result.getSendStatus()); } } catch (RemotingTooMuchRequestException e) { // Broker忙立即重试最多2次 Thread.sleep(100); producer.send(msg, 2000); } catch (Exception e) { // 记录日志走降级方案如写DB暂存 log.error(订单消息发送失败订单ID{}, orderId, e); fallbackToDatabase(orderId, msg); }第二它不保证消息100%不丢。即使SendStatus.SEND_OK也只是表示Broker已接收并写入PageCache还没刷盘。如果Broker进程崩溃且未开启flushDiskTypeSYNC_FLUSH消息仍会丢失。所以同步发送必须配合Broker配置flushDiskTypeSYNC_FLUSH同步刷盘brokerRoleSYNC_MASTER同步复制。这两个参数会让性能下降30%但换来的是金融级可靠性。第三它对Producer实例有状态要求。DefaultMQProducer是线程安全的但内部维护着与Broker的Netty连接池。如果你在Spring Boot里把它声明为Scope(prototype)每次new一个实例就会创建一堆无用连接最终触发Linux文件句柄耗尽。正确做法是全局单例用ServicePostConstruct初始化Service public class OrderProducer { private DefaultMQProducer producer; PostConstruct public void init() throws MQClientException { producer new DefaultMQProducer(order_producer_group); producer.setNamesrvAddr(192.168.1.100:9876); producer.setRetryTimesWhenSendFailed(2); // 发送失败重试2次 producer.setSendMsgTimeout(2000); producer.start(); } }实操心得某电商大促期间我们把sendMsgTimeout从3秒降到1秒结果发现超时率从0.02%升到0.5%但整体系统可用性反而提升——因为快速失败让线程能及时释放避免了线程池阻塞。这印证了一个原则在分布式系统里“快失败”比“慢成功”更重要。3.2 异步发送高吞吐的引擎但油门和刹车都得自己装异步发送sendAsync()是处理海量日志、埋点、通知的标配。代码结构是典型的回调模式producer.sendAsync(msg, new SendCallback() { Override public void onSuccess(SendResult sendResult) { log.info(消息发送成功队列{}, sendResult.getMessageQueue()); } Override public void onException(Throwable throwable) { log.error(消息发送失败, throwable); // 这里必须实现重试逻辑 retrySend(msg, 3); } });但新手常犯的致命错误是以为回调函数里写个log就完了。实际上onException触发的场景极多网络抖动、Broker重启、队列满、权限不足……而RocketMQ的异步发送默认不重试retryTimesWhenSendAsyncFailed0一旦失败消息就永远消失了。所以重试逻辑必须自己写且要带退避策略private void retrySend(Message msg, int maxRetries) { for (int i 0; i maxRetries; i) { try { SendResult result producer.send(msg, 2000); if (result.getSendStatus() SendStatus.SEND_OK) { return; // 成功则退出 } } catch (Exception e) { log.warn(第{}次重试失败, i 1, e); } // 指数退避100ms, 200ms, 400ms... try { Thread.sleep((long) Math.pow(2, i) * 100); } catch (InterruptedException ignored) {} } // 重试3次全失败写入本地DB待人工干预 saveToDeadLetterDB(msg); }另一个关键是异步发送的线程模型。sendAsync会把任务提交到Producer内部的asyncSenderThreadPool线程池默认大小是Runtime.getRuntime().availableProcessors()。如果业务QPS极高如每秒10万条埋点这个线程池可能成为瓶颈。我们曾遇到过线程池满导致RejectedExecutionException所有异步发送被拒绝。解决方案是显式设置线程池producer.setAsyncSenderThreadPoolQueueSize(10000); producer.setAsyncSenderThreadPoolCoreSize(32); producer.setAsyncSenderThreadPoolMaximumSize(64);最后提醒一个隐藏坑异步发送不支持事务消息。如果你需要“发消息更新DB”原子性必须用sendMessageInTransaction()它底层是同步调用。别试图在sendAsync回调里更新DB那只是应用层的巧合不是分布式事务。3.3 单向发送Oneway极致的快但快得毫无保障单向发送sendOneway()是最快的发送方式它连TCP ACK都不等发完就返回。适用场景极其有限只要求“尽力而为”且丢失可接受。比如服务器健康心跳、低优先级监控指标。// 发送心跳丢了就丢了下一秒再发 Message heartbeat new Message(sys_heartbeat, , System.currentTimeMillis() ); producer.sendOneway(heartbeat);但它有三个硬伤无任何返回值你无法知道消息是否到达Broker更别说队列、偏移量。无重试机制网络丢包、Broker宕机消息直接消失。不参与事务绝对不能用于业务核心链路。所以除非你的场景满足“丢了不影响业务”否则别碰sendOneway。我们曾有个团队用它发订单取消通知结果因网络抖动丢失了20%的消息导致下游库存系统多扣了货损失数万元。教训是快不是目的可控才是。4. 全链路验证与压测从“能发”到“稳发”的最后一公里4.1 四步验证法确保Topic和发送逻辑真正可靠光写完代码、创建好Topic还不够必须经过四层验证缺一不可第一步路由验证用sh mqadmin topicRoute -n 192.168.1.100:9876 -t order_create_v2检查Topic路由。正常输出应包含queueDatas列出所有队列如broker-a-0到broker-a-15确认数量为16brokerDatas显示Broker地址和角色如SYNC_MASTERfilterServerTable为空表示未启用过滤服务器正常。如果queueDatas为空说明Topic没创建成功或Broker未注册如果brokerDatas里角色是SLAVE说明你连到了从节点发送会失败。第二步生产者验证写一个最小化测试类只做三件事初始化Producer、发一条消息、打印SendResult。重点观察SendStatus是否为SEND_OKmessageQueue是否落在预期Broker上如broker-aqueueOffset是否递增证明消息确实写入。第三步消费者验证启动一个简单Consumer订阅该Topic打印收到的消息。关键看是否能实时收到延迟100msgetMsgId()是否唯一排除重复消费getKeys()是否匹配发送时的KEY验证消息属性传递。第四步异常注入验证这才是真功夫。用工具模拟故障用iptables禁用Broker端口iptables -A OUTPUT -p tcp --dport 10911 -j DROP验证Producer是否快速失败并走降级用dd写满Broker磁盘dd if/dev/zero of/tmp/fill bs1M count2000验证SendResult是否返回FLUSH_DISK_TIMEOUT重启NameServer观察Producer是否自动切换到备用NameServer需配置多个-n。注意某次验证时我们发现Consumer在Broker重启后有3分钟收不到消息。排查发现是pullInterval参数默认是0即Broker主动推送但推送通道断开后Consumer没及时切到长轮询。解决方案是显式设置consumer.setPullInterval(1000)强制每秒拉一次确保快速恢复。4.2 压测脚本编写用真实流量检验你的设计光靠手工验证不够必须用JMeter或自研压测工具模拟真实流量。我们用Python写了一个轻量压测脚本核心逻辑如下import time from rocketmq.client import Producer, Message def stress_test(): producer Producer(stress_group) producer.set_namesrv_addr(192.168.1.100:9876) producer.start() start_time time.time() success_count 0 fail_count 0 for i in range(10000): # 发1万条 msg Message(order_create_v2, TAG_A, fKEY_{i}, btest_data) try: result producer.send_sync(msg, timeout2000) if result.status 0: # SEND_OK success_count 1 else: fail_count 1 except Exception as e: fail_count 1 # 控制QPS为500 if i % 500 0: time.sleep(1) end_time time.time() print(f总耗时{end_time - start_time:.2f}s成功率{success_count/(success_countfail_count)*100:.2f}%)压测时重点关注三个指标指标健康阈值超标含义应对措施平均RT 20msBroker IO或网络瓶颈检查磁盘IOPS、网卡中断失败率 0.1%网络不稳定或Broker过载增加Broker节点、优化网络队列积压 1000条消费者处理慢扩容消费者、优化消费逻辑我们曾用此脚本压测发现当QPS从1000升到2000时失败率从0.05%飙升至5%。最终定位到是Broker的nettyWorkGroupThreads默认值太小8导致Netty线程处理不过来。将其调至32后问题解决。4.3 监控大盘搭建让问题在用户投诉前暴露生产环境必须有监控否则就是裸奔。我们基于PrometheusGrafana搭建了RocketMQ监控大盘核心指标包括生产侧rocketmq_producer_send_latency_milliseconds发送延迟P99、rocketmq_producer_send_failed_total发送失败数、rocketmq_producer_client_runtime客户端运行时长Broker侧rocketmq_broker_commitlog_flush_latency_milliseconds刷盘延迟、rocketmq_broker_put_message_total每秒写入数、rocketmq_broker_get_message_total每秒读取数消费侧rocketmq_consumer_offset_diff消费延迟即commitlog_max_offset - consumer_offset、rocketmq_consumer_pull_failed_total拉取消息失败数。特别重要的是offset_diff。我们设置了告警规则当max by (topic) (rocketmq_consumer_offset_diff) 10000持续5分钟立即触发企业微信告警。这个阈值是根据业务SLA定的——订单系统要求消费延迟1秒按QPS 1000计算1秒最多积压1000条所以设10000是留足缓冲。实操心得监控不是摆设。某次凌晨offset_diff突增到50万但告警没响。排查发现是Grafana的告警规则里用了sum()聚合而实际应该用max()——因为某个消费者实例卡死其他实例正常sum()把异常值平均掉了。从此我们所有告警规则都加了by (topic, group)标签确保精准定位。5. 常见问题与独家排障技巧那些文档里不会写的血泪教训5.1 “SendResult为空”先查这三个地方producer.send()返回null这是最让人抓狂的问题。别急着重装RocketMQ按顺序查第一查Producer是否start()这是90%的新手错误。DefaultMQProducer必须调用start()才能初始化网络连接。如果忘了这行send()会直接返回null且不报任何异常。解决方案在init()方法末尾加日志producer.start(); log.info(Producer started, clientID: {}, producer.getClientId());如果日志没打印说明start()根本没执行。第二查Nameserver地址是否可达用telnet 192.168.1.100 9876测试连通性。如果超时检查Nameserver进程是否在运行ps -ef | grep namesrv防火墙是否放行9876端口iptables -L -n | grep 9876DNS是否解析正确nslookup your-ns-domain。第三查Topic是否存在且权限正确用sh mqadmin topicList -n ...确认Topic存在再用sh mqadmin topicStatus -n ... -t topic_name检查perm字段。如果是6二进制110表示WRITE,-READ,SUB允许发送如果是4100表示只读发送必失败。独家技巧在Producer代码里加一行System.setProperty(rocketmq.client.logUseSlf4j, true)然后配置logback.xml输出DEBUG日志。当send()返回null时日志里会打印No route info of this topic直接定位到Topic问题。5.2 “消息重复消费”不是RocketMQ的锅是你的代码没防重RocketMQ官方文档说“不保证消息不重复”但很多人以为这是Bug。其实这是分布式系统的必然——网络分区时Producer可能收到超时但Broker其实已写入Producer重试又发一遍。真正的解法是业务层幂等。我们通用的幂等方案是“唯一键状态机”每条消息带keys如订单ID消费时先查DB如果该订单状态已是“已创建”直接return如果不存在用INSERT IGNORE插入订单记录再更新状态。-- 创建幂等表 CREATE TABLE msg_dedup ( msg_key VARCHAR(64) PRIMARY KEY, status TINYINT DEFAULT 0, -- 0:处理中, 1:成功, 2:失败 created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- 消费逻辑伪代码 if (insertIgnore(msg_dedup, orderId, 0) 0) { // 已存在跳过 return; } // 处理业务逻辑 processOrder(orderId); // 更新状态为成功 update msg_dedup set status1 where msg_keyorderId;注意不要用Redis做幂等因为Redis可能主从延迟导致重复判断。DB的强一致性才是保障。5.3 “消费延迟飙升”90%的情况是消费者线程数配错了消费延迟offset_diff突然涨到百万级第一反应不是Broker有问题而是看消费者配置线程数consumer.setConsumeThreadMin(20)和consumer.setConsumeThreadMax(20)必须设为相同值避免动态扩缩容带来的抖动拉取大小consumer.setPullBatchSize(32)太小如8导致频繁拉取网络开销大太大如128导致单次处理时间长延迟增加流控consumer.setPullThresholdForQueue(1000)限制每个队列最多缓存1000条防止内存溢出。我们曾遇到一个案例消费者线程数设为1但Topic有16个队列。结果16个队列的消息全挤在一个线程里串行处理延迟自然爆炸。改成20个线程后延迟从小时级降到毫秒级。5.4 故障速查表5分钟定位80%的问题现象可能原因快速验证命令解决方案send()返回nullProducer未startps -ef | grep java看进程加producer.start()消费者收不到消息Topic未创建或权限不足sh mqadmin topicList -n ...用updateTopic创建发送失败率高Broker磁盘IO满iostat -x 1 3看%util清理store/abort文件扩容磁盘消费延迟持续增长消费者处理逻辑阻塞jstack pid | grep CONSUME优化DB查询加索引NameServer无响应内存溢出jstat -gc pid看OU调大-Xmx如-Xmx4g最后分享一个小技巧所有RocketMQ命令都支持-h查看帮助比如sh mqadmin updateTopic -h。但更实用的是看源码里的org.apache.rocketmq.tools.admin.DefaultMQAdminExtImpl类里面每个方法的注释都写了参数含义和默认值比官网文档还准。我在实际项目中发现80%的RocketMQ问题根源不在代码而在对Topic生命周期和消息语义的理解偏差。当你亲手创建Topic时你是在画一张容量地图当你选择同步或异步发送时你是在为业务SLA投票。这些决策没有标准答案只有结合你的数据量、一致性要求、运维能力后的最优解。所以别迷信“最佳实践”多问问自己如果现在Broker挂了我的用户会看到什么这个答案比任何技术文档都重要。
