How does a Producer send messages to the Broker?
[TOC]
preface
This analysis is based on RocketMQ release-4.5.2.
The objective of the analysis is: how does the Producer in RocketMQ send messages to the Broker?
When it comes to learning the source code, of course, the first is to download the source code, official address. Use the git clone https://github.com/apache/rocketmq.git to source code clone to local.
The project structure
Open the project with IDEA
Rocketmq – client modules
You can see that there are many submodules. This time, we are studying Producer, so open the RocketMQ-Client module. In the unit test, you can find the classes that test the function of Producer.
DefaultMQProducerTest
Open the class and observe its methods
You can see that methods that start with test are unit test methods and can be run directly. The init and Terminate methods are the initialization and destruction methods of the unit tests, respectively.
Init and the terminate
// Create a default client instance @spy private MQClientInstance mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(new ClientConfig()); // mock an object that actually interacts with the broker @mock Private MQClientAPIImpl MQClientAPIImpl; @Mock private NettyRemotingClient nettyRemotingClient; private DefaultMQProducer producer; private Message message; private Message zeroMsg; private Message bigMessage; private String topic = "FooBar"; private String producerGroupPrefix = "FooBar_PID"; @before public void init() throws Exception {String producerGroupTemp = producerGroupPrefix + System.currentTimeMillis(); // Create a default producer = new DefaultMQProducer(producerGroupTemp); Producer. SetNamesrvAddr (127.0.0.1: "9876"); producer.setCompressMsgBodyOverHowmuch(16); message = new Message(topic, new byte[] {'a'}); zeroMsg = new Message(topic, new byte[] {}); bigMessage = new Message(topic, "This is a very huge message!" .getBytes()); producer.start(); / / reflection object set the client instance to the producer in the Field, the Field = DefaultMQProducerImpl. Class. GetDeclaredField (" mQClientFactory "); field.setAccessible(true); field.set(producer.getDefaultMQProducerImpl(), mQClientFactory); / / reflection will be a real objects interact with the broker Set to the client instance field = MQClientInstance. Class. GetDeclaredField (" mQClientAPIImpl "); field.setAccessible(true); field.set(mQClientFactory, mQClientAPIImpl); / / registered client instance producer. GetDefaultMQProducerImpl () getmQClientFactory () registerProducer (producerGroupTemp, producer.getDefaultMQProducerImpl()); / / the mock object interaction Message when (mQClientAPIImpl. SendMessage (anyString (), anyString (), any (Message. Class), any(SendMessageRequestHeader.class), anyLong(), any(CommunicationMode.class), nullable(SendMessageContext.class), any(DefaultMQProducerImpl.class))).thenCallRealMethod(); when(mQClientAPIImpl.sendMessage(anyString(), anyString(), any(Message.class), any(SendMessageRequestHeader.class), anyLong(), any(CommunicationMode.class), nullable(SendCallback.class), nullable(TopicPublishInfo.class), nullable(MQClientInstance.class), anyInt(), nullable(SendMessageContext.class), any(DefaultMQProducerImpl.class))) .thenReturn(createSendResult(SendStatus.SEND_OK)); } // Destroy @after public void terminate() {producer.shutdown(); }Copy the codetestSendMessageSync_Success
The testSendMessageSync_Success() method is selected as the entry point for this analysis. This method is used to test the successful sending of synchronization messages.
DEBUG traces the call chain to see that MQClientAPIImpl#sendMessage is the underlying wrapper that sends messages to the broker. Its by introducing rocketmq – remoting module org.apache.rocketmq.remoting.netty.Net tyRemotingClient classes interact with the Broker. Analysis of the NETty-based RPC protocol with the Broker is not performed here. You can learn more by reading the NettyRemotingClient class mentioned above.
PS: Because mockito is used, there are stacks in the invocation chain that are not related to the message being sent by the producer.
PS: By viewing the stack information of the call chain, you can quickly understand the whole process of a behavior in the source code.
The following source code works up the call chain from the bottom
public SendResult sendMessage( final String addr, final String brokerName, final Message msg, final SendMessageRequestHeader requestHeader, final long timeoutMillis, final CommunicationMode communicationMode, final SendMessageContext context, final DefaultMQProducerImpl producer ) throws RemotingException, MQBrokerException, Return sendMessage(ADDR, brokerName, MSG, requestHeader, timeoutMillis, communicationMode, null, null, null, 0, context, producer); Public SendResult sendMessage(Final String Addr, Final String brokerName, Final Message MSG, final SendMessageRequestHeader requestHeader, final long timeoutMillis, final CommunicationMode communicationMode, final SendCallback sendCallback, final TopicPublishInfo topicPublishInfo, final MQClientInstance instance, final int retryTimesWhenSendFailed, final SendMessageContext context, final DefaultMQProducerImpl producer ) throws RemotingException, MQBrokerException, InterruptedException { long beginStartTime = System.currentTimeMillis(); RemotingCommand request = null; If (sendSmartMsg | | MSG instanceof MessageBatch) {/ / field for all of the class a, b, c, d, etc. Can accelerate FastJson deserialization SendMessageRequestHeaderV2 requestHeaderV2 = SendMessageRequestHeaderV2.createSendMessageRequestHeaderV2(requestHeader); // Create an RPC request object according to request code // This design is in the form of type code, To identify the different types of request request = RemotingCommand. CreateRequestCommand (MSG instanceof MessageBatch? RequestCode.SEND_BATCH_MESSAGE : RequestCode.SEND_MESSAGE_V2, requestHeaderV2); } else { request = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, requestHeader); } // Set the request body, that is, the message body request.setBody(msg.getBody()); Switch (communicationMode) {case oneway: this.remotingClient.invokeOneway(addr, request, timeoutMillis); return null; case ASYNC: final AtomicInteger times = new AtomicInteger(); long costTimeAsync = System.currentTimeMillis() - beginStartTime; if (timeoutMillis < costTimeAsync) { throw new RemotingTooMuchRequestException("sendMessage call timeout"); } this.sendMessageAsync(addr, brokerName, msg, timeoutMillis - costTimeAsync, request, sendCallback, topicPublishInfo, instance, retryTimesWhenSendFailed, times, context, producer); return null; case SYNC: long costTimeSync = System.currentTimeMillis() - beginStartTime; if (timeoutMillis < costTimeSync) { throw new RemotingTooMuchRequestException("sendMessage call timeout"); } return this.sendMessageSync(addr, brokerName, msg, timeoutMillis - costTimeSync, request); default: assert false; break; } return null; }Copy the code private SendResult sendKernelImpl(final Message msg,
final MessageQueue mq,
final CommunicationMode communicationMode,
final SendCallback sendCallback,
final TopicPublishInfo topicPublishInfo,
final long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
long beginStartTime = System.currentTimeMillis();
// 查找 brokerName 对应 broker，master 节点的地址
String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
// 查找失败，尝试重新从 NameServer 拉取
if (null == brokerAddr) {
tryToFindTopicPublishInfo(mq.getTopic());
brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
}
SendMessageContext context = null;
if (brokerAddr != null) {
// 根据 VIP Channel 设置，更新 broker 节点地址
brokerAddr = MixAll.brokerVIPChannel(this.defaultMQProducer.isSendMessageWithVIPChannel(), brokerAddr);
byte[] prevBody = msg.getBody();
try {
//for MessageBatch,ID has been set in the generating process
if (!(msg instanceof MessageBatch)) {
// 设置 自定义属性 UNIQ_KEY -> 0A0A15A01F3C18B4AAC22DB7B6AC0000
MessageClientIDSetter.setUniqID(msg);
}
boolean topicWithNamespace = false;
if (null != this.mQClientFactory.getClientConfig().getNamespace()) {
// 设置 自定义属性 INSTANCE_ID -> <NameSpace>
msg.setInstanceId(this.mQClientFactory.getClientConfig().getNamespace());
topicWithNamespace = true;
}
// 消息设置 处理标识，用于标识消息经过什么样的处理，可以查看该类 org.apache.rocketmq.common.sysflag.MessageSysFlag ，该类是设计较好的标识处理，可以借鉴
int sysFlag = 0;
boolean msgBodyCompressed = false;
// 根据 DefaultMQProducer#compressMsgBodyOverHowmuch 选择是否压缩，默认超过 4K 则压缩，压缩算法为 zip
if (this.tryToCompressMessage(msg)) {
// 设置压缩标识，COMPRESSED_FLAG = 0x1
sysFlag |= MessageSysFlag.COMPRESSED_FLAG;
msgBodyCompressed = true;
}
// 获取属性，判断是否是事务消息， PROPERTY_TRANSACTION_PREPARED = "TRAN_MSG"
final String tranMsg = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
if (tranMsg != null && Boolean.parseBoolean(tranMsg)) {
// 设置事务标识，TRANSACTION_PREPARED_TYPE = 0x1 << 2
sysFlag |= MessageSysFlag.TRANSACTION_PREPARED_TYPE;
}
// hook 操作，这段是检测是否有发送权限 hook 操作， Hook 接口为 org.apache.rocketmq.client.hook.CheckForbiddenHook， 注意：在 DefaultMQProducerImpl 中，该类是以列表形式存在的
if (hasCheckForbiddenHook()) {
CheckForbiddenContext checkForbiddenContext = new CheckForbiddenContext();
checkForbiddenContext.setNameSrvAddr(this.defaultMQProducer.getNamesrvAddr());
checkForbiddenContext.setGroup(this.defaultMQProducer.getProducerGroup());
checkForbiddenContext.setCommunicationMode(communicationMode);
checkForbiddenContext.setBrokerAddr(brokerAddr);
checkForbiddenContext.setMessage(msg);
checkForbiddenContext.setMq(mq);
checkForbiddenContext.setUnitMode(this.isUnitMode());
this.executeCheckForbiddenHook(checkForbiddenContext);
}
// hook 操作，这段是执行发送消息前的 hook 操作， Hook 接口为 org.apache.rocketmq.client.hook.SendMessageHook， 注意：在 DefaultMQProducerImpl 中，该类是以列表形式存在的
if (this.hasSendMessageHook()) {
context = new SendMessageContext();
context.setProducer(this);
context.setProducerGroup(this.defaultMQProducer.getProducerGroup());
context.setCommunicationMode(communicationMode);
context.setBornHost(this.defaultMQProducer.getClientIP());
context.setBrokerAddr(brokerAddr);
context.setMessage(msg);
context.setMq(mq);
context.setNamespace(this.defaultMQProducer.getNamespace());
String isTrans = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
if (isTrans != null && isTrans.equals("true")) {
context.setMsgType(MessageType.Trans_Msg_Half);
}
if (msg.getProperty("__STARTDELIVERTIME") != null || msg.getProperty(MessageConst.PROPERTY_DELAY_TIME_LEVEL) != null) {
context.setMsgType(MessageType.Delay_Msg);
}
this.executeSendMessageHookBefore(context);
}
// 设置 request header
SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
requestHeader.setTopic(msg.getTopic());
requestHeader.setDefaultTopic(this.defaultMQProducer.getCreateTopicKey());
requestHeader.setDefaultTopicQueueNums(this.defaultMQProducer.getDefaultTopicQueueNums());
requestHeader.setQueueId(mq.getQueueId());
requestHeader.setSysFlag(sysFlag);
requestHeader.setBornTimestamp(System.currentTimeMillis());
requestHeader.setFlag(msg.getFlag());
requestHeader.setProperties(MessageDecoder.messageProperties2String(msg.getProperties()));
requestHeader.setReconsumeTimes(0);
requestHeader.setUnitMode(this.isUnitMode());
requestHeader.setBatch(msg instanceof MessageBatch);
if (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
String reconsumeTimes = MessageAccessor.getReconsumeTime(msg);
if (reconsumeTimes != null) {
requestHeader.setReconsumeTimes(Integer.valueOf(reconsumeTimes));
MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_RECONSUME_TIME);
}
String maxReconsumeTimes = MessageAccessor.getMaxReconsumeTimes(msg);
if (maxReconsumeTimes != null) {
requestHeader.setMaxReconsumeTimes(Integer.valueOf(maxReconsumeTimes));
MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_MAX_RECONSUME_TIMES);
}
}
SendResult sendResult = null;
switch (communicationMode) {
case ASYNC:
Message tmpMessage = msg;
boolean messageCloned = false;
if (msgBodyCompressed) {
//If msg body was compressed, msgbody should be reset using prevBody.
//Clone new message using commpressed message body and recover origin massage.
//Fix bug:https://github.com/apache/rocketmq-externals/issues/66
tmpMessage = MessageAccessor.cloneMessage(msg);
messageCloned = true;
msg.setBody(prevBody);
}
if (topicWithNamespace) {
if (!messageCloned) {
tmpMessage = MessageAccessor.cloneMessage(msg);
messageCloned = true;
}
msg.setTopic(NamespaceUtil.withoutNamespace(msg.getTopic(), this.defaultMQProducer.getNamespace()));
}
long costTimeAsync = System.currentTimeMillis() - beginStartTime;
if (timeout < costTimeAsync) {
throw new RemotingTooMuchRequestException("sendKernelImpl call timeout");
}
// 异步发送消息
sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
brokerAddr,
mq.getBrokerName(),
tmpMessage,
requestHeader,
timeout - costTimeAsync,
communicationMode,
sendCallback,
topicPublishInfo,
this.mQClientFactory,
this.defaultMQProducer.getRetryTimesWhenSendAsyncFailed(),
context,
this);
break;
case ONEWAY:
case SYNC:
long costTimeSync = System.currentTimeMillis() - beginStartTime;
if (timeout < costTimeSync) {
throw new RemotingTooMuchRequestException("sendKernelImpl call timeout");
}
// oneway 或同步发送消息
sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
brokerAddr,
mq.getBrokerName(),
msg,
requestHeader,
timeout - costTimeSync,
communicationMode,
context,
this);
break;
default:
assert false;
break;
}
// hook 操作，这段是执行发送消息后的 hook 操作， Hook 接口为 org.apache.rocketmq.client.hook.SendMessageHook， 注意：在 DefaultMQProducerImpl 中，该类是以列表形式存在的
if (this.hasSendMessageHook()) {
context.setSendResult(sendResult);
this.executeSendMessageHookAfter(context);
}
return sendResult;
} catch (RemotingException e) {
if (this.hasSendMessageHook()) {
context.setException(e);
this.executeSendMessageHookAfter(context);
}
throw e;
} catch (MQBrokerException e) {
if (this.hasSendMessageHook()) {
context.setException(e);
this.executeSendMessageHookAfter(context);
}
throw e;
} catch (InterruptedException e) {
if (this.hasSendMessageHook()) {
context.setException(e);
this.executeSendMessageHookAfter(context);
}
throw e;
} finally {
msg.setBody(prevBody);
msg.setTopic(NamespaceUtil.withoutNamespace(msg.getTopic(), this.defaultMQProducer.getNamespace()));
}
}
throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
}Copy the code private SendResult sendDefaultImpl(
Message msg,
final CommunicationMode communicationMode,
final SendCallback sendCallback,
final long timeout
) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
// 确保 Producer 状态为 RUNNING 态，所有状态可查看 org.apache.rocketmq.common.ServiceState 枚举类
this.makeSureStateOK();
// 校验消息是否符合规则，该工具类是比较好的参数校验封装形式，可以参考借鉴
Validators.checkMessage(msg, this.defaultMQProducer);
final long invokeID = random.nextLong();
// 第一次执行发送消息前的时间戳
long beginTimestampFirst = System.currentTimeMillis();
// 当前次发送消息前的时间戳
long beginTimestampPrev = beginTimestampFirst;
// 当前次发送消息后的时间戳
long endTimestamp = beginTimestampFirst;
// 从 NameServer 获取 topic 相关信息，包含 topic 中的 queue 相关信息； queue 路由相关信息
TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
// 当 (topic 相关信息不为 null) 并且 (topic 中的 queue 列表不为 null 或者 空队列)
if (topicPublishInfo != null && topicPublishInfo.ok()) {
boolean callTimeout = false;
MessageQueue mq = null;
Exception exception = null;
SendResult sendResult = null;
// 当模式为 SYNC 时， 默认执行次数为 3 次，包含 1 次正常调用，2 次重试；其他只执行 1 次
int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;
int times = 0;
// 第几次发送对应的 broker 信息
String[] brokersSent = new String[timesTotal];
for (; times < timesTotal; times++) {
// 获取上次发送的 broker 名称
String lastBrokerName = null == mq ? null : mq.getBrokerName();
// 选择一个 queue 进行发送。有失败重试策略，默认使用 RoundRobin 算法，可以通过 DefaultMQProducer#setSendLatencyFaultEnable 设置启用 LatencyFault 策略
MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
if (mqSelected != null) {
mq = mqSelected;
brokersSent[times] = mq.getBrokerName();
try {
// 记录发送前的时间戳
beginTimestampPrev = System.currentTimeMillis();
if (times > 0) {
//Reset topic with namespace during resend.
msg.setTopic(this.defaultMQProducer.withNamespace(msg.getTopic()));
}
// 计算花费时间
long costTime = beginTimestampPrev - beginTimestampFirst;
// 花费时间 超过了 timeout ，则超时处理
if (timeout < costTime) {
callTimeout = true;
break;
}
// 发送消息
sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);
endTimestamp = System.currentTimeMillis();
// 当设置启用 LatencyFault 策略时，更新 FaultItem
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
// 根据模式，选择发送消息后的处理方式
switch (communicationMode) {
case ASYNC:
return null;
case ONEWAY:
return null;
case SYNC:
// 模式为 SYNC 时，会有重试处理
if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {
continue;
}
}
return sendResult;
default:
break;
}
// 以下代码为异常处理
} catch (RemotingException e) {
endTimestamp = System.currentTimeMillis();
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
log.warn(msg.toString());
exception = e;
continue;
} catch (MQClientException e) {
endTimestamp = System.currentTimeMillis();
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
log.warn(msg.toString());
exception = e;
continue;
} catch (MQBrokerException e) {
endTimestamp = System.currentTimeMillis();
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
log.warn(msg.toString());
exception = e;
switch (e.getResponseCode()) {
case ResponseCode.TOPIC_NOT_EXIST:
case ResponseCode.SERVICE_NOT_AVAILABLE:
case ResponseCode.SYSTEM_ERROR:
case ResponseCode.NO_PERMISSION:
case ResponseCode.NO_BUYER_ID:
case ResponseCode.NOT_IN_CURRENT_UNIT:
continue;
default:
if (sendResult != null) {
return sendResult;
}
throw e;
}
} catch (InterruptedException e) {
endTimestamp = System.currentTimeMillis();
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
log.warn(String.format("sendKernelImpl exception, throw exception, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
log.warn(msg.toString());
log.warn("sendKernelImpl exception", e);
log.warn(msg.toString());
throw e;
}
} else {
break;
}
}
if (sendResult != null) {
return sendResult;
}
String info = String.format("Send [%d] times, still failed, cost [%d]ms, Topic: %s, BrokersSent: %s",
times,
System.currentTimeMillis() - beginTimestampFirst,
msg.getTopic(),
Arrays.toString(brokersSent));
info += FAQUrl.suggestTodo(FAQUrl.SEND_MSG_FAILED);
MQClientException mqClientException = new MQClientException(info, exception);
if (callTimeout) {
throw new RemotingTooMuchRequestException("sendDefaultImpl call timeout");
}
if (exception instanceof MQBrokerException) {
mqClientException.setResponseCode(((MQBrokerException) exception).getResponseCode());
} else if (exception instanceof RemotingConnectException) {
mqClientException.setResponseCode(ClientErrorCode.CONNECT_BROKER_EXCEPTION);
} else if (exception instanceof RemotingTimeoutException) {
mqClientException.setResponseCode(ClientErrorCode.ACCESS_BROKER_TIMEOUT);
} else if (exception instanceof MQClientException) {
mqClientException.setResponseCode(ClientErrorCode.BROKER_NOT_EXIST_EXCEPTION);
}
throw mqClientException;
}
List<String> nsList = this.getmQClientFactory().getMQClientAPIImpl().getNameServerAddressList();
if (null == nsList || nsList.isEmpty()) {
throw new MQClientException(
"No name server address, please set it." + FAQUrl.suggestTodo(FAQUrl.NAME_SERVER_ADDR_NOT_EXIST_URL), null).setResponseCode(ClientErrorCode.NO_NAME_SERVER_EXCEPTION);
}
throw new MQClientException("No route info of this topic, " + msg.getTopic() + FAQUrl.suggestTodo(FAQUrl.NO_TOPIC_ROUTE_INFO),
null).setResponseCode(ClientErrorCode.NOT_FOUND_TOPIC_EXCEPTION);
}Copy the codepublic SendResult send(Message msg, long timeout) throws MQClientException, RemotingException, MQBrokerException, Return this.sendDefaultImpl(MSG, communicationMode.sync, null, timeout); } /** * DEFAULT SYNC ------------------------------------------------------- */ public SendResult send( Message msg) Throws MQClientException, RemotingException, MQBrokerException, InterruptedException {// Send a message, The default timeout time of 3000 ms return send (MSG, enclosing defaultMQProducer. GetSendMsgTimeout ()); }Copy the codeThis class uses the facade pattern, which is simply a facade class that encapsulates complex internal details and provides a unified invocation interface to the client.
Refer to the blogger’s previous post on Design Patterns Learning Notes — Facade Patterns for an extension.
/** * Send message in synchronous mode. This method returns only when the sending procedure totally completes. * </p> * * <strong>Warn:</strong> this method has internal retry-mechanism, that is, internal implementation will retry * {@link #retryTimesWhenSendFailed} times before claiming failure. As a result, multiple messages may potentially * delivered to broker(s). It's up to the application developers to resolve potential duplication issue. * * @param msg Message to send. * @return {@link SendResult} instance to inform senders details of the deliverable, say Message ID of the message, * {@link SendStatus} indicating broker storage/replication status, message queue sent to, etc. * @throws MQClientException if there is any client error. * @throws RemotingException if there is any network-tier error. * @throws MQBrokerException if there is any error with broker. * @throws InterruptedException if the sending thread is interrupted. */ @Override public SendResult send( Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {// Validators. CheckMessage (MSG, this); // Wrap namespace with the NamespaceUtil utility class, Logic to see org.apache.rocketmq.com mon. Protocol. NamespaceUtilTest# testWrapNamespace unit tests msg.setTopic(withNamespace(msg.getTopic())); / / send a message return this. DefaultMQProducerImpl. Send (MSG); }Copy the code@Test public void testSendMessageSync_Success() throws RemotingException, InterruptedException, MQBrokerException, MQClientException {// Mock gets routing information for a Topic from NameServer when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createTopicRoute()); // Sending messages SendResult SendResult = producer. Send (message); assertThat(sendResult.getSendStatus()).isEqualTo(SendStatus.SEND_OK); assertThat(sendResult.getOffsetMsgId()).isEqualTo("123"); assertThat(sendResult.getQueueOffset()).isEqualTo(456L); }Copy the codeThis article is published by OpenWrite!
This article is published by OpenWrite!