RocketMQ 高可用设计之消息发送重试

private SendResult sendDefaultImpl(    Message msg,    final CommunicationMode communicationMode,    final SendCallback sendCallback,    final long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {    this.makeSureStateOK();    Validators.checkMessage(msg, this.defaultMQProducer);
    final long invokeID = random.nextLong();    long beginTimestampFirst = System.currentTimeMillis();    long beginTimestampPrev = beginTimestampFirst;    long endTimestamp = beginTimestampFirst;    TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());    if (topicPublishInfo != null && topicPublishInfo.ok()) {        boolean callTimeout = false;        MessageQueue mq = null;        Exception exception = null;        SendResult sendResult = null;        int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;        int times = 0;        String[] brokersSent = new String[timesTotal];        for (; times < timesTotal; times++) {            String lastBrokerName = null == mq ? null : mq.getBrokerName();            MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);            if (mqSelected != null) {                mq = mqSelected;                brokersSent[times] = mq.getBrokerName();                try {                    beginTimestampPrev = System.currentTimeMillis();                    long costTime = beginTimestampPrev - beginTimestampFirst;                    if (timeout < costTime) {                        callTimeout = true;                        break;                    }
                    sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);                    endTimestamp = System.currentTimeMillis();                    this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);                    switch (communicationMode) {                        case ASYNC:                            return null;                        case ONEWAY:                            return null;                        case 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);}