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

RocketMQ简介

backend 2025/7/2 3:08:12

生产与消费流程：

消费者会启动一个后台线程，不断地从Broker拉取消息，并放入到本地的队列中。消费者通过长轮询（Long Polling）机制来优化轮询效率和新消息的实时性。在没有新消息可拉取时，Broker会挂起消费者的请求，并在有新消息到达时立即返回给消费者，或者在请求超时后返回空结果。

消息确认机制

确定消息生产并接收。生产者->broker

单向发送

消费者直接给broker，不关心broker是否接收到。producer.sendOneway(message);

同步发送

同步发送⽅式下，消息⽣产者在往Broker端发送消息后，会阻塞当前线程，等待Broker端的相应结果。producer.send(msg);

异步发送

⽣产者在向Broker发送消息时，会同时注册⼀个回调函数。接下来⽣产者并不等待Broker的
响应。当Broker端有响应数据过来时，⾃动触发回调函数进⾏对应的处理。

producer.send(msg, new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {countDownLatch.countDown();System.out.printf("%-10d OK %s %n", index,sendResult.getMsgId());}@Overridepublic void onException(Throwable e) {countDownLatch.countDown();System.out.printf("%-10d Exception %s %n", index, e);e.printStackTrace();}
});

确定消息被消费。broker->消费者

broker等待消费者返回接收到的信号。如果等不到，broker也会重新发，默认重发16次失败后，消息会进入到死信队列。这个等待返回信息和重发的操作是异步的，不会阻塞其他信息的推送。

广播模式和集群模式

⼴播模式和集群模式是RocketMQ的消费者端处理消息最基本的两种模式。集群模式下，⼀个消息，只会被⼀个消费者组中的多个消费者实例共同处理⼀次。⼴播模式下，⼀个消息，则会推送给所有消费者实例处理，不再关⼼消费者组

顺序消息

用MessageQueueSelector确定发送的队列，路由是自己代码写

SendResult sendResult = producer.send(msg, new MessageQueueSelector() {@Overridepublic MessageQueue select(List<MessageQueue> mqs, Message msg,Object arg) {Integer id = (Integer) arg;int index = id % mqs.size();return mqs.get(index);}
}, orderId);

延迟消息

生产者msg.setDelayTimeLevel(3);

延迟消息的难点其实是性能，需要不断进⾏定时轮训。全部扫描所有消息是不可能的，RocketMQ的实现⽅式是预设⼀个系统Topic，名字叫做SCHEDULE_TOPIC_XXXX。在这个Topic下，预设18个延迟队列。然后每次只针对这18个队列⾥的消息进⾏延迟操作，这样就不⽤⼀直扫描所有的消息了。

批量消息

    List<Message> messages = new ArrayList<>();messages.add(new Message(topic, "Tag", "OrderID001", "Hello world
0".getBytes()));messages.add(new Message(topic, "Tag", "OrderID002", "Hello world
1".getBytes()));messages.add(new Message(topic, "Tag", "OrderID003", "Hello world
2".getBytes()));producer.send(messages);

过滤消息

订阅过滤

sql过滤

事务——两阶段确认

下单问题。订单状态成功改变之后才能执行清空购物车、清空支付单等一系列操作。业务侧发消息给rocketMQ，rocketMQ把消息标记为“暂不投递”状态（第一阶段），rocketMQ返回ack给业务侧，业务侧执行sql，确定订单状态已经入库，业务侧发消息（commit/rollback）给rocketMQ（第二阶段），rocketMQ把消息标记为ok状态，这时候的消息才能被消费者消费。

权限

RocketMQ针对每个Topic，就有完整的权限控制。⽐如，在控制平台中，就可以很⽅便的给每个Topic配置权限。是否可写、是否可订阅。

url白名单

核心源码

@RocketMQMessageListener注解方式创建消费者，是如何创建的呢？⼊⼝在rocketmq-spring-boot-2.2.2.jar中的org.apache.rocketmq.spring.autoconfigure.ListenerContainerConfiguration类中。这个ListenerContainerConfiguration类继承了Spring当中的SmartInitializingSingleton接⼝，当Spring容器当中所有⾮懒加载的实例加载完成后，就会触发他的afterSingletonsInstantiated⽅法进⾏初始化。在这个⽅法中会去扫描所有带有注解@RocketMQMessageListener注解的类，将他注册到内部⼀个Container容器当中。重点看afterSingletonsInstantiated方法。