RocketMQ源码解析(三)-Producer
2018-12-30 本文已影响0人
空挡
发送方式
producer发送消息支持3种方式，同步、异步和Oneway。
- 同步发送：客户端提交消息到broker后会等待返回结果，相对来说是最常用的方式。
- 异步发送：调用发送接口时会注册一个callback类，发送线程继续其它业务逻辑，producer在收到broker结果后回调。比较适合不想发送结果影响正常业务逻辑的情况。
- Oneway：Producer提交消息后，无论broker是否正常接收消息都不关心。适合于追求高吞吐、能容忍消息丢失的场景，比如日志收集。
发送实例
public class SyncProducer {
public static void main(String[] args) throws Exception {
//Instantiate with a producer group name.
DefaultMQProducer producer = new
DefaultMQProducer("please_rename_unique_group_name");
// Specify name server addresses.
producer.setNamesrvAddr("localhost:9876");
//Launch the instance.
producer.start();
for (int i = 0; i < 100; i++) {
//Create a message instance, specifying topic, tag and message body.
Message msg = new Message("TopicTest" /* Topic */,
"TagA" /* Tag */,
("Hello RocketMQ " +
i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
);
//Call send message to deliver message to one of brokers.
SendResult sendResult = producer.send(msg);
System.out.printf("%s%n", sendResult);
}
//Shut down once the producer instance is not longer in use.
producer.shutdown();
}
}
发送消息首先需要初始化一个DefaultMQProducer，设置group name和nameserv的地址。Producer启动后就可以往指定的topic发送消息。
MQProducer初始化
Producer的调用关系是
MQProducer -> DefaultMQProducer ->DefaultMQProducerImpl
DefaultMQProducer是一个Facade类，封装了DefaultMQProducerImpl内部实现。我们来看下Producer的启动过程，DefaultMQProducerImpl.start()。
public void start(final boolean startFactory) throws MQClientException {
switch (this.serviceState) {
case CREATE_JUST:
this.serviceState = ServiceState.START_FAILED;
//参数检查，不能使用系统默认的GroupName
this.checkConfig();
//设置clientInstanceName，使用进程ID (PID)
if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
this.defaultMQProducer.changeInstanceNameToPID();
}
// 初始化MQClientInstance,一个进程只会存在一个MQClientInstance， 设置clientId （IP@PID）
this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQProducer, rpcHook);
//将当前Producer注册进MQClientInsance，保证一个producerName值对应一个Producer
boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this);
if (!registerOK) {
this.serviceState = ServiceState.CREATE_JUST;
throw new MQClientException("The producer group[" + this.defaultMQProducer.getProducerGroup()
+ "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
null);
}
this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo());
//启动MQClientInstance
if (startFactory) {
mQClientFactory.start();
}
log.info("the producer [{}] start OK. sendMessageWithVIPChannel={}", this.defaultMQProducer.getProducerGroup(),
this.defaultMQProducer.isSendMessageWithVIPChannel());
//设置状态为RUNNING
this.serviceState = ServiceState.RUNNING;
break;
case RUNNING:
case START_FAILED:
case SHUTDOWN_ALREADY:
throw new MQClientException("The producer service state not OK, maybe started once, "
+ this.serviceState
+ FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
null);
default:
break;
}
//向所有broker发送一次心跳
this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
}
从上面的代码可以看出，start的过程主要就是初始化和启动一个MQClientInstance，将producer注册到instance中。我们来看下MQClientInstance的启动过程。
MQClientInstance启动过程
public void start() throws MQClientException {
synchronized (this) {
switch (this.serviceState) {
case CREATE_JUST:
this.serviceState = ServiceState.START_FAILED;
// 1、如果NameservAddr为空，尝试从http server获取nameserv的地址
if (null == this.clientConfig.getNamesrvAddr()) {
this.mQClientAPIImpl.fetchNameServerAddr();
}
// Start request-response channel
// 2、启动MQClientAPIImpl，初始化NettyClient
this.mQClientAPIImpl.start();
// 3、开启Client的定时任务
this.startScheduledTask();
// 4、Start pull service,开始处理PullRequest
this.pullMessageService.start();
// 5、Start rebalance service
this.rebalanceService.start();
// Start push service
//6、启动Client内置的producer
this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
log.info("the client factory [{}] start OK", this.clientId);
this.serviceState = ServiceState.RUNNING;
break;
case RUNNING:
break;
case SHUTDOWN_ALREADY:
break;
case START_FAILED:
throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null);
default:
break;
}
}
}
1、如果producer在初始化的时候没有设置nameserv的地址，则会尝试从一个http server获取nameserv。这个httpserver是可以配置的，这种方式非常适合于有统一配置中心的系统
3、这里开启的定时任务有以下几个：
1）获取nameserv地址，就是重复的做第1步，这样就可以动态切换nameserv的地址
2）从nameserv更新topicRouteInfo，对于producer来说topic的路由信息是最重要的
3）将缓存的broker信息和最新的topicRouteInfo做对比，清除已经下线的broker
4）向broker发送心跳
4 ~ 6，producer和consumer公用一个MQClientInstance的实现。这几步初始化是给consumer用的，后面讲consumer的时候再讲。
Producer启动完成以后，就可以发送消息了，下面我们来看下一条普通的message的发送过程
消息发送
Producer默认采用SYNC方式提交消息，消息提交给broker收到response后返回。方法是DefaultMQProducerImpl.send( Message msg)
/**
* DEFAULT SYNC -------------------------------------------------------
*/
public SendResult send(
Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
return send(msg, this.defaultMQProducer.getSendMsgTimeout());
}
public SendResult send(Message msg,
long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
return this.sendDefaultImpl(msg, CommunicationMode.SYNC, null, timeout);
}
private SendResult sendDefaultImpl(
Message msg,
final CommunicationMode communicationMode,
final SendCallback sendCallback,
final long timeout
) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
//1、参数检查，消息不能发给系统预留的topic，消息体是否超过最大长度
this.makeSureStateOK();
Validators.checkMessage(msg, this.defaultMQProducer);
final long invokeID = random.nextLong();
long beginTimestampFirst = System.currentTimeMillis();
long beginTimestampPrev = beginTimestampFirst;
long endTimestamp = beginTimestampFirst;
//2、根据消息的topic，获取该topic的路由信息
TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
if (topicPublishInfo != null && topicPublishInfo.ok()) {
boolean callTimeout = false;
....
//3、发送重试次数
int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;
int times = 0;
//用来缓存发送和重试中已经用过的broker
String[] brokersSent = new String[timesTotal];
for (; times < timesTotal; times++) {
String lastBrokerName = null == mq ? null : mq.getBrokerName();
//4、从所有topic可用queue中选择一个queue
MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
if (mqSelected != null) {//获取Queue成功
mq = mqSelected;
brokersSent[times] = mq.getBrokerName();
try {
...
//5、提交消息到mq
sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);
endTimestamp = System.currentTimeMillis();
//6、成功，更新本次调用时间到MQFaultStrategy中
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
switch (communicationMode) {
//异步和ONEWAY调用后就直接返回了
case ASYNC:
return null;
case ONEWAY:
return null;
//7、如果broker存储失败，判断是否要重试
case SYNC:
if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {
continue;
}
}
return sendResult;
default:
break;
}
// 8、调用接口异常，更新状态到MQFaultStrategy中
} catch (RemotingException e) {
endTimestamp = System.currentTimeMillis();
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
...
exception = e;
continue;
} catch (MQClientException e) {
endTimestamp = System.currentTimeMillis();
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
...
exception = e;
continue;
} catch (MQBrokerException e) {
endTimestamp = System.currentTimeMillis();
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
...
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);
...
throw e;
}
} else {
break;
}
}
//9、成功则返回结果
if (sendResult != null) {
return sendResult;
}
...
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;
}
...
}
从上面的发送逻辑可以看出，无论哪种发送方式，最终都是调用的sendDefaultImpl来提交消息。
第2步：获取topic的所有路由信息，详细逻辑后面讲
第3步：SYNC发送可以设置失败重试次数
第4步：因为每个topic会在集群的多个broker上存在多个queue，所以这里会选择一个合适的queue，也就是在producer端实现负载均衡的功能，详细逻辑后面讲
第6和8步：无论提交消息成功或者失败，都会更新结果到MQFaultStrategy中，也就是第4中选取queue时采用的策略
第7步：对于消息提交成功，不止有SUCCESS一种状态，还有别的情况下也会认为成功的，比如broker接收和处理消息成功了，但是写给slave失败了，或者数据落盘失败了等。针对于存储失败的情况，客户端可以选择是否要重新发送。
以上就是消息发送的整个流程，下面分解下每一步的实现
获取topic路由
private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {
TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
if (null == topicPublishInfo || !topicPublishInfo.ok()) {
this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());
//如果TopicPushlishInfo不存在，则会尝试从Nameserv更新topic路由信息
this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
topicPublishInfo = this.topicPublishInfoTable.get(topic);
}
if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {
return topicPublishInfo;
} else {
this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer);
topicPublishInfo = this.topicPublishInfoTable.get(topic);
return topicPublishInfo;
}
}
根据topic直接从内存的缓存中获取路由信息，缓存的更新在前面的定时任务已经讲过。
如果TopicPushlishInfo不存在，则会尝试从Nameserv更新信息。更新策略是：
1）按topicName去nameserv找指定topic的route信息；
2）如果第一步没获取到则尝试获取默认创建topic（TBW102）的route信息，前提是broker支持默认创建。
最终，如果没有获取到topic的route信息，则报错中止消息发送
Queue选取策略
选择Queue
Queue的选取是发送端实现负责均衡的核心，根据client是否开启了延时容错，实现轮询和加可用性轮询的选取策略。
public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
//直接调用MQFaultStrategy的方法
return this.mqFaultStrategy.selectOneMessageQueue(tpInfo, lastBrokerName);
}
//MQFaultStrategy的方法实现
public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
//开启了延时容错
if (this.sendLatencyFaultEnable) {
try {
//1、首先获取上次使用的Queue index+1
int index = tpInfo.getSendWhichQueue().getAndIncrement();
for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
if (pos < 0)
pos = 0;
//2、找到index对应的queue
MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
//3、如果queue对应的broker可用，则使用该broker
if (latencyFaultTolerance.isAvailable(mq.getBrokerName())) {
if (null == lastBrokerName || mq.getBrokerName().equals(lastBrokerName))
return mq;
}
}
//4、如果上一步没找个合适的broker，则从所有的broker中选择一个相对合适的，并且broker是可写的。
final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
if (writeQueueNums > 0) {
final MessageQueue mq = tpInfo.selectOneMessageQueue();
if (notBestBroker != null) {
mq.setBrokerName(notBestBroker);
mq.setQueueId(tpInfo.getSendWhichQueue().getAndIncrement() % writeQueueNums);
}
return mq;
} else {
latencyFaultTolerance.remove(notBestBroker);
}
} catch (Exception e) {
log.error("Error occurred when selecting message queue", e);
}
//5、如果以上都没找到，则直接按顺序选择下一个
return tpInfo.selectOneMessageQueue();
}
//6、未开启延时容错，直接按顺序选下一个
return tpInfo.selectOneMessageQueue(lastBrokerName);
}
Producer为每个topic缓存了一个全局index，每次发送之后+1,然后从所有queue列表中选择index位置上的queue，这样就实现了轮询的效果。
如果开启了延时容错，则会考虑broker的可用性：
第1) 2)步：根据全局index找到queue
第3)步：如果根据延时容错判断queue所在的broker当前可用，并且是第一次发送，或者是重试并且和上次用的broker是同一个，则使用这个queue。这里面有两个逻辑，一个是broker的可用性是如何判断的，这个我们下面说；第二个是为什么重试的时候要选上次的broker，下面说下我的理解。
由前面的发送逻辑中的第6和8步知道，有两种情况会重试，一种是broker返回处理成功但是store失败，一种是broker返回失败。
对于返回失败的情况，其实会直接更新broker为短时不可用状态,这个在第一个if条件就已经通不过了；而对于store失败的情况，说明broker当前是正常的，重发还是发给同一个broker有利于防止消息重复。
第4)步：如果将所有queue按照第3)步的情况过一遍，发现都不符合条件，则从所有broker中选择一个相对好的。
第5)步：如果第4不中的broker不支持写入，则跟未开启延时容错一样的逻辑，直接选下一个queue
Broker延时控制逻辑
由上面的queue的选择策略可以知道，queue的选择除了轮询以外，就是根据Broker的可用性。回看下消息发送的第6步和第8步，在消息发送后会更新时间和发送状态到MQFaultStrategy中,代码如下：
public void updateFaultItem(final String brokerName, final long currentLatency, boolean isolation) {
if (this.sendLatencyFaultEnable) {
//1、根据发送结果，计算broker不可用时长
long duration = computeNotAvailableDuration(isolation ? 30000 : currentLatency);
//2、更新Broker不可用时长
this.latencyFaultTolerance.updateFaultItem(brokerName, currentLatency, duration);
}
}
第1步：根据上次消息发送时长和结果，计算Broker应该多长时间不可用，如果上次发送失败的话，发送时长按30秒计算。
MQFaultStrategy维护了一个broker延时列表，如下：
private long[] latencyMax = {50L, 100L, 550L, 1000L, 2000L, 3000L, 15000L};
private long[] notAvailableDuration = {0L, 0L, 30000L, 60000L, 120000L, 180000L, 600000L};
以上两个列表是一一对应的，当发送时长低于100ms时，设置broker不可用时长为0，之后依次增加，如果超过15秒，则有10分钟不可用。可以看到如果上次发送失败的话，也是10分钟不可用，如果重试肯定不会选择相同的broker。
消息提交sendKernelImpl()
Producer发送消息最终是调用sendKernelImpl()完成提交的，代码如下：
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的地址
String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
//Double check,如果地址为空，则从nameserv中再获取一次
if (null == brokerAddr) {
tryToFindTopicPublishInfo(mq.getTopic());
brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
}
SendMessageContext context = null;
if (brokerAddr != null) {
//切换到VIP channel
//Broker启动时会开启2个端口接收客户端数据，其中一个端口只接收producer的消息，
//不接受consumer的拉取请求，被称为VIP channel
brokerAddr = MixAll.brokerVIPChannel(this.defaultMQProducer.isSendMessageWithVIPChannel(), brokerAddr);
byte[] prevBody = msg.getBody();
try {
//for MessageBatch,ID has been set in the generating process
//客户端设置的id
if (!(msg instanceof MessageBatch)) {
MessageClientIDSetter.setUniqID(msg);
}
//如果消息body过长，则压缩并设置标记位
int sysFlag = 0;
boolean msgBodyCompressed = false;
if (this.tryToCompressMessage(msg)) {
sysFlag |= MessageSysFlag.COMPRESSED_FLAG;
msgBodyCompressed = true;
}
final String tranMsg = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
if (tranMsg != null && Boolean.parseBoolean(tranMsg)) {
sysFlag |= MessageSysFlag.TRANSACTION_PREPARED_TYPE;
}
//回调Forbidden Hook
if (hasCheckForbiddenHook()) {
...
...
}
// 回调SendMessage Hook
if (this.hasSendMessageHook()) {
...
...
}
//设置消息头
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);
//要求重新发送的消息，设置重试次数和延时时间
//仅Consumer用
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);
}
}
//通过NettyClient发送消息到Broker
SendResult sendResult = null;
switch (communicationMode) {
case ASYNC:
Message tmpMessage = msg;
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);
msg.setBody(prevBody);
}
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");
}
sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
brokerAddr,
mq.getBrokerName(),
msg,
requestHeader,
timeout - costTimeSync,
communicationMode,
context,
this);
break;
default:
assert false;
break;
}
//回调Send message Hook
if (this.hasSendMessageHook()) {
context.setSendResult(sendResult);
this.executeSendMessageHookAfter(context);
}
return sendResult;
} catch (RemotingException e) {
...
} catch (MQBrokerException e) {
...
} catch (InterruptedException e) {
...
} finally {
msg.setBody(prevBody);
}
}
//Broker地址获取失败，抛出异常中止发送
throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
}
最后一步，通过前面选择的queue和broker获取broker 地址，封装消息包并发送到broker，客户端支持单条消息发送，也支持多条消息封装到一个包中发送。Client会和broker保持长连接，提高发送速度。
