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

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

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

Message.getTopic介绍

暂无

代码示例

代码示例来源：origin: stackoverflow.com

  1. for (int i = 0; i < mMessageList.size(); i++){
  2.   Message mMessageModel = mMessageList.get(i);
  3.   String My_Topic = mMessageModel.getTopic();
  4.   // other stuff
  6. }

代码示例来源：origin: beston123/Tarzan

  1. private void checkMessage(Message message) throws MQClientException {
  2.   if(message.getTopic() == null){
  3.     message.setTopic(getTopic());
  4.   }
  5.   RocketMQValidators.checkMessage(message);
  6. }

代码示例来源：origin: com.alibaba.rocketmq/rocketmq-client

  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: kuangye098/rocketmq

  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: coffeewar/enode-master

  1. String.format("Connect broker failed, Topic: %s", message.getTopic()),
  2.       FAQ.CONNECT_BROKER_FAILED));
  3.   throw new RocketMQClientException(FAQ.errorMessage(String.format(
  4.       "Send message to broker timeout, %dms, Topic: %s",
  5.       this.defaultMQProducer.getSendMsgTimeout(), message.getTopic()),
  6.       FAQ.SEND_MSG_TO_BROKER_TIMEOUT));
  7.   MQBrokerException excep = (MQBrokerException) e.getCause();
  8.   throw new RocketMQClientException(FAQ.errorMessage(
  9.       String.format("Receive a broker exception, Topic: %s, %s", message.getTopic(),
  10.           excep.getErrorMessage()), FAQ.BROKER_RESPONSE_EXCEPTION));
  11. if (-1 == excep.getResponseCode()) {
  12.   throw new RocketMQClientException(FAQ.errorMessage(
  13.       String.format("Topic does not exist, Topic: %s, %s", message.getTopic(),
  14.           excep.getErrorMessage()), FAQ.TOPIC_ROUTE_NOT_EXIST));
  15. } else if (ResponseCode.MESSAGE_ILLEGAL == excep.getResponseCode()) {
  16.   throw new RocketMQClientException(FAQ.errorMessage(String.format(
  17.       "ONS Client check message exception, Topic: %s, %s", message.getTopic(),
  18.       excep.getErrorMessage()), FAQ.CLIENT_CHECK_MSG_EXCEPTION));

代码示例来源：origin: com.alibaba.rocketmq/rocketmq-client

  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("unknow exception", e);
  12.   }
  13. }

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

  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("unknow exception", e);
  12.   }
  13. }

代码示例来源：origin: com.alibaba.rocketmq/rocketmq-client

  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.   this.makeSureStateOK();
  9.   Validators.checkMessage(msg, this.defaultMQProducer);
  10.   TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
  11.   if (topicPublishInfo != null && topicPublishInfo.ok()) {
  12.     MessageQueue mq = null;
  13.     try {
  14.       mq = selector.select(topicPublishInfo.getMessageQueueList(), msg, arg);
  15.     } catch (Throwable e) {
  16.       throw new MQClientException("select message queue throwed exception.", e);
  17.     }
  18.     if (mq != null) {
  19.       return this.sendKernelImpl(msg, mq, communicationMode, sendCallback, null, timeout);
  20.     } else {
  21.       throw new MQClientException("select message queue return null.", null);
  22.     }
  23.   }
  24.   throw new MQClientException("No route info for this topic, " + msg.getTopic(), null);
  25. }

代码示例来源：origin: kuangye098/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.   this.makeSureStateOK();
  9.   Validators.checkMessage(msg, this.defaultMQProducer);
  10.   TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
  11.   if (topicPublishInfo != null && topicPublishInfo.ok()) {
  12.     MessageQueue mq = null;
  13.     try {
  14.       mq = selector.select(topicPublishInfo.getMessageQueueList(), msg, arg);
  15.     } catch (Throwable e) {
  16.       throw new MQClientException("select message queue throwed exception.", e);
  17.     }
  18.     if (mq != null) {
  19.       return this.sendKernelImpl(msg, mq, communicationMode, sendCallback, null, timeout);
  20.     } else {
  21.       throw new MQClientException("select message queue return null.", null);
  22.     }
  23.   }
  24.   throw new MQClientException("No route info for this topic, " + msg.getTopic(), null);
  25. }

代码示例来源：origin: com.alibaba.rocketmq/rocketmq-client

  1. long beginTimestampPrev = beginTimestampFirst;
  2. long endTimestamp = beginTimestampFirst;
  3. TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
  4. if (topicPublishInfo != null && topicPublishInfo.ok()) {
  5.   MessageQueue mq = null;
  6.       msg.getTopic(), //
  7.       Arrays.toString(brokersSent));
  8. throw new MQClientException("No route info of this topic, " + msg.getTopic() + FAQUrl.suggestTodo(FAQUrl.NO_TOPIC_ROUTE_INFO),
  9.     null).setResponseCode(ClientErrorCode.NotFoundTopicException);

代码示例来源：origin: coffeewar/enode-master

  1. @Override
  2. public void sendMessageBefore(SendMessageContext context) {
  3.   // 如果是消息轨迹本身的发送链路，则不需要再记录
  4.   if (context == null || context.getMessage().getTopic().startsWith(MixAll.SYSTEM_TOPIC_PREFIX)) {
  5.     return;
  6.   }
  7.   OnsTraceContext onsContext = new OnsTraceContext();
  8.   onsContext.setTraceBeans(new ArrayList<OnsTraceBean>(1));
  9.   context.setMqTraceContext(onsContext);
  10.   onsContext.setTraceType(OnsTraceType.Pub);
  11.   onsContext.setGroupName(context.getProducerGroup());
  12.   OnsTraceBean traceBean = new OnsTraceBean();
  13.   traceBean.setTopic(context.getMessage().getTopic());
  14.   traceBean.setTags(context.getMessage().getTags());
  15.   traceBean.setKeys(context.getMessage().getKeys());
  16.   traceBean.setStoreHost(context.getBrokerAddr());
  17.   traceBean.setBodyLength(context.getMessage().getBody().length);
  18.   traceBean.setMsgType(context.getMsgType());
  19.   onsContext.getTraceBeans().add(traceBean);
  20. }

代码示例来源：origin: songxinjianqwe/EShop-SOA

  1. .updateTime(LocalDateTime.now())
  2. .messageStatus(MessageStatus.UNCONSUMED)
  3. .topic(msg.getTopic())
  4. .sendTimes(0)
  5. .build();

代码示例来源：origin: beston123/Tarzan

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

代码示例来源：origin: com.alibaba.rocketmq/rocketmq-client

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

代码示例来源：origin: coffeewar/enode-master

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

代码示例来源：origin: com.alibaba.rocketmq/rocketmq-client

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

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

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

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

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

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

  1. MessageQueue tmpmq = producer.selectOneMessageQueue(topicPublishInfo, brokerName);
  2. String addr = instance.findBrokerAddressInPublish(tmpmq.getBrokerName());
  3. log.info("async send msg by retry {} times. topic={}, brokerAddr={}, brokerName={}", tmp, msg.getTopic(), addr,
  4.     tmpmq.getBrokerName());
  5. try {

代码示例来源：origin: com.alibaba.rocketmq/rocketmq-client

  1. MessageQueue tmpmq = producer.selectOneMessageQueue(topicPublishInfo, brokerName);
  2. String addr = instance.findBrokerAddressInPublish(tmpmq.getBrokerName());
  3. log.info("async send msg by retry {} times. topic={}, brokerAddr={}, brokerName={}", tmp, msg.getTopic(), addr,
  4.     tmpmq.getBrokerName());
  5. try {

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

Message

相关文章

    查看更多

    热门标签

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

    最新文章

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