org.apache.rocketmq.common.message.Message.getTopic()方法的使用及代码示例

x33g5p2x  于2022-01-25 转载在 其他  
字(11.3k)|赞(0)|评价(0)|浏览(272)

本文整理了Java中org.apache.rocketmq.common.message.Message.getTopic()方法的一些代码示例，展示了Message.getTopic()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。Message.getTopic()方法的具体详情如下：
包路径：org.apache.rocketmq.common.message.Message
类名称：Message
方法名：getTopic

Message.getTopic介绍

暂无

代码示例

代码示例来源：origin: apache/rocketmq

  1. for (; nextIndex < messages.size(); nextIndex++) {
  2.   Message message = messages.get(nextIndex);
  3.   int tmpSize = message.getTopic().length() + message.getBody().length;
  4.   Map<String, String> properties = message.getProperties();
  5.   for (Map.Entry<String, String> entry : properties.entrySet()) {

代码示例来源：origin: apache/rocketmq

  1. public SendResult send(Message msg, MessageQueue mq, long timeout)
  2.   throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
  3.   long beginStartTime = System.currentTimeMillis();
  4.   this.makeSureStateOK();
  5.   Validators.checkMessage(msg, this.defaultMQProducer);
  6.   if (!msg.getTopic().equals(mq.getTopic())) {
  7.     throw new MQClientException("message's topic not equal mq's topic", null);
  8.   }
  9.   long costTime = System.currentTimeMillis() - beginStartTime;
  10.   if (timeout < costTime) {
  11.     throw new RemotingTooMuchRequestException("call timeout");
  12.   }
  13.   return this.sendKernelImpl(msg, mq, CommunicationMode.SYNC, null, null, timeout);
  14. }

代码示例来源：origin: apache/rocketmq

  1. @Override
  2. public void run() {
  3.   try {
  4.     makeSureStateOK();
  5.     Validators.checkMessage(msg, defaultMQProducer);
  6.     if (!msg.getTopic().equals(mq.getTopic())) {
  7.       throw new MQClientException("message's topic not equal mq's topic", null);
  8.     }
  9.     long costTime = System.currentTimeMillis() - beginStartTime;
  10.     if (timeout > costTime) {
  11.       try {
  12.         sendKernelImpl(msg, mq, CommunicationMode.ASYNC, sendCallback, null,
  13.           timeout - costTime);
  14.       } catch (MQBrokerException e) {
  15.         throw new MQClientException("unknown exception", e);
  16.       }
  17.     } else {
  18.       sendCallback.onException(new RemotingTooMuchRequestException("call timeout"));
  19.     }
  20.   } catch (Exception e) {
  21.     sendCallback.onException(e);
  22.   }
  23. }

代码示例来源：origin: apache/rocketmq

  1. public static Message cloneMessage(final Message msg) {
  2.   Message newMsg = new Message(msg.getTopic(), msg.getBody());
  3.   newMsg.setFlag(msg.getFlag());
  4.   newMsg.setProperties(msg.getProperties());
  5.   return newMsg;
  6. }

代码示例来源：origin: apache/rocketmq

  1. private SendResult sendSelectImpl(
  2.   Message msg,
  3.   MessageQueueSelector selector,
  4.   Object arg,
  5.   final CommunicationMode communicationMode,
  6.   final SendCallback sendCallback, final long timeout
  7. ) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
  8.   long beginStartTime = System.currentTimeMillis();
  9.   this.makeSureStateOK();
  10.   Validators.checkMessage(msg, this.defaultMQProducer);
  11.   TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
  12.   if (topicPublishInfo != null && topicPublishInfo.ok()) {
  13.     MessageQueue mq = null;
  14.     try {
  15.       mq = selector.select(topicPublishInfo.getMessageQueueList(), msg, arg);
  16.     } catch (Throwable e) {
  17.       throw new MQClientException("select message queue throwed exception.", e);
  18.     }
  19.     long costTime = System.currentTimeMillis() - beginStartTime;
  20.     if (timeout < costTime) {
  21.       throw new RemotingTooMuchRequestException("sendSelectImpl call timeout");
  22.     }
  23.     if (mq != null) {
  24.       return this.sendKernelImpl(msg, mq, communicationMode, sendCallback, null, timeout - costTime);
  25.     } else {
  26.       throw new MQClientException("select message queue return null.", null);
  27.     }
  28.   }
  29.   throw new MQClientException("No route info for this topic, " + msg.getTopic(), null);
  30. }

代码示例来源：origin: apache/rocketmq

  1. (SendMessageResponseHeader) response.decodeCommandCustomHeader(SendMessageResponseHeader.class);
  2. MessageQueue messageQueue = new MessageQueue(msg.getTopic(), brokerName, responseHeader.getQueueId());

代码示例来源：origin: apache/rocketmq

  1. private SendResult send(final Message message, long timeout) {
  2.   checkMessageType(message);
  3.   org.apache.rocketmq.common.message.Message rmqMessage = msgConvert((BytesMessage) message);
  4.   try {
  5.     org.apache.rocketmq.client.producer.SendResult rmqResult = this.rocketmqProducer.send(rmqMessage, timeout);
  6.     if (!rmqResult.getSendStatus().equals(SendStatus.SEND_OK)) {
  7.       log.error(String.format("Send message to RocketMQ failed, %s", message));
  8.       throw new OMSRuntimeException("-1", "Send message to RocketMQ broker failed.");
  9.     }
  10.     message.sysHeaders().put(Message.BuiltinKeys.MESSAGE_ID, rmqResult.getMsgId());
  11.     return OMSUtil.sendResultConvert(rmqResult);
  12.   } catch (Exception e) {
  13.     log.error(String.format("Send message to RocketMQ failed, %s", message), e);
  14.     throw checkProducerException(rmqMessage.getTopic(), message.sysHeaders().getString(Message.BuiltinKeys.MESSAGE_ID), e);
  15.   }
  16. }

代码示例来源：origin: apache/rocketmq

  1. @Override
  2. public void sendMessageBefore(SendMessageContext context) {
  3.   //if it is message trace data,then it doesn't recorded
  4.   if (context == null || context.getMessage().getTopic().startsWith(((AsyncTraceDispatcher) localDispatcher).getTraceTopicName())) {
  5.     return;
  6.   }
  7.   //build the context content of TuxeTraceContext
  8.   TraceContext tuxeContext = new TraceContext();
  9.   tuxeContext.setTraceBeans(new ArrayList<TraceBean>(1));
  10.   context.setMqTraceContext(tuxeContext);
  11.   tuxeContext.setTraceType(TraceType.Pub);
  12.   tuxeContext.setGroupName(context.getProducerGroup());
  13.   //build the data bean object of message trace
  14.   TraceBean traceBean = new TraceBean();
  15.   traceBean.setTopic(context.getMessage().getTopic());
  16.   traceBean.setTags(context.getMessage().getTags());
  17.   traceBean.setKeys(context.getMessage().getKeys());
  18.   traceBean.setStoreHost(context.getBrokerAddr());
  19.   traceBean.setBodyLength(context.getMessage().getBody().length);
  20.   traceBean.setMsgType(context.getMsgType());
  21.   tuxeContext.getTraceBeans().add(traceBean);
  22. }

代码示例来源：origin: apache/rocketmq

  1. long beginTimestampPrev = beginTimestampFirst;
  2. long endTimestamp = beginTimestampFirst;
  3. TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
  4. if (topicPublishInfo != null && topicPublishInfo.ok()) {
  5.   boolean callTimeout = false;
  6.     times,
  7.     System.currentTimeMillis() - beginTimestampFirst,
  8.     msg.getTopic(),
  9.     Arrays.toString(brokersSent));
  10. throw new MQClientException("No route info of this topic, " + msg.getTopic() + FAQUrl.suggestTodo(FAQUrl.NO_TOPIC_ROUTE_INFO),
  11.   null).setResponseCode(ClientErrorCode.NOT_FOUND_TOPIC_EXCEPTION);

代码示例来源：origin: apache/rocketmq

  1. /**
  2.  * Validate message
  3.  */
  4. public static void checkMessage(Message msg, DefaultMQProducer defaultMQProducer)
  5.   throws MQClientException {
  6.   if (null == msg) {
  7.     throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message is null");
  8.   }
  9.   // topic
  10.   Validators.checkTopic(msg.getTopic());
  11.   // body
  12.   if (null == msg.getBody()) {
  13.     throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message body is null");
  14.   }
  15.   if (0 == msg.getBody().length) {
  16.     throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message body length is zero");
  17.   }
  18.   if (msg.getBody().length > defaultMQProducer.getMaxMessageSize()) {
  19.     throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL,
  20.       "the message body size over max value, MAX: " + defaultMQProducer.getMaxMessageSize());
  21.   }
  22. }

代码示例来源：origin: apache/rocketmq

  1. @Override
  2. public void sendMessageAfter(SendMessageContext context) {
  3.   if (context == null || context.getMessage().getTopic().startsWith(((AsyncTraceDispatcher) localDispatcher).getTraceTopicName())
  4.     || context.getMqTraceContext() == null) {
  5.     return;

代码示例来源：origin: apache/rocketmq

  1. private MessageExtBrokerInner makeOpMessageInner(Message message, MessageQueue messageQueue) {
  2.   MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
  3.   msgInner.setTopic(message.getTopic());
  4.   msgInner.setBody(message.getBody());
  5.   msgInner.setQueueId(messageQueue.getQueueId());
  6.   msgInner.setTags(message.getTags());
  7.   msgInner.setTagsCode(MessageExtBrokerInner.tagsString2tagsCode(msgInner.getTags()));
  8.   msgInner.setSysFlag(0);
  9.   MessageAccessor.setProperties(msgInner, message.getProperties());
  10.   msgInner.setPropertiesString(MessageDecoder.messageProperties2String(message.getProperties()));
  11.   msgInner.setBornTimestamp(System.currentTimeMillis());
  12.   msgInner.setBornHost(this.storeHost);
  13.   msgInner.setStoreHost(this.storeHost);
  14.   msgInner.setWaitStoreMsgOK(false);
  15.   MessageClientIDSetter.setUniqID(msgInner);
  16.   return msgInner;
  17. }

代码示例来源：origin: apache/rocketmq

  1. System.out.println(messageByMsgId);
  2. Assert.assertEquals(message.getTopic(), messageByMsgId.getTopic());
  3. Assert.assertEquals(message.getTopic(), messageByOffset.getTopic());

代码示例来源：origin: apache/rocketmq

  1. log.info("async send msg by retry {} times. topic={}, brokerAddr={}, brokerName={}", tmp, msg.getTopic(), addr,
  2.   retryBrokerName);
  3. try {

代码示例来源：origin: apache/rocketmq

  1. requestHeader.setTopic(msg.getTopic());
  2. requestHeader.setDefaultTopic(this.defaultMQProducer.getCreateTopicKey());
  3. requestHeader.setDefaultTopicQueueNums(this.defaultMQProducer.getDefaultTopicQueueNums());

代码示例来源：origin: didi/DDMQ

  1. public SendResult send(Message msg, MessageQueue mq, long timeout)
  2.   throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
  3.   this.makeSureStateOK();
  4.   Validators.checkMessage(msg, this.defaultMQProducer);
  5.   if (!msg.getTopic().equals(mq.getTopic())) {
  6.     throw new MQClientException("message's topic not equal mq's topic", null);
  7.   }
  8.   return this.sendKernelImpl(msg, mq, CommunicationMode.SYNC, null, null, timeout);
  9. }

代码示例来源：origin: didi/DDMQ

  1. public void send(Message msg, MessageQueue mq, SendCallback sendCallback, long timeout)
  2.   throws MQClientException, RemotingException, InterruptedException {
  3.   this.makeSureStateOK();
  4.   Validators.checkMessage(msg, this.defaultMQProducer);
  5.   if (!msg.getTopic().equals(mq.getTopic())) {
  6.     throw new MQClientException("message's topic not equal mq's topic", null);
  7.   }
  8.   try {
  9.     this.sendKernelImpl(msg, mq, CommunicationMode.ASYNC, sendCallback, null, timeout);
  10.   } catch (MQBrokerException e) {
  11.     throw new MQClientException("unknown exception", e);
  12.   }
  13. }

代码示例来源：origin: didi/DDMQ

  1. @Override
  2. public void send(final Message message) {
  3.   checkMessageType(message);
  4.   org.apache.rocketmq.common.message.Message rmqMessage = OMSUtil.msgConvert((BytesMessage) message);
  5.   try {
  6.     Validators.checkMessage(rmqMessage, this.rocketmqProducer);
  7.   } catch (MQClientException e) {
  8.     throw checkProducerException(rmqMessage.getTopic(), message.headers().getString(MessageHeader.MESSAGE_ID), e);
  9.   }
  10.   msgCacheQueue.add(message);
  11. }

代码示例来源：origin: didi/DDMQ

  1. private SendResult send(final Message message, long timeout) {
  2.   checkMessageType(message);
  3.   org.apache.rocketmq.common.message.Message rmqMessage = msgConvert((BytesMessage) message);
  4.   try {
  5.     org.apache.rocketmq.client.producer.SendResult rmqResult = this.rocketmqProducer.send(rmqMessage, timeout);
  6.     if (!rmqResult.getSendStatus().equals(SendStatus.SEND_OK)) {
  7.       log.error(String.format("Send message to RocketMQ failed, %s", message));
  8.       throw new OMSRuntimeException("-1", "Send message to RocketMQ broker failed.");
  9.     }
  10.     message.headers().put(MessageHeader.MESSAGE_ID, rmqResult.getMsgId());
  11.     return OMSUtil.sendResultConvert(rmqResult);
  12.   } catch (Exception e) {
  13.     log.error(String.format("Send message to RocketMQ failed, %s", message), e);
  14.     throw checkProducerException(rmqMessage.getTopic(), message.headers().getString(MessageHeader.MESSAGE_ID), e);
  15.   }
  16. }

代码示例来源：origin: didi/DDMQ

  1. private SendMessageRequestHeader getMessageRequestHeader(Message msg, MessageQueue mq) {
  2.   SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
  3.   requestHeader.setProducerGroup(producer.getProducerGroup());
  4.   requestHeader.setTopic(msg.getTopic());
  5.   requestHeader.setDefaultTopic(producer.getCreateTopicKey());
  6.   requestHeader.setDefaultTopicQueueNums(producer.getDefaultTopicQueueNums());
  7.   requestHeader.setQueueId(mq.getQueueId());
  8.   requestHeader.setSysFlag(0);
  9.   requestHeader.setBornTimestamp(System.currentTimeMillis());
  10.   requestHeader.setFlag(msg.getFlag());
  11.   requestHeader.setProperties(MessageDecoder.messageProperties2String(msg.getProperties()));
  12.   requestHeader.setReconsumeTimes(0);
  13.   requestHeader.setUnitMode(producer.isUnitMode());
  14.   return requestHeader;
  15. }

内容来源于网络，如有侵权，请联系作者删除！

Message

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com