RocketMq快速入门-北京尧图网络

mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnabletrue开启确保 Broker 启动时就允许自动创建主题对比维度Message普通消息MessageExt扩展消息谁在用生产者Producer发送消息时使用消费者Consumer接收消息时使用包含什么只有Topic、Tag、业务 Body消息体包含全部 Message 内容系统元数据用途只负责告诉 Broker “我要发什么内容”告诉消费者“你收到的这条消息在服务器里的状态是什么”正确的消费者管理方式长期运行保持单例正确的模式是将消费者设计为长期运行的单例。在应用启动时创建并启动它然后让它持续监听应用关闭时才统一关闭。优雅关闭只做一次shutdown()应该只在你确定要彻底停止消息消费并释放资源时例如应用正常下线作为最后的清理步骤被调用一次。避免在监听器中操作绝对不要在MessageListener的consumeMessage()方法内部去调用consumer.shutdown()。这会导致消费者在消息处理过程中被意外销毁。RocketMQ的消息分发遵循“组内竞争组间复制”原则同组Group相同竞争关系一条消息只投递给组内一个消费者负载均衡。异组Group不同复制关系每条消息会被每个组独立消费一次业务解耦。同组内的“争抢”不是随机抢单条而是“抢队列”RocketMQ的实际分配单位是队列MessageQueue而不是单条消息。比如一个Topic有4个队列组里有2个消费者那么就是消费者A拿队列1和2消费者B拿队列3和4各自抢占一部分队列。所以同组内并不是每条消息都随机乱抢而是先分好“责任田”队列再各自消费田里的消息。但是也可以在RocketMQ中同组使用广播模式即发送n条消息同一组下每个消费者都消费n条。consumer.setMessageModel(MessageModel.BROADCASTING);核心运行机制对比底层原理对比维度同步消息 (Sync)异步消息 (Async)线程状态发送线程阻塞Blocking等待网络IO和Broker响应。发送线程非阻塞Non-blocking发完即返回。结果获取直接在send()方法的返回值里拿到SendResult。通过实现SendCallback接口在onSuccess()或onException()里拿结果。可靠性高。能立即感知成功/失败便于重试或回滚。较高。有回调机制但需要处理好回调线程和主线程的上下文传递。异步发送sendCallback后立即执行shutdown()会因强制销毁连接和线程池导致网络请求中断而报错。单向消息// 单向发送没有返回值没有回调 producer.sendOneway(msg);WaitStoreMsgOK是 RocketMQ 消息的一个属性用于控制消息发送的可靠性。它的核心作用是决定生产者发送消息时是否需要等待 Broker服务端将消息成功写入磁盘落盘后才返回成功响应。当WaitStoreMsgOK true默认生产者发送消息后会一直等待 Broker 的响应。Broker 收到消息后会先将其写入磁盘根据配置可能是同步或异步刷盘只有在写入成功或在一定超时时间内后才会向生产者返回SEND_OK状态。这保证了消息的高可靠性但会带来额外的网络延迟略微降低发送性能。当WaitStoreMsgOK false生产者发送消息后Broker只要将消息写入内存PageCache即返回成功无需等待落盘完成。这种方式发送性能更高、延迟更低但可靠性降低因为在 Broker 发生断电等故障时未落盘的消息可能会丢失。consumer.subscribe(topic6, MessageSelector.bySql(age18));消费者模式只有PUSH模式的消费者支持SQL过滤。订阅一致性同一消费者组ConsumerGroup内的所有消费者其订阅关系包括过滤表达式必须完全一致RocketMQ Message 追加属性1. 两种属性的本质区别属性类型设置方法用途举例系统属性直接setXxx()方法RocketMQ 框架内部识别和使用。例如延迟级别、Keys、Flag 等。msg.setKeys(order_123)、msg.setDelayTimeLevel(3)用户属性putUserProperty(key, value)业务自定义存放在消息的properties字段中。用于消息过滤Tag/SQL或携带业务元数据。msg.putUserProperty(age, 18)、msg.putUserProperty(region, shanghai)putUserProperty(Key, Value)Key 和 Value 都必须是 String且 Key 禁止__开头。并非所有系统属性都开放给了SQL过滤。像KEYS、DELAY、RETRY_TOPIC等内部系统属性通常无法直接在SQL表达式中使用。但是UserProperty都可以用来sql过滤维度具体内容关键细节 / 避坑指南接口全称java.io.Serializable位于java.io包属于标记接口内部无任何抽象方法只起标识作用。接口作用允许类的对象进行序列化对象→字节流和反序列化字节流→对象。没有此标记的类在写出或传输时会直接抛出NotSerializableException。核心常量serialVersionUID序列化版本号private static final long serialVersionUID 1L;必须手动显式定义若不写类结构一变如增删字段JVM自动生成的哈希值就变了反序列化必报InvalidClassException。排除字段transient关键字修饰敏感字段如密码、Token使其不被序列化。反序列化后该字段恢复为默认值null/0。继承规则• 父类实现了Serializable→ 子类自动可序列化。• 父类未实现 → 子类反序列化时会调用父类的无参构造器。若父类未实现且没有无参构造器反序列化直接报错。Springboot中快速使用RocketMQ生产者发送消息:方法名大白话解释要不要等结果有返回值吗参数能直接传对象吗什么时候用syncSend同步发送消息发出去原地等着直到服务器说“收到了”才往下走。✅要等阻塞✅ 有能拿到消息ID❌ 不行得手动转成字符串重要消息比如下单、付款必须确认发送成功才能继续。asyncSend异步发送消息发出去不管了先往下执行。等服务器回复了再回头通过“回调”通知你。❌不等非阻塞❌ 没有结果在回调里拿❌ 不行得手动转成字符串不重要但量大的消息比如记日志不能让它拖慢主流程。convertAndSend转换后发送直接把 Java 对象比如User、Order扔进去它自动转成 JSON再发出去。✅要等同 syncSend❌ 没有✅能直接传对象就行懒得手动转 JSON 的时候直接把对象往里一丢最省事。// 1. syncSend等结果手动转JSON String json {\name\:\张三\}; SendResult result rocketMQTemplate.syncSend(topic, json); System.out.println(消息ID result.getMsgId()); // 立刻能拿到 // 2. asyncSend不等手动转JSON结果靠回调 rocketMQTemplate.asyncSend(topic, json, new SendCallback() { // 成功或失败都在这里处理但主线程已经往下跑了 }); // 3. convertAndSend不用转JSON直接传对象但没返回值 rocketMQTemplate.convertAndSend(topic, new User(张三)); // 发就完了不用管结果单向消息rocketMQTemplate.sendOneWay(topic, 这是一条单向消息);延时消息在RocketMQ4.x中不支持精确的延迟时间/ 发送延迟消息最后一个参数 3 就是延迟级别 // 级别3对应延迟10秒 (详见附录的级别表) rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(message).build(), 3000, // 发送超时时间单位毫秒 3); // ⬅️ 延迟级别3 代表延迟 10 秒从5.0开始支持精确延迟时间// 延迟时间单位是毫秒 // 5000 5秒300000 5分钟1800000 30分钟 long delayInMilliseconds 300000; // 5分钟 // 使用 syncSendDelayTimeMills 发送延迟消息 rocketMQTemplate.syncSendDelayTimeMills(topic, message, delayInMilliseconds);批量消息直接发送MessageList即可消费者消费数据import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.springframework.stereotype.Component; // Component把这个类交给Spring管理必须加 // RocketMQMessageListener告诉Spring我要监听哪个Topic Component RocketMQMessageListener( topic my-topic, // 监听哪个主题必须和生产者发的一样 consumerGroup my-consumer-group // 消费者组名必须填唯一就行 ) public class MyConsumer implements RocketMQListenerString { // 实现这个接口的 onMessage 方法 // 只要生产者发了消息这个方法就会被自动调用消息内容就是参数 message Override public void onMessage(String message) { // 写你的业务逻辑 System.out.println(收到消息啦 message); // 比如保存到数据库、调用其他接口... } }要做tag过滤RocketMQMessageListener注解里加上selectorExpression参数要做sql过滤RocketMQMessageListener注解加上selectorType SelectorType.SQL92搭配selectorExpression即可。要把同组内消费模式改为广播在注解内加上messageModel MessageModel.BROADCASTING即可。如果生产者不指定MessageQueueSelectorRocketMQ 默认采用轮询Round-robin或随机策略发送消息。订单号 A 的第1条消息创建发到了队列0。订单号 A 的第2条消息支付发到了队列1。由于队列0和队列1可能被不同的消费者实例或不同的消费线程并行拉取且处理速度不同队列0的消息可能还没处理完队列1的支付消息已经被消费了这就产生了乱序。发送端MessageQueueSelector的作用它的作用是将同一个业务标识如订单ID的消息固定路由到同一个队列。// 正确用法示例 producer.send(msg, new MessageQueueSelector() { Override public MessageQueue select(ListMessageQueue mqs, Message msg, Object arg) { // arg 即为传入的订单ID String orderId (String) arg; // 对订单ID取模确保同一个订单永远进同一个队列 int index Math.abs(orderId.hashCode()) % mqs.size(); return mqs.get(index); } }, orderId);注意如果队列数量发生变化比如扩容取模结果会变导致同一个订单落到不同队列因此顺序敏感的场景下生产环境严禁动态调整队列数量。消费端必须配合“顺序消费模式”最关键的一环即使发送端用 Selector 把消息都发到了队列0如果消费端用错模式依然会乱序错误做法使用MessageListenerConcurrently并发消费。队列0里的消息虽然是有序的但消费端会用多线程并发拉取线程A处理消息1耗时2秒线程B处理消息2耗时0.1秒消息2先处理完业务上就乱序了。正确做法必须使用MessageListenerOrderly顺序消费。consumer.registerMessageListener(new MessageListenerOrderly() { Override public ConsumeOrderlyStatus consumeMessage(ListMessageExt msgs, ConsumeOrderlyContext context) { // 框架底层会加锁队列锁确保同一个队列的消息在同一时刻**只有一个线程**在处理 // 消息1不处理完消息2绝对无法被拉取消费 return ConsumeOrderlyStatus.SUCCESS; } });消费失败时的“避坑”准则极易引发乱序如果在顺序消费模式中某条消息如消息2处理失败千万不能返回RECONSUME_LATER放入重试队列因为重试队列是独立的会导致消息2跑到后面去消息3、4先被消费造成乱序。正确做法返回SUSPEND_CURRENT_QUEUE_A_MOMENT挂起当前队列让消费者稍后默认几秒后重新拉取整个队列头部的消息并且阻塞后续消息的拉取直到这条失败的消息被成功消费。在springboot中使用// 1. 必须加 Service 注解交给 Spring 管理 // 2. 必须加 RocketMQTransactionListener 注解并指定生产组和 yml 里保持一致 Service RocketMQTransactionListener(txProducerGroup tx-order-producer-group) public class OrderTransactionListener implements RocketMQLocalTransactionListener { // 模拟注入你的业务 Service比如操作数据库的 Autowired private OrderService orderService; /** * 方法一执行本地事务这是 Broker 发送半消息后立马调用的方法 * 入参说明 * Message msg发来的消息体 * Object arg发送时传进来的额外参数比如订单号 */ Override public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) { // 1. 把参数 arg 强转为订单号 String orderId (String) arg; try { // 2. 执行你真正的本地业务比如更新订单状态为“已支付” // 这里假设 orderService.updateOrderStatus 返回 boolean boolean success orderService.updateOrderStatus(orderId, PAID); // 3. 根据业务执行结果告诉 Broker 是提交还是回滚 if (success) { // 本地事务成功 → 提交消息消费者可以看到了 return RocketMQLocalTransactionState.COMMIT; } else { // 本地事务失败 → 回滚消息丢弃消费者看不见 return RocketMQLocalTransactionState.ROLLBACK; } } catch (Exception e) { // 如果发生了异常比如数据库超时不知道成功还是失败 // 返回 UNKNOWN让 Broker 过一会来回查再决定最终状态 return RocketMQLocalTransactionState.UNKNOWN; } } /** * 方法二事务回查Broker 没收到提交/回滚确认时会主动调用这个方法 * 注意这个方法是 Broker 主动调用的可能被调用多次所以查询逻辑要幂等 */ Override public RocketMQLocalTransactionState checkLocalTransaction(Message msg) { // 1. 从消息头里取出业务 ID之前在发送时塞进去了 String orderId (String) msg.getHeaders().get(orderId); // 2. 去数据库查询这个订单的真实状态 Order order orderService.getOrderById(orderId); // 3. 根据查到的结果回复 Broker if (order ! null PAID.equals(order.getStatus())) { // 订单已支付 → 提交消息 return RocketMQLocalTransactionState.COMMIT; } else if (order ! null INIT.equals(order.getStatus())) { // 订单还是初始状态 → 说明本地事务没执行成功回滚 return RocketMQLocalTransactionState.ROLLBACK; } else { // 查不到订单或者状态异常继续返回 UNKNOWN让 Broker 稍后再来回查 return RocketMQLocalTransactionState.UNKNOWN; } } }brokerID0表示为主节点

RocketMq快速入门

相关推荐

U-Net为何仍是医学图像分割首选？核心架构与实战避坑指南

Win7精简系统日文输入法安装全解析：双版本组件与排查指南

从零手写RISC-V SBI：MySBI与BenOS的启动与中断调试实践

更多资讯

Hermes Agent性能监控告警策略：阈值设置与通知渠道

探索Hermes Agent的自然语言推理应用：问答系统与知识图谱

Hermes Agent容器网络策略配置：入站与出站规则设计

Hermes Agent日志分析工具集成：ELK Stack与机器学习插件全攻略

优化Hermes Agent的存储选择：云存储与本地存储全面比较指南

基于Springboot小型诊疗预约平台【附源码+文档】

今日更新

AI辅助技术栈评估：提升研发决策效率60%

从生成Token到直接决策：Jev如何重塑AI Agent工程实践

32G统一内存跑大模型：内存账、TPS实测与端云分流

本周精选

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

本月热门

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

读完文章，想聊聊您的网站？