首页 > 代码库 > Rocketmq消息持久化
Rocketmq消息持久化
本文编写,参考:https://my.oschina.net/bieber/blog/725646
producer Send()的Message最终将由broker处理,处理类为:SendMessageProcessor ,处理方法:processRequet.
public class SendMessageProcessor extends AbstractSendMessageProcessor implements NettyRequestProcessor {
private List<ConsumeMessageHook> consumeMessageHookList;
public SendMessageProcessor(final BrokerController brokerController) {
super(brokerController);
}
@Override
public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {}
上述方法,并不是直接处理消息,而是交由MessageStore处理,相关代码如下:
MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
msgInner.setTopic(requestHeader.getTopic());
msgInner.setQueueId(queueIdInt);
//......
PutMessageResult putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
然而MessageStore也不直接持久化消息,转交给 CommitLog
long beginTime = this.getSystemClock().now();
PutMessageResult result = this.commitLog.putMessages(messageExtBatch);
从MappedFileQueue中取出最新的一条:
MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
//写消息
result = mappedFile.appendMessages(messageExtBatch, this.appendMessageCallback);
//持久化到磁盘
handleDiskFlush(result, putMessageResult, messageExtBatch);
handleHA(result, putMessageResult, messageExtBatch);
cousumer 从broker读消息。
Rocketmq消息持久化
声明:以上内容来自用户投稿及互联网公开渠道收集整理发布,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任,若内容有误或涉及侵权可进行投诉: 投诉/举报 工作人员会在5个工作日内联系你,一经查实,本站将立刻删除涉嫌侵权内容。