[聚合文章] 分布式消息队列 RocketMQ 源码分析 —— Message 发送与接收

1: @Override
2: public SendResult send(Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException{
3: return this.defaultMQProducerImpl.send(msg);
4: }

 1: public SendResult send(Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException{
 2: return send(msg, this.defaultMQProducer.getSendMsgTimeout());
 3: }
 4: 
 5: public SendResult send(Message msg, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException{
 6: return this.sendDefaultImpl(msg, CommunicationMode.SYNC, null, timeout);
 7: }
 8: 
 9: private SendResult sendDefaultImpl(//
10: Message msg, //
11: final CommunicationMode communicationMode, //
12: final SendCallback sendCallback, //
13: final long timeout//
14: ) throws MQClientException, RemotingException, MQBrokerException, InterruptedException{
 15: // 校验 Producer 处于运行状态
 16: this.makeSureStateOK();
 17: // 校验消息格式
 18: Validators.checkMessage(msg, this.defaultMQProducer);
 19: //
 20: final long invokeID = random.nextLong(); // 调用编号；用于下面打印日志，标记为同一次发送消息
 21: long beginTimestampFirst = System.currentTimeMillis();
 22: long beginTimestampPrev = beginTimestampFirst;
 23: long endTimestamp = beginTimestampFirst;
 24: // 获取 Topic路由信息
 25: TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
 26: if (topicPublishInfo != null && topicPublishInfo.ok()) {
 27: MessageQueue mq = null; // 最后选择消息要发送到的队列
 28: Exception exception = null;
 29: SendResult sendResult = null; // 最后一次发送结果
 30: int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1; // 同步多次调用
 31: int times = 0; // 第几次发送
 32: String[] brokersSent = new String[timesTotal]; // 存储每次发送消息选择的broker名
 33: // 循环调用发送消息，直到成功
 34: for (; times < timesTotal; times++) {
 35: String lastBrokerName = null == mq ? null : mq.getBrokerName();
 36: MessageQueue tmpmq = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName); // 选择消息要发送到的队列
 37: if (tmpmq != null) {
 38: mq = tmpmq;
 39: brokersSent[times] = mq.getBrokerName();
 40: try {
 41: beginTimestampPrev = System.currentTimeMillis();
 42: // 调用发送消息核心方法
 43: sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout);
 44: endTimestamp = System.currentTimeMillis();
 45: // 更新Broker可用性信息
 46: this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
 47: switch (communicationMode) {
 48: case ASYNC:
 49: return null;
 50: case ONEWAY:
 51: return null;
 52: case SYNC:
 53: if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
 54: if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) { // 同步发送成功但存储有问题时 && 配置存储异常时重新发送开关 时，进行重试
 55: continue;
 56: }
 57: }
 58: return sendResult;
 59: default:
 60: break;
 61: }
 62: } catch (RemotingException e) { // 打印异常，更新Broker可用性信息，更新继续循环
 63: endTimestamp = System.currentTimeMillis();
 64: this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
 65: log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
 66: log.warn(msg.toString());
 67: exception = e;
 68: continue;
 69: } catch (MQClientException e) { // 打印异常，更新Broker可用性信息，继续循环
 70: endTimestamp = System.currentTimeMillis();
 71: this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
 72: log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
 73: log.warn(msg.toString());
 74: exception = e;
 75: continue;
 76: } catch (MQBrokerException e) { // 打印异常，更新Broker可用性信息，部分情况下的异常，直接返回，结束循环
 77: endTimestamp = System.currentTimeMillis();
 78: this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
 79: log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
 80: log.warn(msg.toString());
 81: exception = e;
 82: switch (e.getResponseCode()) {
 83: // 如下异常continue，进行发送消息重试
 84: case ResponseCode.TOPIC_NOT_EXIST:
 85: case ResponseCode.SERVICE_NOT_AVAILABLE:
 86: case ResponseCode.SYSTEM_ERROR:
 87: case ResponseCode.NO_PERMISSION:
 88: case ResponseCode.NO_BUYER_ID:
 89: case ResponseCode.NOT_IN_CURRENT_UNIT:
 90: continue;
 91: // 如果有发送结果，进行返回，否则，抛出异常；
 92: default:
 93: if (sendResult != null) {
 94: return sendResult;
 95: }
 96: throw e;
 97: }
 98: } catch (InterruptedException e) {
 99: endTimestamp = System.currentTimeMillis();
100: this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
101: log.warn(String.format("sendKernelImpl exception, throw exception, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
102: log.warn(msg.toString());
103: throw e;
104: }
105: } else {
106: break;
107: }
108: }
109: // 返回发送结果
110: if (sendResult != null) {
111: return sendResult;
112: }
113: // 根据不同情况，抛出不同的异常
114: String info = String.format("Send [%d] times, still failed, cost [%d]ms, Topic: %s, BrokersSent: %s", times, System.currentTimeMillis() - beginTimestampFirst,
115: msg.getTopic(), Arrays.toString(brokersSent)) + FAQUrl.suggestTodo(FAQUrl.SEND_MSG_FAILED);
116: MQClientException mqClientException = new MQClientException(info, exception);
117: if (exception instanceof MQBrokerException) {
118: mqClientException.setResponseCode(((MQBrokerException) exception).getResponseCode());
119: } else if (exception instanceof RemotingConnectException) {
120: mqClientException.setResponseCode(ClientErrorCode.CONNECT_BROKER_EXCEPTION);
121: } else if (exception instanceof RemotingTimeoutException) {
122: mqClientException.setResponseCode(ClientErrorCode.ACCESS_BROKER_TIMEOUT);
123: } else if (exception instanceof MQClientException) {
124: mqClientException.setResponseCode(ClientErrorCode.BROKER_NOT_EXIST_EXCEPTION);
125: }
126: throw mqClientException;
127: }
128: // Namesrv找不到异常
129: List<String> nsList = this.getmQClientFactory().getMQClientAPIImpl().getNameServerAddressList();
130: if (null == nsList || nsList.isEmpty()) {
131: throw new MQClientException(
132: "No name server address, please set it." + FAQUrl.suggestTodo(FAQUrl.NAME_SERVER_ADDR_NOT_EXIST_URL), null).setResponseCode(ClientErrorCode.NO_NAME_SERVER_EXCEPTION);
133: }
134: // 消息路由找不到异常
135: throw new MQClientException("No route info of this topic, " + msg.getTopic() + FAQUrl.suggestTodo(FAQUrl.NO_TOPIC_ROUTE_INFO),
136: null).setResponseCode(ClientErrorCode.NOT_FOUND_TOPIC_EXCEPTION);
137: }

 1: private TopicPublishInfo tryToFindTopicPublishInfo(final String topic){
 2: // 缓存中获取 Topic发布信息
 3: TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
 4: // 当无可用的 Topic发布信息时，从Namesrv获取一次
 5: if (null == topicPublishInfo || !topicPublishInfo.ok()) {
 6: this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());
 7: this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
 8: topicPublishInfo = this.topicPublishInfoTable.get(topic);
 9: }
10: // 若获取的 Topic发布信息时候可用，则返回
11: if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {
12: return topicPublishInfo;
13: } else { // 使用 {@link DefaultMQProducer#createTopicKey} 对应的 Topic发布信息。用于 Topic发布信息不存在 && Broker支持自动创建Topic
14: this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer);
15: topicPublishInfo = this.topicPublishInfoTable.get(topic);
16: return topicPublishInfo;
17: }
18: }

 1: public class MQFaultStrategy{
 2: private final static Logger log = ClientLogger.getLog();
 3: 
 4: /**
5: * 延迟故障容错，维护每个Broker的发送消息的延迟
6: * key：brokerName
7: */
 8: private final LatencyFaultTolerance<String> latencyFaultTolerance = new LatencyFaultToleranceImpl();
 9: /**
10: * 发送消息延迟容错开关
11: */
12: private boolean sendLatencyFaultEnable = false;
13: /**
14: * 延迟级别数组
15: */
16: private long[] latencyMax = {50L, 100L, 550L, 1000L, 2000L, 3000L, 15000L};
17: /**
18: * 不可用时长数组
19: */
20: private long[] notAvailableDuration = {0L, 0L, 30000L, 60000L, 120000L, 180000L, 600000L};
21: 
22: /**
23: * 根据 Topic发布信息 选择一个消息队列
24: *
25: * @param tpInfo Topic发布信息
26: * @param lastBrokerName brokerName
27: * @return 消息队列
28: */
29: public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName){
30: if (this.sendLatencyFaultEnable) {
31: try {
32: // 获取 brokerName=lastBrokerName && 可用的一个消息队列
33: int index = tpInfo.getSendWhichQueue().getAndIncrement();
34: for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
35: int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
36: if (pos < 0)
37: pos = 0;
38: MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
39: if (latencyFaultTolerance.isAvailable(mq.getBrokerName())) {
40: if (null == lastBrokerName || mq.getBrokerName().equals(lastBrokerName))
41: return mq;
42: }
43: }
44: // 选择一个相对好的broker，并获得其对应的一个消息队列，不考虑该队列的可用性
45: final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
46: int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
47: if (writeQueueNums > 0) {
48: final MessageQueue mq = tpInfo.selectOneMessageQueue();
49: if (notBestBroker != null) {
50: mq.setBrokerName(notBestBroker);
51: mq.setQueueId(tpInfo.getSendWhichQueue().getAndIncrement() % writeQueueNums);
52: }
53: return mq;
54: } else {
55: latencyFaultTolerance.remove(notBestBroker);
56: }
57: } catch (Exception e) {
58: log.error("Error occurred when selecting message queue", e);
59: }
60: // 选择一个消息队列，不考虑队列的可用性
61: return tpInfo.selectOneMessageQueue();
62: }
63: // 获得 lastBrokerName 对应的一个消息队列，不考虑该队列的可用性
64: return tpInfo.selectOneMessageQueue(lastBrokerName);
65: }
66: 
67: /**
68: * 更新延迟容错信息
69: *
70: * @param brokerName brokerName
71: * @param currentLatency 延迟
72: * @param isolation 是否隔离。当开启隔离时，默认延迟为30000。目前主要用于发送消息异常时
73: */
74: public void updateFaultItem(final String brokerName, final long currentLatency, boolean isolation){
75: if (this.sendLatencyFaultEnable) {
76: long duration = computeNotAvailableDuration(isolation ? 30000 : currentLatency);
77: this.latencyFaultTolerance.updateFaultItem(brokerName, currentLatency, duration);
78: }
79: }
80: 
81: /**
82: * 计算延迟对应的不可用时间
83: *
84: * @param currentLatency 延迟
85: * @return 不可用时间
86: */
87: private long computeNotAvailableDuration(final long currentLatency){
88: for (int i = latencyMax.length - 1; i >= 0; i--) {
89: if (currentLatency >= latencyMax[i])
90: return this.notAvailableDuration[i];
91: }
92: return 0;
93: }

 1: public interface LatencyFaultTolerance<T>{
 2: 
 3: /**
4: * 更新对应的延迟和不可用时长
5: *
6: * @param name 对象
7: * @param currentLatency 延迟
8: * @param notAvailableDuration 不可用时长
9: */
10: void updateFaultItem(final T name, final long currentLatency, final long notAvailableDuration);
11: 
12: /**
13: * 对象是否可用
14: *
15: * @param name 对象
16: * @return 是否可用
17: */
18: boolean isAvailable(final T name);
19: 
20: /**
21: * 移除对象
22: *
23: * @param name 对象
24: */
25: void remove(final T name);
26: 
27: /**
28: * 获取一个对象
29: *
30: * @return 对象
31: */
32: T pickOneAtLeast();
33: }

 1: public class LatencyFaultToleranceImpl implements LatencyFaultTolerance<String>{
 2: 
 3: /**
4: * 对象故障信息Table
5: */
 6: private final ConcurrentHashMap<String, FaultItem> faultItemTable = new ConcurrentHashMap<>(16);
 7: /**
8: * 对象选择Index
9: * @see #pickOneAtLeast()
10: */
11: private final ThreadLocalIndex whichItemWorst = new ThreadLocalIndex();
12: 
13: @Override
14: public void updateFaultItem(final String name, final long currentLatency, final long notAvailableDuration){
15: FaultItem old = this.faultItemTable.get(name);
16: if (null == old) {
17: // 创建对象
18: final FaultItem faultItem = new FaultItem(name);
19: faultItem.setCurrentLatency(currentLatency);
20: faultItem.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);
21: // 更新对象
22: old = this.faultItemTable.putIfAbsent(name, faultItem);
23: if (old != null) {
24: old.setCurrentLatency(currentLatency);
25: old.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);
26: }
27: } else { // 更新对象
28: old.setCurrentLatency(currentLatency);
29: old.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);
30: }
31: }
32: 
33: @Override
34: public boolean isAvailable(final String name){
35: final FaultItem faultItem = this.faultItemTable.get(name);
36: if (faultItem != null) {
37: return faultItem.isAvailable();
38: }
39: return true;
40: }
41: 
42: @Override
43: public void remove(final String name){
44: this.faultItemTable.remove(name);
45: }
46: 
47: /**
48: * 选择一个相对优秀的对象
49: *
50: * @return 对象
51: */
52: @Override
53: public String pickOneAtLeast(){
54: // 创建数组
55: final Enumeration<FaultItem> elements = this.faultItemTable.elements();
56: List<FaultItem> tmpList = new LinkedList<>();
57: while (elements.hasMoreElements()) {
58: final FaultItem faultItem = elements.nextElement();
59: tmpList.add(faultItem);
60: }
61: //
62: if (!tmpList.isEmpty()) {
63: // 打乱 + 排序。TODO 疑问：应该只能二选一。猜测Collections.shuffle(tmpList)去掉。
64: Collections.shuffle(tmpList);
65: Collections.sort(tmpList);
66: // 选择顺序在前一半的对象
67: final int half = tmpList.size() / 2;
68: if (half <= 0) {
69: return tmpList.get(0).getName();
70: } else {
71: final int i = this.whichItemWorst.getAndIncrement() % half;
72: return tmpList.get(i).getName();
73: }
74: }
75: return null;
76: }
77: }

 1: class FaultItem implements Comparable<FaultItem>{
 2: /**
3: * 对象名
4: */
 5: private final String name;
 6: /**
7: * 延迟
8: */
 9: private volatile long currentLatency;
10: /**
11: * 开始可用时间
12: */
13: private volatile long startTimestamp;
14: 
15: public FaultItem(final String name){
16: this.name = name;
17: }
18: 
19: /**
20: * 比较对象
21: * 可用性 > 延迟 > 开始可用时间
22: *
23: * @param other other
24: * @return 升序
25: */
26: @Override
27: public int compareTo(final FaultItem other){
28: if (this.isAvailable() != other.isAvailable()) {
29: if (this.isAvailable())
30: return -1;
31: 
32: if (other.isAvailable())
33: return 1;
34: }
35: 
36: if (this.currentLatency < other.currentLatency)
37: return -1;
38: else if (this.currentLatency > other.currentLatency) {
39: return 1;
40: }
41: 
42: if (this.startTimestamp < other.startTimestamp)
43: return -1;
44: else if (this.startTimestamp > other.startTimestamp) {
45: return 1;
46: }
47: 
48: return 0;
49: }
50: 
51: /**
52: * 是否可用：当开始可用时间大于当前时间
53: *
54: * @return 是否可用
55: */
56: public boolean isAvailable(){
57: return (System.currentTimeMillis() - startTimestamp) >= 0;
58: }
59: 
60: @Override
61: public int hashCode(){
62: int result = getName() != null ? getName().hashCode() : 0;
63: result = 31 * result + (int) (getCurrentLatency() ^ (getCurrentLatency() >>> 32));
64: result = 31 * result + (int) (getStartTimestamp() ^ (getStartTimestamp() >>> 32));
65: return result;
66: }
67: 
68: @Override
69: public boolean equals(final Object o){
70: if (this == o)
71: return true;
72: if (!(o instanceof FaultItem))
73: return false;
74: 
75: final FaultItem faultItem = (FaultItem) o;
76: 
77: if (getCurrentLatency() != faultItem.getCurrentLatency())
78: return false;
79: if (getStartTimestamp() != faultItem.getStartTimestamp())
80: return false;
81: return getName() != null ? getName().equals(faultItem.getName()) : faultItem.getName() == null;
82: 
83: }
84: }

 1: private SendResult sendKernelImpl(final Message msg, //
2: final MessageQueue mq, //
3: final CommunicationMode communicationMode, //
4: final SendCallback sendCallback, //
5: final TopicPublishInfo topicPublishInfo, //
6: final long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException{
 7: // 获取 broker地址
 8: String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
 9: if (null == brokerAddr) {
 10: tryToFindTopicPublishInfo(mq.getTopic());
 11: brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
 12: }
 13: //
 14: SendMessageContext context = null;
 15: if (brokerAddr != null) {
 16: // 是否使用broker vip通道。broker会开启两个端口对外服务。
 17: brokerAddr = MixAll.brokerVIPChannel(this.defaultMQProducer.isSendMessageWithVIPChannel(), brokerAddr);
 18: byte[] prevBody = msg.getBody(); // 记录消息内容。下面逻辑可能改变消息内容，例如消息压缩。
 19: try {
 20: // 设置唯一编号
 21: MessageClientIDSetter.setUniqID(msg);
 22: // 消息压缩
 23: int sysFlag = 0;
 24: if (this.tryToCompressMessage(msg)) {
 25: sysFlag |= MessageSysFlag.COMPRESSED_FLAG;
 26: }
 27: // 事务
 28: final String tranMsg = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
 29: if (tranMsg != null && Boolean.parseBoolean(tranMsg)) {
 30: sysFlag |= MessageSysFlag.TRANSACTION_PREPARED_TYPE;
 31: }
 32: // hook：发送消息校验
 33: if (hasCheckForbiddenHook()) {
 34: CheckForbiddenContext checkForbiddenContext = new CheckForbiddenContext();
 35: checkForbiddenContext.setNameSrvAddr(this.defaultMQProducer.getNamesrvAddr());
 36: checkForbiddenContext.setGroup(this.defaultMQProducer.getProducerGroup());
 37: checkForbiddenContext.setCommunicationMode(communicationMode);
 38: checkForbiddenContext.setBrokerAddr(brokerAddr);
 39: checkForbiddenContext.setMessage(msg);
 40: checkForbiddenContext.setMq(mq);
 41: checkForbiddenContext.setUnitMode(this.isUnitMode());
 42: this.executeCheckForbiddenHook(checkForbiddenContext);
 43: }
 44: // hook：发送消息前逻辑
 45: if (this.hasSendMessageHook()) {
 46: context = new SendMessageContext();
 47: context.setProducer(this);
 48: context.setProducerGroup(this.defaultMQProducer.getProducerGroup());
 49: context.setCommunicationMode(communicationMode);
 50: context.setBornHost(this.defaultMQProducer.getClientIP());
 51: context.setBrokerAddr(brokerAddr);
 52: context.setMessage(msg);
 53: context.setMq(mq);
 54: String isTrans = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
 55: if (isTrans != null && isTrans.equals("true")) {
 56: context.setMsgType(MessageType.Trans_Msg_Half);
 57: }
 58: if (msg.getProperty("__STARTDELIVERTIME") != null || msg.getProperty(MessageConst.PROPERTY_DELAY_TIME_LEVEL) != null) {
 59: context.setMsgType(MessageType.Delay_Msg);
 60: }
 61: this.executeSendMessageHookBefore(context);
 62: }
 63: // 构建发送消息请求
 64: SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
 65: requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
 66: requestHeader.setTopic(msg.getTopic());
 67: requestHeader.setDefaultTopic(this.defaultMQProducer.getCreateTopicKey());
 68: requestHeader.setDefaultTopicQueueNums(this.defaultMQProducer.getDefaultTopicQueueNums());
 69: requestHeader.setQueueId(mq.getQueueId());
 70: requestHeader.setSysFlag(sysFlag);
 71: requestHeader.setBornTimestamp(System.currentTimeMillis());
 72: requestHeader.setFlag(msg.getFlag());
 73: requestHeader.setProperties(MessageDecoder.messageProperties2String(msg.getProperties()));
 74: requestHeader.setReconsumeTimes(0);
 75: requestHeader.setUnitMode(this.isUnitMode());
 76: if (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) { // 消息重发Topic
 77: String reconsumeTimes = MessageAccessor.getReconsumeTime(msg);
 78: if (reconsumeTimes != null) {
 79: requestHeader.setReconsumeTimes(Integer.valueOf(reconsumeTimes));
 80: MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_RECONSUME_TIME);
 81: }
 82: String maxReconsumeTimes = MessageAccessor.getMaxReconsumeTimes(msg);
 83: if (maxReconsumeTimes != null) {
 84: requestHeader.setMaxReconsumeTimes(Integer.valueOf(maxReconsumeTimes));
 85: MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_MAX_RECONSUME_TIMES);
 86: }
 87: }
 88: // 发送消息
 89: SendResult sendResult = null;
 90: switch (communicationMode) {
 91: case ASYNC:
 92: sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(//
 93: brokerAddr, // 1
 94: mq.getBrokerName(), // 2
 95: msg, // 3
 96: requestHeader, // 4
 97: timeout, // 5
 98: communicationMode, // 6
 99: sendCallback, // 7
100: topicPublishInfo, // 8
101: this.mQClientFactory, // 9
102: this.defaultMQProducer.getRetryTimesWhenSendAsyncFailed(), // 10
103: context, //
104: this);
105: break;
106: case ONEWAY:
107: case SYNC:
108: sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
109: brokerAddr,
110: mq.getBrokerName(),
111: msg,
112: requestHeader,
113: timeout,
114: communicationMode,
115: context,
116: this);
117: break;
118: default:
119: assert false;
120: break;
121: }
122: // hook：发送消息后逻辑
123: if (this.hasSendMessageHook()) {
124: context.setSendResult(sendResult);
125: this.executeSendMessageHookAfter(context);
126: }
127: // 返回发送结果
128: return sendResult;
129: } catch (RemotingException e) {
130: if (this.hasSendMessageHook()) {
131: context.setException(e);
132: this.executeSendMessageHookAfter(context);
133: }
134: throw e;
135: } catch (MQBrokerException e) {
136: if (this.hasSendMessageHook()) {
137: context.setException(e);
138: this.executeSendMessageHookAfter(context);
139: }
140: throw e;
141: } catch (InterruptedException e) {
142: if (this.hasSendMessageHook()) {
143: context.setException(e);
144: this.executeSendMessageHookAfter(context);
145: }
146: throw e;
147: } finally {
148: msg.setBody(prevBody);
149: }
150: }
151: // broker为空抛出异常
152: throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
153: }

 1: @Override
 2: public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException{
 3: SendMessageContext mqtraceContext;
 4: switch (request.getCode()) {
 5: case RequestCode.CONSUMER_SEND_MSG_BACK:
 6: return this.consumerSendMsgBack(ctx, request);
 7: default:
 8: // 解析请求
 9: SendMessageRequestHeader requestHeader = parseRequestHeader(request);
 10: if (requestHeader == null) {
 11: return null;
 12: }
 13: // 发送请求Context。在 hook 场景下使用
 14: mqtraceContext = buildMsgContext(ctx, requestHeader);
 15: // hook：处理发送消息前逻辑
 16: this.executeSendMessageHookBefore(ctx, request, mqtraceContext);
 17: // 处理发送消息逻辑
 18: final RemotingCommand response = this.sendMessage(ctx, request, mqtraceContext, requestHeader);
 19: // hook：处理发送消息后逻辑
 20: this.executeSendMessageHookAfter(response, mqtraceContext);
 21: return response;
 22: }
 23: }
 24: 
 25: private RemotingCommand sendMessage(final ChannelHandlerContext ctx, //
26: final RemotingCommand request, //
27: final SendMessageContext sendMessageContext, //
28: final SendMessageRequestHeader requestHeader) throws RemotingCommandException{
 29: 
 30: // 初始化响应
 31: final RemotingCommand response = RemotingCommand.createResponseCommand(SendMessageResponseHeader.class);
 32: final SendMessageResponseHeader responseHeader = (SendMessageResponseHeader) response.readCustomHeader();
 33: response.setOpaque(request.getOpaque());
 34: response.addExtField(MessageConst.PROPERTY_MSG_REGION, this.brokerController.getBrokerConfig().getRegionId());
 35: response.addExtField(MessageConst.PROPERTY_TRACE_SWITCH, String.valueOf(this.brokerController.getBrokerConfig().isTraceOn()));
 36: 
 37: if (log.isDebugEnabled()) {
 38: log.debug("receive SendMessage request command, {}", request);
 39: }
 40: 
 41: // 如果未开始接收消息，抛出系统异常
 42: @SuppressWarnings("SpellCheckingInspection")
 43: final long startTimstamp = this.brokerController.getBrokerConfig().getStartAcceptSendRequestTimeStamp();
 44: if (this.brokerController.getMessageStore().now() < startTimstamp) {
 45: response.setCode(ResponseCode.SYSTEM_ERROR);
 46: response.setRemark(String.format("broker unable to service, until %s", UtilAll.timeMillisToHumanString2(startTimstamp)));
 47: return response;
 48: }
 49: 
 50: // 消息配置(Topic配置）校验
 51: response.setCode(-1);
 52: super.msgCheck(ctx, requestHeader, response);
 53: if (response.getCode() != -1) {
 54: return response;
 55: }
 56: 
 57: final byte[] body = request.getBody();
 58: 
 59: // 如果队列小于0，从可用队列随机选择
 60: int queueIdInt = requestHeader.getQueueId();
 61: TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());
 62: if (queueIdInt < 0) {
 63: queueIdInt = Math.abs(this.random.nextInt() % 99999999) % topicConfig.getWriteQueueNums();
 64: }
 65: 
 66: //
 67: int sysFlag = requestHeader.getSysFlag();
 68: if (TopicFilterType.MULTI_TAG == topicConfig.getTopicFilterType()) {
 69: sysFlag |= MessageSysFlag.MULTI_TAGS_FLAG;
 70: }
 71: 
 72: // 对RETRY类型的消息处理。如果超过最大消费次数，则topic修改成"%DLQ%" + 分组名，即加入 死信队列(Dead Letter Queue)
 73: String newTopic = requestHeader.getTopic();
 74: if (null != newTopic && newTopic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
 75: // 获取订阅分组配置
 76: String groupName = newTopic.substring(MixAll.RETRY_GROUP_TOPIC_PREFIX.length());
 77: SubscriptionGroupConfig subscriptionGroupConfig =
 78: this.brokerController.getSubscriptionGroupManager().findSubscriptionGroupConfig(groupName);
 79: if (null == subscriptionGroupConfig) {
 80: response.setCode(ResponseCode.SUBSCRIPTION_GROUP_NOT_EXIST);
 81: response.setRemark("subscription group not exist, " + groupName + " " + FAQUrl.suggestTodo(FAQUrl.SUBSCRIPTION_GROUP_NOT_EXIST));
 82: return response;
 83: }
 84: // 计算最大可消费次数
 85: int maxReconsumeTimes = subscriptionGroupConfig.getRetryMaxTimes();
 86: if (request.getVersion() >= MQVersion.Version.V3_4_9.ordinal()) {
 87: maxReconsumeTimes = requestHeader.getMaxReconsumeTimes();
 88: }
 89: int reconsumeTimes = requestHeader.getReconsumeTimes() == null ? 0 : requestHeader.getReconsumeTimes();
 90: if (reconsumeTimes >= maxReconsumeTimes) { // 超过最大消费次数
 91: newTopic = MixAll.getDLQTopic(groupName);
 92: queueIdInt = Math.abs(this.random.nextInt() % 99999999) % DLQ_NUMS_PER_GROUP;
 93: topicConfig = this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(newTopic, //
 94: DLQ_NUMS_PER_GROUP, //
 95: PermName.PERM_WRITE, 0
 96: );
 97: if (null == topicConfig) {
 98: response.setCode(ResponseCode.SYSTEM_ERROR);
 99: response.setRemark("topic[" + newTopic + "] not exist");
100: return response;
101: }
102: }
103: }
104: 
105: // 创建MessageExtBrokerInner
106: MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
107: msgInner.setTopic(newTopic);
108: msgInner.setBody(body);
109: msgInner.setFlag(requestHeader.getFlag());
110: MessageAccessor.setProperties(msgInner, MessageDecoder.string2messageProperties(requestHeader.getProperties()));
111: msgInner.setPropertiesString(requestHeader.getProperties());
112: msgInner.setTagsCode(MessageExtBrokerInner.tagsString2tagsCode(topicConfig.getTopicFilterType(), msgInner.getTags()));
113: msgInner.setQueueId(queueIdInt);
114: msgInner.setSysFlag(sysFlag);
115: msgInner.setBornTimestamp(requestHeader.getBornTimestamp());
116: msgInner.setBornHost(ctx.channel().remoteAddress());
117: msgInner.setStoreHost(this.getStoreHost());
118: msgInner.setReconsumeTimes(requestHeader.getReconsumeTimes() == null ? 0 : requestHeader.getReconsumeTimes());
119: 
120: // 校验是否不允许发送事务消息
121: if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) {
122: String traFlag = msgInner.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
123: if (traFlag != null) {
124: response.setCode(ResponseCode.NO_PERMISSION);
125: response.setRemark(
126: "the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1() + "] sending transaction message is forbidden");
127: return response;
128: }
129: }
130: 
131: // 添加消息
132: PutMessageResult putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
133: if (putMessageResult != null) {
134: boolean sendOK = false;
135: 
136: switch (putMessageResult.getPutMessageStatus()) {
137: // Success
138: case PUT_OK:
139: sendOK = true;
140: response.setCode(ResponseCode.SUCCESS);
141: break;
142: case FLUSH_DISK_TIMEOUT:
143: response.setCode(ResponseCode.FLUSH_DISK_TIMEOUT);
144: sendOK = true;
145: break;
146: case FLUSH_SLAVE_TIMEOUT:
147: response.setCode(ResponseCode.FLUSH_SLAVE_TIMEOUT);
148: sendOK = true;
149: break;
150: case SLAVE_NOT_AVAILABLE:
151: response.setCode(ResponseCode.SLAVE_NOT_AVAILABLE);
152: sendOK = true;
153: break;
154: 
155: // Failed
156: case CREATE_MAPEDFILE_FAILED:
157: response.setCode(ResponseCode.SYSTEM_ERROR);
158: response.setRemark("create mapped file failed, server is busy or broken.");
159: break;
160: case MESSAGE_ILLEGAL:
161: case PROPERTIES_SIZE_EXCEEDED:
162: response.setCode(ResponseCode.MESSAGE_ILLEGAL);
163: response.setRemark(
164: "the message is illegal, maybe msg body or properties length not matched. msg body length limit 128k, msg properties length limit 32k.");
165: break;
166: case SERVICE_NOT_AVAILABLE:
167: response.setCode(ResponseCode.SERVICE_NOT_AVAILABLE);
168: response.setRemark(
169: "service not available now, maybe disk full, " + diskUtil() + ", maybe your broker machine memory too small.");
170: break;
171: case OS_PAGECACHE_BUSY:
172: response.setCode(ResponseCode.SYSTEM_ERROR);
173: response.setRemark("[PC_SYNCHRONIZED]broker busy, start flow control for a while");
174: break;
175: case UNKNOWN_ERROR:
176: response.setCode(ResponseCode.SYSTEM_ERROR);
177: response.setRemark("UNKNOWN_ERROR");
178: break;
179: default:
180: response.setCode(ResponseCode.SYSTEM_ERROR);
181: response.setRemark("UNKNOWN_ERROR DEFAULT");
182: break;
183: }
184: 
185: String owner = request.getExtFields().get(BrokerStatsManager.COMMERCIAL_OWNER);
186: if (sendOK) {
187: // 统计
188: this.brokerController.getBrokerStatsManager().incTopicPutNums(msgInner.getTopic());
189: this.brokerController.getBrokerStatsManager().incTopicPutSize(msgInner.getTopic(), putMessageResult.getAppendMessageResult().getWroteBytes());
190: this.brokerController.getBrokerStatsManager().incBrokerPutNums();
191: 
192: // 响应
193: response.setRemark(null);
194: responseHeader.setMsgId(putMessageResult.getAppendMessageResult().getMsgId());
195: responseHeader.setQueueId(queueIdInt);
196: responseHeader.setQueueOffset(putMessageResult.getAppendMessageResult().getLogicsOffset());
197: doResponse(ctx, request, response);
198: 
199: // hook：设置发送成功到context
200: if (hasSendMessageHook()) {
201: sendMessageContext.setMsgId(responseHeader.getMsgId());
202: sendMessageContext.setQueueId(responseHeader.getQueueId());
203: sendMessageContext.setQueueOffset(responseHeader.getQueueOffset());
204: 
205: int commercialBaseCount = brokerController.getBrokerConfig().getCommercialBaseCount();
206: int wroteSize = putMessageResult.getAppendMessageResult().getWroteBytes();
207: int incValue = (int) Math.ceil(wroteSize / BrokerStatsManager.SIZE_PER_COUNT) * commercialBaseCount;
208: 
209: sendMessageContext.setCommercialSendStats(BrokerStatsManager.StatsType.SEND_SUCCESS);
210: sendMessageContext.setCommercialSendTimes(incValue);
211: sendMessageContext.setCommercialSendSize(wroteSize);
212: sendMessageContext.setCommercialOwner(owner);
213: }
214: return null;
215: } else {
216: // hook：设置发送失败到context
217: if (hasSendMessageHook()) {
218: int wroteSize = request.getBody().length;
219: int incValue = (int) Math.ceil(wroteSize / BrokerStatsManager.SIZE_PER_COUNT);
220: 
221: sendMessageContext.setCommercialSendStats(BrokerStatsManager.StatsType.SEND_FAILURE);
222: sendMessageContext.setCommercialSendTimes(incValue);
223: sendMessageContext.setCommercialSendSize(wroteSize);
224: sendMessageContext.setCommercialOwner(owner);
225: }
226: }
227: } else {
228: response.setCode(ResponseCode.SYSTEM_ERROR);
229: response.setRemark("store putMessage return null");
230: }
231: 
232: return response;
233: }

 1: protected RemotingCommand msgCheck(final ChannelHandlerContext ctx,
2: final SendMessageRequestHeader requestHeader, final RemotingCommand response){
 3: // 检查 broker 是否有写入权限
 4: if (!PermName.isWriteable(this.brokerController.getBrokerConfig().getBrokerPermission())
 5: && this.brokerController.getTopicConfigManager().isOrderTopic(requestHeader.getTopic())) {
 6: response.setCode(ResponseCode.NO_PERMISSION);
 7: response.setRemark("the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1()
 8: + "] sending message is forbidden");
 9: return response;
10: }
11: // 检查topic是否可以被发送。目前是{@link MixAll.DEFAULT_TOPIC}不被允许发送
12: if (!this.brokerController.getTopicConfigManager().isTopicCanSendMessage(requestHeader.getTopic())) {
13: String errorMsg = "the topic[" + requestHeader.getTopic() + "] is conflict with system reserved words.";
14: log.warn(errorMsg);
15: response.setCode(ResponseCode.SYSTEM_ERROR);
16: response.setRemark(errorMsg);
17: return response;
18: }
19: TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());
20: if (null == topicConfig) { // 不能存在topicConfig，则进行创建
21: int topicSysFlag = 0;
22: if (requestHeader.isUnitMode()) {
23: if (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
24: topicSysFlag = TopicSysFlag.buildSysFlag(false, true);
25: } else {
26: topicSysFlag = TopicSysFlag.buildSysFlag(true, false);
27: }
28: }
29: // 创建topic配置
30: log.warn("the topic {} not exist, producer: {}", requestHeader.getTopic(), ctx.channel().remoteAddress());
31: topicConfig = this.brokerController.getTopicConfigManager().createTopicInSendMessageMethod(//
32: requestHeader.getTopic(), //
33: requestHeader.getDefaultTopic(), //
34: RemotingHelper.parseChannelRemoteAddr(ctx.channel()), //
35: requestHeader.getDefaultTopicQueueNums(), topicSysFlag);
36: if (null == topicConfig) {
37: if (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
38: topicConfig =
39: this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(
40: requestHeader.getTopic(), 1, PermName.PERM_WRITE | PermName.PERM_READ,
41: topicSysFlag);
42: }
43: }
44: // 如果没配置
45: if (null == topicConfig) {
46: response.setCode(ResponseCode.TOPIC_NOT_EXIST);
47: response.setRemark("topic[" + requestHeader.getTopic() + "] not exist, apply first please!"
48: + FAQUrl.suggestTodo(FAQUrl.APPLY_TOPIC_URL));
49: return response;
50: }
51: }
52: // 队列编号是否正确
53: int queueIdInt = requestHeader.getQueueId();
54: int idValid = Math.max(topicConfig.getWriteQueueNums(), topicConfig.getReadQueueNums());
55: if (queueIdInt >= idValid) {
56: String errorInfo = String.format("request queueId[%d] is illegal, %s Producer: %s",
57: queueIdInt,
58: topicConfig.toString(),
59: RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
60: log.warn(errorInfo);
61: response.setCode(ResponseCode.SYSTEM_ERROR);
62: response.setRemark(errorInfo);
63: return response;
64: }
65: return response;
66: }

 1: public PutMessageResult putMessage(MessageExtBrokerInner msg){
 2: if (this.shutdown) {
 3: log.warn("message store has shutdown, so putMessage is forbidden");
 4: return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
 5: }
 6: 
 7: // 从节点不允许写入
 8: if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) {
 9: long value = this.printTimes.getAndIncrement();
10: if ((value % 50000) == 0) {
11: log.warn("message store is slave mode, so putMessage is forbidden ");
12: }
13: 
14: return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
15: }
16: 
17: // store是否允许写入
18: if (!this.runningFlags.isWriteable()) {
19: long value = this.printTimes.getAndIncrement();
20: if ((value % 50000) == 0) {
21: log.warn("message store is not writeable, so putMessage is forbidden " + this.runningFlags.getFlagBits());
22: }
23: 
24: return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
25: } else {
26: this.printTimes.set(0);
27: }
28: 
29: // 消息过长
30: if (msg.getTopic().length() > Byte.MAX_VALUE) {
31: log.warn("putMessage message topic length too long " + msg.getTopic().length());
32: return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null);
33: }
34: 
35: // 消息附加属性过长
36: if (msg.getPropertiesString() != null && msg.getPropertiesString().length() > Short.MAX_VALUE) {
37: log.warn("putMessage message properties length too long " + msg.getPropertiesString().length());
38: return new PutMessageResult(PutMessageStatus.PROPERTIES_SIZE_EXCEEDED, null);
39: }
40: 
41: if (this.isOSPageCacheBusy()) {
42: return new PutMessageResult(PutMessageStatus.OS_PAGECACHE_BUSY, null);
43: }
44: 
45: long beginTime = this.getSystemClock().now();
46: // 添加消息到commitLog
47: PutMessageResult result = this.commitLog.putMessage(msg);
48: 
49: long eclipseTime = this.getSystemClock().now() - beginTime;
50: if (eclipseTime > 500) {
51: log.warn("putMessage not in lock eclipse time(ms)={}, bodyLength={}", eclipseTime, msg.getBody().length);
52: }
53: this.storeStatsService.setPutMessageEntireTimeMax(eclipseTime);
54: 
55: if (null == result || !result.isOk()) {
56: this.storeStatsService.getPutMessageFailedTimes().incrementAndGet();
57: }
58: 
59: return result;
60: }