rocketmq之send

2021-10-27 19:03:31 阅读：186 来源： 互联网

标签：return request ctx send requestHeader mqtraceContext public rocketmq

CommunicationMode

public enum CommunicationMode {
    SYNC, // 同步
    ASYNC, // 异步
    ONEWAY, // 发出去就不管了
}

重试次数

# 同步默认 = 1+ 2 =3 ， # 非同步(异步、OneWay) = 1  (总共就1次，其实就是没有重试)
int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;

发送容错

broker

SendMessageProcessor

实现接口

public class SendMessageProcessor extends AbstractSendMessageProcessor implements NettyRequestProcessor {
}

processRequest

public RemotingCommand processRequest(ChannelHandlerContext ctx,
                                          RemotingCommand request) throws RemotingCommandException {
	RemotingCommand response = null;
	try {
		response = asyncProcessRequest(ctx, request).get(); // Future.get()
	} catch (InterruptedException | ExecutionException e) {
		log.error("process SendMessage error, request : " + request.toString(), e);
	}
	return response;
}

asyncProcessRequest

public CompletableFuture<RemotingCommand> asyncProcessRequest(ChannelHandlerContext ctx,
                                                                  RemotingCommand request) throws RemotingCommandException {
	final SendMessageContext mqtraceContext;
	switch (request.getCode()) {
		case RequestCode.CONSUMER_SEND_MSG_BACK:
			return this.asyncConsumerSendMsgBack(ctx, request);
		default:
			SendMessageRequestHeader requestHeader = parseRequestHeader(request);
			if (requestHeader == null) {
				return CompletableFuture.completedFuture(null);
			}
			mqtraceContext = buildMsgContext(ctx, requestHeader);
			this.executeSendMessageHookBefore(ctx, request, mqtraceContext);
			if (requestHeader.isBatch()) { // 批量消息
				return this.asyncSendBatchMessage(ctx, request, mqtraceContext, requestHeader);
			} else { // 非批量消息
				return this.asyncSendMessage(ctx, request, mqtraceContext, requestHeader);
			}
	}
}

标签：return,request,ctx,send,requestHeader,mqtraceContext,public,rocketmq
来源： https://blog.csdn.net/kq1983/article/details/120957597

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9