com.alibaba.rocketmq.common.message.Message类的使用及代码示例

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

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

Message介绍

暂无

代码示例

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

private static Message buildMessage(final int messageSize) throws UnsupportedEncodingException {
    Message msg = new Message();
    msg.setTopic("BenchmarkTest");

    StringBuilder sb = new StringBuilder();
    for (int i = 0; i < messageSize; i += 10) {
      sb.append("hello baby");
    }

    msg.setBody(sb.toString().getBytes(RemotingHelper.DEFAULT_CHARSET));

    return msg;
  }
}

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

Message message = new Message();
message.setTopic(Constants.TARZAN_TEST_TOPIC);
message.setTags(TestConstants.MESSAGE_TAG);
message.setKeys(messageKey);
message.setBody(messageKey.getBytes());
  LOGGER.error("准备消息 '{}' 失败, {}", message.getKeys(), result.getErrorMsg());
  return;
LOGGER.info("准备消息 '" + message.getKeys() + "' 成功, 事务Id=" + tid);
  LOGGER.info("本地事务处理成功，提交消息'" + message.getKeys() + "'.");
  testMessageNotifier.commitMessage(tid, message);
}else {
  LOGGER.info("本地事务处理失败，回滚消息'" + message.getKeys() + "'.");
  testMessageNotifier.rollbackMessage(tid);

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

for (int i = 0; i < mMessageList.size(); i++){
  Message mMessageModel = mMessageList.get(i);
  String My_Topic = mMessageModel.getTopic();
  // other stuff

}

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

public Message(String topic, String tags, String keys, int flag, byte[] body, boolean waitStoreMsgOK) {
  this.topic = topic;
  this.flag = flag;
  this.body = body;
  if (tags != null && tags.length() > 0)
    this.setTags(tags);
  if (keys != null && keys.length() > 0)
    this.setKeys(keys);
  this.setWaitStoreMsgOK(waitStoreMsgOK);
}

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

private void checkMessage(Message message) throws MQClientException {
  if(message.getTopic() == null){
    message.setTopic(getTopic());
  }
  RocketMQValidators.checkMessage(message);
}

代码示例来源：origin: majinkai/skye

public void send(List<Span> spans) {
  try {
    for (Span span : spans) {
      LOG.debug("Ready for send {}", span);
      Message message = new Message(Header.MQ_TOPIC, Header.MQ_TAG, Serializer.ser(span));
      SendResult sendResult = producer.send(message);
      LOG.debug("SendResult={}", sendResult);
    }
  } catch (Exception e) {
    LOG.error("Send message exception", e);
  }
}

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

public boolean sendMessageBack(final MessageExt msg) {
  try {
    Message newMsg = new Message(MixAll.getRetryTopic(this.defaultMQPushConsumer.getConsumerGroup()), msg.getBody());
    String originMsgId = MessageAccessor.getOriginMessageId(msg);
    MessageAccessor.setOriginMessageId(newMsg, UtilAll.isBlank(originMsgId) ? msg.getMsgId() : originMsgId);
    newMsg.setFlag(msg.getFlag());
    MessageAccessor.setProperties(newMsg, msg.getProperties());
    MessageAccessor.putProperty(newMsg, MessageConst.PROPERTY_RETRY_TOPIC, msg.getTopic());
    MessageAccessor.setReconsumeTime(newMsg, String.valueOf(msg.getReconsumeTimes()));
    MessageAccessor.setMaxReconsumeTimes(newMsg, String.valueOf(this.defaultMQPushConsumer.getMaxReconsumeTimes()));
    newMsg.setDelayTimeLevel(3 + msg.getReconsumeTimes());
    this.defaultMQPushConsumer.getDefaultMQPushConsumerImpl().getmQClientFactory().getDefaultMQProducer().send(newMsg);
    return true;
  } catch (Exception e) {
    log.error("sendMessageBack exception, group: " + this.consumerGroup + " msg: " + msg.toString(), e);
  }
  return false;
}

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

@Transactional
@Override
public void check() {
  List<Long> all = mapper.findMessageIdsByStatusCreatedAfter(Arrays.asList(MessageStatus.UNCONSUMED, MessageStatus.CONSUME_FAILED), MQProducerConfig.CHECK_GAP);
  Message checkMessage = new Message();
  checkMessage.setTopic(config.getTopic());
  checkMessage.setTags(config.getCheckKeys());
  checkMessage.setBody(ProtoStuffUtil.serialize(all));
  try {
    producer.send(checkMessage);
  } catch (Exception e) {
    log.info("发送check消息失败，暂不做处理，不会影响数据一致性");
    e.printStackTrace();
  }
}

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

public static void main(String[] args) throws MQClientException, InterruptedException {
    DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
    producer.start();

    try {
      for (int i = 0; i < 6000000; i++) {
        Message msg = new Message("TopicFilter7",// topic
            "TagA",// tag
            "OrderID001",// key
            ("Hello MetaQ").getBytes(RemotingHelper.DEFAULT_CHARSET));// body

        msg.putUserProperty("SequenceId", String.valueOf(i));

        SendResult sendResult = producer.send(msg);
        System.out.println(sendResult);
      }
    } catch (Exception e) {
      e.printStackTrace();
    }

    producer.shutdown();
  }
}

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

/**
 * 发送数据的接口
 *
 * @param keySet 本批次包含的keyset
 * @param data   本批次的轨迹数据
 */
public void sendTraceDataByMQ(Set<String> keySet, String data) {
  String topic = OnsTraceConstants.traceTopic + currentRegionId;
  final Message message = new Message(topic, data.getBytes());
  message.setKeys(keySet);
  try {
    traceProducer.send(message, new SendCallback() {
      @Override
      public void onSuccess(SendResult sendResult) {
      }
      @Override
      public void onException(Throwable e) {
        //todo 对于发送失败的数据，如何保存，保证所有轨迹数据都记录下来
        clientlog.info("send trace data failed ,the msgidSet is" + message.getKeys());
      }
    }, 5000);
  } catch (Exception e) {
    clientlog.info("send trace data failed ,the msgidSet is" + message.getKeys());
  }
}

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

/**
   * 发送数据的接口
   *
   * @param keySet 本批次包含的keyset
   * @param data 本批次的轨迹数据
   */
  private void sendTraceDataByMQ(Set<String> keySet, final String data, String currentRegionId) {
    String topic = OnsTraceConstants.traceTopic + currentRegionId;
    final Message message = new Message(topic, data.getBytes());
    message.setKeys(keySet);
    try {
      traceProducer.send(message, new SendCallback() {
        @Override
        public void onSuccess(SendResult sendResult) {
        }
        @Override
        public void onException(Throwable e) {
          //todo 对于发送失败的数据，如何保存，保证所有轨迹数据都记录下来
          clientlog.info("send trace data ,the traceData is " + data);
        }
      }, 5000);
    } catch (Exception e) {
      clientlog.info("send trace data,the traceData is" + data);
    }
  }
}

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

Message test = new Message();
MessageClientIDSetter.setUniqID(test);
System.out.println(test.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX));

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

byte[] prevBody = msg.getBody();
try {
  final String tranMsg = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
  if (tranMsg != null && Boolean.parseBoolean(tranMsg)) {
    sysFlag |= MessageSysFlag.TransactionPreparedType;
    context.setMessage(msg);
    context.setMq(mq);
    String isTrans = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
    if (isTrans != null && isTrans.equals("true")) {
      context.setMsgType(MessageType.Trans_Msg_Half);
    if (msg.getProperty("__STARTDELIVERTIME") != null || msg.getProperty(MessageConst.PROPERTY_DELAY_TIME_LEVEL) != null) {
      context.setMsgType(MessageType.Delay_Msg);
  requestHeader.setTopic(msg.getTopic());
  requestHeader.setDefaultTopic(this.defaultMQProducer.getCreateTopicKey());
  requestHeader.setDefaultTopicQueueNums(this.defaultMQProducer.getDefaultTopicQueueNums());
  requestHeader.setSysFlag(sysFlag);
  requestHeader.setBornTimestamp(System.currentTimeMillis());
  requestHeader.setFlag(msg.getFlag());
  requestHeader.setProperties(MessageDecoder.messageProperties2String(msg.getProperties()));
  requestHeader.setReconsumeTimes(0);
  requestHeader.setUnitMode(this.isUnitMode());
  msg.setBody(prevBody);

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

public static void checkMessage(Message msg, DefaultMQProducer defaultMQProducer)
      throws MQClientException {
    if (null == msg) {
      throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message is null");
    }
    // topic
    Validators.checkTopic(msg.getTopic());
    // body
    if (null == msg.getBody()) {
      throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message body is null");
    }

    if (0 == msg.getBody().length) {
      throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message body length is zero");
    }

    if (defaultMQProducer != null && msg.getBody().length > defaultMQProducer.getMaxMessageSize()) {
      throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL,
          "the message body size over max value, MAX: " + defaultMQProducer.getMaxMessageSize());
    }
  }
}

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

private boolean tryToCompressMessage(final Message msg) {
  byte[] body = msg.getBody();
  if (body != null) {
    if (body.length >= this.defaultMQProducer.getCompressMsgBodyOverHowmuch()) {
      try {
        byte[] data = UtilAll.compress(body, zipCompressLevel);
        if (data != null) {
          msg.setBody(data);
          return true;
        }
      } catch (IOException e) {
        log.error("tryToCompressMessage exception", e);
        log.warn(msg.toString());
      }
    }
  }
  return false;
}

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

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

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

public void setKeys(Collection<String> keys) {
  StringBuffer sb = new StringBuffer();
  for (String k : keys) {
    sb.append(k);
    sb.append(MessageConst.KEY_SEPARATOR);
  }
  this.setKeys(sb.toString().trim());
}

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

private RocketMQBody buildMQBody(Message message){
  RocketMQBody mqBody = new RocketMQBody();
  mqBody.setProducerGroup(getGroupId());
  mqBody.setTopic(getTopic());
  mqBody.setTags(message.getTags());
  mqBody.setMessageKey(message.getKeys());
  mqBody.setMessageBody(message.getBody());
  return mqBody;
}

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

long beginTimestampPrev = beginTimestampFirst;
long endTimestamp = beginTimestampFirst;
TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
if (topicPublishInfo != null && topicPublishInfo.ok()) {
  MessageQueue mq = null;
        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;
        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;
        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()) {
        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());
        exception = e;

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

try {
  if (sendResult.getTransactionId() != null) {
    msg.putUserProperty("__transactionId__", sendResult.getTransactionId());
    log.info(msg.toString());
  log.info(msg.toString());
  localException = e;

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

Message

相关文章

    查看更多
    微信公众号

    热门标签

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

    最新文章

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