RocketMQ獲取指定消息的實現(xiàn)方法（源碼）
更新時間：2020年08月16日 10:14:01 作者：貓毛·波拿巴
這篇文章主要給大家介紹了關于RocketMQ獲取指定消息的實現(xiàn)方法，文中通過示例代碼介紹的非常詳細，對大家學習或者使用RocketMQ具有一定的參考學習價值，需要的朋友們下面來一起學習學習吧
概要
消息查詢是什么？
消息查詢就是根據(jù)用戶提供的msgId從MQ中取出該消息
RocketMQ如果有多個節(jié)點如何查詢？
問題：RocketMQ分布式結構中，數(shù)據(jù)分散在各個節(jié)點，即便是同一Topic的數(shù)據(jù)，也未必都在一個broker上?？蛻舳嗽趺粗罃?shù)據(jù)該去哪個節(jié)點上查？
猜想1：逐個訪問broker節(jié)點查詢數(shù)據(jù)
猜想2：有某種數(shù)據(jù)中心存在，該中心知道所有消息存儲的位置，只要向該中心查詢即可得到消息具體位置，進而取得消息內(nèi)容
實際：
1.消息Id中含有消息所在的broker的地址信息（IP\Port）以及該消息在CommitLog中的偏移量。
2.客戶端實現(xiàn)會從msgId字符串中解析出broker地址，向指定broker節(jié)查詢消息。
問題：CommitLog文件有多個，只有偏移量估計不能確定在哪個文件吧？
實際：單個Broker節(jié)點內(nèi)offset是全局唯一的，不是每個CommitLog文件的偏移量都是從0開始的。單個節(jié)點內(nèi)所有CommitLog文件共用一套偏移量，每個文件的文件名為其第一個消息的偏移量。所以可以根據(jù)偏移量和文件名確定CommitLog文件。
源碼閱讀
0.使用方式
MessageExt msg = consumer.viewMessage(msgId);
1.消息ID解析
這個了解下就可以了
public class MessageId {
private SocketAddress address;
private long offset;
public MessageId(SocketAddress address, long offset) {
this.address = address;
this.offset = offset;
}
//get-set
}
//from MQAdminImpl.java
public MessageExt viewMessage(
String msgId) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
MessageId messageId = null;
try {
//從msgId字符串中解析出address和offset
//address = ip:port
//offset為消息在CommitLog文件中的偏移量
messageId = MessageDecoder.decodeMessageId(msgId);
} catch (Exception e) {
throw new MQClientException(ResponseCode.NO_MESSAGE, "query message by id finished, but no message.");
}
return this.mQClientFactory.getMQClientAPIImpl().viewMessage(RemotingUtil.socketAddress2String(messageId.getAddress()),
messageId.getOffset(), timeoutMillis);
}
//from MessageDecoder.java
public static MessageId decodeMessageId(final String msgId) throws UnknownHostException {
SocketAddress address;
long offset;
//ipv4和ipv6的區(qū)別
//如果msgId總長度超過32字符，則為ipv6
int ipLength = msgId.length() == 32 ? 4 * 2 : 16 * 2;
byte[] ip = UtilAll.string2bytes(msgId.substring(0, ipLength));
byte[] port = UtilAll.string2bytes(msgId.substring(ipLength, ipLength + 8));
ByteBuffer bb = ByteBuffer.wrap(port);
int portInt = bb.getInt(0);
address = new InetSocketAddress(InetAddress.getByAddress(ip), portInt);
// offset
byte[] data = UtilAll.string2bytes(msgId.substring(ipLength + 8, ipLength + 8 + 16));
bb = ByteBuffer.wrap(data);
offset = bb.getLong(0);
return new MessageId(address, offset);
}
2.長連接客戶端RPC實現(xiàn)
要發(fā)請求首先得先建立連接，這里方法可以看到創(chuàng)建連接相關的操作。值得注意的是，第一次訪問的時候可能連接還沒建立，建立連接需要消耗一段時間。代碼中對這個時間也做了判斷，如果連接建立完成后，發(fā)現(xiàn)已經(jīng)超時，則不再發(fā)出請求。目的應該是盡可能減少請求線程的阻塞時間。
//from NettyRemotingClient.java
@Override
public RemotingCommand invokeSync(String addr, final RemotingCommand request, long timeoutMillis)
throws InterruptedException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException {
long beginStartTime = System.currentTimeMillis();
//這里會先檢查有無該地址的通道，有則返回，無則創(chuàng)建
final Channel channel = this.getAndCreateChannel(addr);
if (channel != null && channel.isActive()) {
try {
//前置鉤子
doBeforeRpcHooks(addr, request);
//判斷通道建立完成時是否已到達超時時間，如果超時直接拋出異常。不發(fā)請求
long costTime = System.currentTimeMillis() - beginStartTime;
if (timeoutMillis < costTime) {
throw new RemotingTimeoutException("invokeSync call timeout");
}
//同步調(diào)用
RemotingCommand response = this.invokeSyncImpl(channel, request, timeoutMillis - costTime);
//后置鉤子
doAfterRpcHooks(RemotingHelper.parseChannelRemoteAddr(channel), request, response); //后置鉤子
return response;
} catch (RemotingSendRequestException e) {
log.warn("invokeSync: send request exception, so close the channel[{}]", addr);
this.closeChannel(addr, channel);
throw e;
} catch (RemotingTimeoutException e) {
if (nettyClientConfig.isClientCloseSocketIfTimeout()) {
this.closeChannel(addr, channel);
log.warn("invokeSync: close socket because of timeout, {}ms, {}", timeoutMillis, addr);
}
log.warn("invokeSync: wait response timeout exception, the channel[{}]", addr);
throw e;
}
} else {
this.closeChannel(addr, channel);
throw new RemotingConnectException(addr);
}
}
下一步看看它的同步調(diào)用做了什么處理。注意到它會構建一個Future對象加入待響應池，發(fā)出請求報文后就掛起線程，然后等待喚醒（waitResponse內(nèi)部使用CountDownLatch等待）。
//from NettyRemotingAbstract.javapublic RemotingCommand invokeSyncImpl(final Channel channel, final RemotingCommand request,
final long timeoutMillis)
throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException {
//請求id
final int opaque = request.getOpaque();
try {
//請求存根
final ResponseFuture responseFuture = new ResponseFuture(channel, opaque, timeoutMillis, null, null);
//加入待響應的請求池
this.responseTable.put(opaque, responseFuture);
final SocketAddress addr = channel.remoteAddress();
//將請求發(fā)出，成功發(fā)出時更新狀態(tài)
channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture f) throws Exception {
if (f.isSuccess()) { //若成功發(fā)出，更新請求狀態(tài)為“已發(fā)出”
responseFuture.setSendRequestOK(true);
return;
} else {
responseFuture.setSendRequestOK(false);
}
//若發(fā)出失敗，則從池中移除（沒用了，釋放資源）
responseTable.remove(opaque);
responseFuture.setCause(f.cause());
//putResponse的時候會喚醒等待的線程
responseFuture.putResponse(null);
log.warn("send a request command to channel <" + addr + "> failed.");
}
});
//只等待一段時間，不會一直等下去
//若正常響應，則收到響應后，此線程會被喚醒，繼續(xù)執(zhí)行下去
//若超時，則到達該時間后線程蘇醒，繼續(xù)執(zhí)行
RemotingCommand responseCommand = responseFuture.waitResponse(timeoutMillis);
if (null == responseCommand) {
if (responseFuture.isSendRequestOK()) {
throw new RemotingTimeoutException(RemotingHelper.parseSocketAddressAddr(addr), timeoutMillis,
responseFuture.getCause());
} else {
throw new RemotingSendRequestException(RemotingHelper.parseSocketAddressAddr(addr), responseFuture.getCause());
}
}
return responseCommand;
} finally {
//正常響應完成時，將future釋放（正常邏輯）
//超時時，將future釋放。這個請求已經(jīng)作廢了，后面如果再收到響應，就可以直接丟棄了（由于找不到相關的響應鉤子，就不處理了）
this.responseTable.remove(opaque);
}
}
好，我們再來看看收到報文的時候是怎么處理的。我們都了解JDK中的Future的原理，大概就是將這個任務提交給其他線程處理，該線程處理完畢后會將結果寫入到Future對象中，寫入時如果有線程在等待該結果，則喚醒這些線程。這里也差不多，只不過執(zhí)行線程在服務端，服務執(zhí)行完畢后會將結果通過長連接發(fā)送給客戶端，客戶端收到后根據(jù)報文中的ID信息從待響應池中找到Future對象，然后就是類似的處理了。
class NettyClientHandler extends SimpleChannelInboundHandler<RemotingCommand> {
//底層解碼完畢得到RemotingCommand的報文
@Override
protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception {
processMessageReceived(ctx, msg);
}
}
public void processMessageReceived(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception {
final RemotingCommand cmd = msg;
if (cmd != null) {
//判斷類型
switch (cmd.getType()) {
case REQUEST_COMMAND:
processRequestCommand(ctx, cmd);
break;
case RESPONSE_COMMAND:
processResponseCommand(ctx, cmd);
break;
default:
break;
}
}
}
public void processResponseCommand(ChannelHandlerContext ctx, RemotingCommand cmd) {
//取得消息id
final int opaque = cmd.getOpaque();
//從待響應池中取得對應請求
final ResponseFuture responseFuture = responseTable.get(opaque);
if (responseFuture != null) {
//將響應值注入到ResponseFuture對象中，等待線程可從這個對象獲取結果
responseFuture.setResponseCommand(cmd);
//請求已處理完畢，釋放該請求
responseTable.remove(opaque);
//如果有回調(diào)函數(shù)的話則回調(diào)（由當前線程處理）
if (responseFuture.getInvokeCallback() != null) {
executeInvokeCallback(responseFuture);
} else {
//沒有的話，則喚醒等待線程（由等待線程做處理）
responseFuture.putResponse(cmd);
responseFuture.release();
}
} else {
log.warn("receive response, but not matched any request, " + RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
log.warn(cmd.toString());
}
}
總結一下，客戶端的處理時序大概是這樣的：
結構大概是這樣的：
3.服務端的處理
//todo 服務端待補充CommitLog文件映射相關內(nèi)容
class NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception {
processMessageReceived(ctx, msg);
}
}
//from NettyRemotingAbscract.java
public void processMessageReceived(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception {
final RemotingCommand cmd = msg;
if (cmd != null) {
switch (cmd.getType()) {
case REQUEST_COMMAND: //服務端走這里
processRequestCommand(ctx, cmd);
break;
case RESPONSE_COMMAND:
processResponseCommand(ctx, cmd);
break;
default:
break;
}
}
}
//from NettyRemotingAbscract.java
public void processRequestCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) {
//查看有無該請求code相關的處理器
final Pair<NettyRequestProcessor, ExecutorService> matched = this.processorTable.get(cmd.getCode());
//如果沒有，則使用默認處理器（可能沒有默認處理器）
final Pair<NettyRequestProcessor, ExecutorService> pair = null == matched ? this.defaultRequestProcessor : matched;
final int opaque = cmd.getOpaque();
if (pair != null) {
Runnable run = new Runnable() {
@Override
public void run() {
try {
doBeforeRpcHooks(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd);
final RemotingResponseCallback callback = new RemotingResponseCallback() {
@Override
public void callback(RemotingCommand response) {
doAfterRpcHooks(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd, response);
if (!cmd.isOnewayRPC()) {
if (response != null) { //不為null，則由本類將響應值寫會給請求方
response.setOpaque(opaque);
response.markResponseType();
try {
ctx.writeAndFlush(response);
} catch (Throwable e) {
log.error("process request over, but response failed", e);
log.error(cmd.toString());
log.error(response.toString());
}
} else { //為null，意味著processor內(nèi)部已經(jīng)將響應處理了，這里無需再處理。
}
}
}
};
if (pair.getObject1() instanceof AsyncNettyRequestProcessor) {//QueryMessageProcessor為異步處理器
AsyncNettyRequestProcessor processor = (AsyncNettyRequestProcessor)pair.getObject1();
processor.asyncProcessRequest(ctx, cmd, callback);
} else {
NettyRequestProcessor processor = pair.getObject1();
RemotingCommand response = processor.processRequest(ctx, cmd);
doAfterRpcHooks(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd, response);
callback.callback(response);
}
} catch (Throwable e) {
log.error("process request exception", e);
log.error(cmd.toString());
if (!cmd.isOnewayRPC()) {
final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_ERROR,
RemotingHelper.exceptionSimpleDesc(e));
response.setOpaque(opaque);
ctx.writeAndFlush(response);
}
}
}
};
if (pair.getObject1().rejectRequest()) {
final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY,
"[REJECTREQUEST]system busy, start flow control for a while");
response.setOpaque(opaque);
ctx.writeAndFlush(response);
return;
}
try {
final RequestTask requestTask = new RequestTask(run, ctx.channel(), cmd);
pair.getObject2().submit(requestTask);
} catch (RejectedExecutionException e) {
if ((System.currentTimeMillis() % 10000) == 0) {
log.warn(RemotingHelper.parseChannelRemoteAddr(ctx.channel())
+ ", too many requests and system thread pool busy, RejectedExecutionException "
+ pair.getObject2().toString()
+ " request code: " + cmd.getCode());
}
if (!cmd.isOnewayRPC()) {
final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY,
"[OVERLOAD]system busy, start flow control for a while");
response.setOpaque(opaque);
ctx.writeAndFlush(response);
}
}
} else {
String error = " request type " + cmd.getCode() + " not supported";
final RemotingCommand response =
RemotingCommand.createResponseCommand(RemotingSysResponseCode.REQUEST_CODE_NOT_SUPPORTED, error);
response.setOpaque(opaque);
ctx.writeAndFlush(response);
log.error(RemotingHelper.parseChannelRemoteAddr(ctx.channel()) + error);
}
}
//from QueryMessageProcesor.java
@Override
public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)
throws RemotingCommandException {
switch (request.getCode()) {
case RequestCode.QUERY_MESSAGE:
return this.queryMessage(ctx, request);
case RequestCode.VIEW_MESSAGE_BY_ID: //通過msgId查詢消息
return this.viewMessageById(ctx, request);
default:
break;
}
return null;
}
public RemotingCommand viewMessageById(ChannelHandlerContext ctx, RemotingCommand request)
throws RemotingCommandException {
final RemotingCommand response = RemotingCommand.createResponseCommand(null);
final ViewMessageRequestHeader requestHeader =
(ViewMessageRequestHeader) request.decodeCommandCustomHeader(ViewMessageRequestHeader.class);
response.setOpaque(request.getOpaque());
//getMessagetStore得到當前映射到內(nèi)存中的CommitLog文件，然后根據(jù)偏移量取得數(shù)據(jù)
final SelectMappedBufferResult selectMappedBufferResult =
this.brokerController.getMessageStore().selectOneMessageByOffset(requestHeader.getOffset());
if (selectMappedBufferResult != null) {
response.setCode(ResponseCode.SUCCESS);
response.setRemark(null);
//將響應通過socket寫回給客戶端
try {
//response對象的數(shù)據(jù)作為header
//消息內(nèi)容作為body
FileRegion fileRegion =
new OneMessageTransfer(response.encodeHeader(selectMappedBufferResult.getSize()),
selectMappedBufferResult);
ctx.channel().writeAndFlush(fileRegion).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
selectMappedBufferResult.release();
if (!future.isSuccess()) {
log.error("Transfer one message from page cache failed, ", future.cause());
}
}
});
} catch (Throwable e) {
log.error("", e);
selectMappedBufferResult.release();
}
return null; //如果有值，則直接寫回給請求方。這里返回null是不需要由外層處理響應。
} else {
response.setCode(ResponseCode.SYSTEM_ERROR);
response.setRemark("can not find message by the offset, " + requestHeader.getOffset());
}
return response;
}
總結
到此這篇關于RocketMQ獲取指定消息的文章就介紹到這了,更多相關RocketMQ獲取指定消息內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
SpringBoot3整合郵件服務實現(xiàn)郵件發(fā)送功能
本文介紹了spring boot整合email服務,實現(xiàn)發(fā)送驗證碼,郵件（普通文本郵件、靜態(tài)資源郵件、附件郵件）,文中通過代碼示例介紹的非常詳細,堅持看完相信對你有幫助,需要的朋友可以參考下2024-05-05
springboot如何為web層添加統(tǒng)一請求前綴
這篇文章主要介紹了springboot如何為web層添加統(tǒng)一請求前綴，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-02-02
本文分享一個java實現(xiàn)計算周期性提醒的示例，可以計算父親節(jié)、母親節(jié)這樣的節(jié)日，也可以定義如每月最好一個周五，以方便安排會議2014-04-04
在進行項目debug時遇到了無法啟動的問題,項目一直處于正在啟動狀態(tài),但未出現(xiàn)任何報錯信息,分析原因可能是存在不合法的斷點位置,即斷點未打在方法內(nèi)部,解決方法是檢查所有斷點信息,并移除非法斷點,之后項目能夠正常啟動2023-02-02
這篇文章主要介紹了使用Jenkins配置Git+Maven的自動化構建的方法，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2019-01-01
這篇文章主要為大家詳細介紹了Spring框架實現(xiàn)AOP的兩種方式，文中的示例代碼講解詳細，對我們學習有一定的借鑒價值，需要的可以參考一下2022-09-09
這篇文章主要介紹了淺談Java日志簡單理解，詳細的介紹了slf4j的概念和使用，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2018-07-07
在日常開發(fā)過程中總是以單線程的思維去編碼,沒有考慮到在多線程狀態(tài)下的運行狀況.由此引發(fā)的結果就是請求過多,應用無法響應.為了解決請求過多的問題,又衍生出了線程池的概念.本文記錄了Java中線程池的使用及工作原理,需要的朋友可以參考下2021-05-05
JetBrains IntelliJ IDEA 優(yōu)化教超詳細程
這篇文章主要介紹了JetBrains IntelliJ IDEA 優(yōu)化教超詳細程,本文通過圖文并茂的形式給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2021-03-03
Synchronized關鍵字是一個種鎖，其有很多名字,例如重量級鎖、悲觀鎖、可重入鎖、、非公平、對象鎖等等,這篇文章主要給大家介紹了關于Java中復雜的Synchronized關鍵字使用方法的相關資料,需要的朋友可以參考下2024-01-01