org.apache.rocketmq.common.message.Message类的使用及代码示例

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

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

Message介绍

暂无

代码示例

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

  1. @Test
  2. public void testBatchSend_CheckProperties() throws Exception {
  3.   List<Message> messageList = new ArrayList<>();
  4.   Message message = new Message();
  5.   message.setTopic(topic);
  6.   message.setKeys("keys123");
  7.   message.setTags("tags123");
  8.   message.setWaitStoreMsgOK(false);
  9.   message.setBuyerId("buyerid123");
  10.   message.setFlag(123);
  11.   message.setBody("body".getBytes());
  12.   messageList.add(message);
  13.   Assert.assertEquals(message.getTopic(), messageByMsgId.getTopic());
  14.   Assert.assertEquals(message.getTopic(), messageByOffset.getTopic());
  15.   Assert.assertEquals(message.getKeys(), messageByOffset.getKeys());
  16.   Assert.assertEquals(message.getKeys(), messageByMsgId.getKeys());
  17.   Assert.assertEquals(message.getTags(), messageByOffset.getTags());
  18.   Assert.assertEquals(message.getTags(), messageByMsgId.getTags());
  19.   Assert.assertEquals(message.isWaitStoreMsgOK(), messageByOffset.isWaitStoreMsgOK());
  20.   Assert.assertEquals(message.isWaitStoreMsgOK(), messageByMsgId.isWaitStoreMsgOK());
  21.   Assert.assertEquals(message.getBuyerId(), messageByOffset.getBuyerId());
  22.   Assert.assertEquals(message.getBuyerId(), messageByMsgId.getBuyerId());
  23.   Assert.assertEquals(message.getFlag(), messageByOffset.getFlag());
  24.   Assert.assertEquals(message.getFlag(), messageByMsgId.getFlag());

代码示例来源：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. public static org.apache.rocketmq.common.message.Message msgConvert(BytesMessage omsMessage) {
  2.   org.apache.rocketmq.common.message.Message rmqMessage = new org.apache.rocketmq.common.message.Message();
  3.   rmqMessage.setBody(omsMessage.getBody(byte[].class));
  4.   KeyValue sysHeaders = omsMessage.sysHeaders();
  5.   KeyValue userHeaders = omsMessage.userHeaders();
  6.   //All destinations in RocketMQ use Topic
  7.   rmqMessage.setTopic(sysHeaders.getString(BuiltinKeys.DESTINATION));
  8.   if (sysHeaders.containsKey(BuiltinKeys.START_TIME)) {
  9.     long deliverTime = sysHeaders.getLong(BuiltinKeys.START_TIME, 0);
  10.     if (deliverTime > 0) {
  11.       rmqMessage.putUserProperty(RocketMQConstants.START_DELIVER_TIME, String.valueOf(deliverTime));
  12.     }
  13.   }
  14.   for (String key : userHeaders.keySet()) {
  15.     MessageAccessor.putProperty(rmqMessage, key, userHeaders.getString(key));
  16.   }
  17.   //System headers has a high priority
  18.   for (String key : sysHeaders.keySet()) {
  19.     MessageAccessor.putProperty(rmqMessage, key, sysHeaders.getString(key));
  20.   }
  21.   return rmqMessage;
  22. }

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

  1. public Message(String topic, String tags, String keys, int flag, byte[] body, boolean waitStoreMsgOK) {
  2.   this.topic = topic;
  3.   this.flag = flag;
  4.   this.body = body;
  5.   if (tags != null && tags.length() > 0)
  6.     this.setTags(tags);
  7.   if (keys != null && keys.length() > 0)
  8.     this.setKeys(keys);
  9.   this.setWaitStoreMsgOK(waitStoreMsgOK);
  10. }

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

  1. private static Message buildMessage(final String topic, final int messageSize) throws UnsupportedEncodingException {
  2.   Message msg = new Message();
  3.   msg.setTopic(topic);
  4.   StringBuilder sb = new StringBuilder();
  5.   for (int i = 0; i < messageSize; i += 11) {
  6.     sb.append("hello jodie");
  7.   }
  8.   msg.setBody(sb.toString().getBytes(MixAll.DEFAULT_CHARSET));
  9.   return msg;
  10. }

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

  1. public static void main(String[] args) throws Exception {
  2.     DefaultMQProducer producer = new DefaultMQProducer("BatchProducerGroupName");
  3.     producer.start();
  5.     //If you just send messages of no more than 1MiB at a time, it is easy to use batch
  6.     //Messages of the same batch should have: same topic, same waitStoreMsgOK and no schedule support
  7.     String topic = "BatchTest";
  8.     List<Message> messages = new ArrayList<>();
  9.     messages.add(new Message(topic, "Tag", "OrderID001", "Hello world 0".getBytes()));
  10.     messages.add(new Message(topic, "Tag", "OrderID002", "Hello world 1".getBytes()));
  11.     messages.add(new Message(topic, "Tag", "OrderID003", "Hello world 2".getBytes()));
  13.     producer.send(messages);
  14.   }
  15. }

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

  1. /**
  2.  * Info,error,warn,callback method implementation
  3.  */
  4. @Override
  5. protected void append(ILoggingEvent event) {
  6.   if (!isStarted()) {
  7.     return;
  8.   }
  9.   String logStr = this.layout.doLayout(event);
  10.   try {
  11.     Message msg = new Message(topic, tag, logStr.getBytes());
  12.     msg.getProperties().put(ProducerInstance.APPENDER_TYPE, ProducerInstance.LOGBACK_APPENDER);
  13.     //Send message and do not wait for the ack from the message broker.
  14.     producer.sendOneway(msg);
  15.   } catch (Exception e) {
  16.     addError("Could not send message in RocketmqLogbackAppender [" + name + "]. Message is : " + logStr, e);
  17.   }
  18. }

代码示例来源：origin: spring-cloud-incubator/spring-cloud-alibaba

  1. try {
  2.   if (message.getPayload() instanceof byte[]) {
  3.     toSend = new Message(destination, (byte[]) message.getPayload());
  4.     toSend = new Message(destination,
  5.         ((String) message.getPayload()).getBytes());
  6.       message);
  7.   headerAccessor.setLeaveMutable(true);
  8.   toSend.setDelayTimeLevel(headerAccessor.getDelayTimeLevel());
  9.   toSend.setTags(headerAccessor.getTags());
  10.   toSend.setKeys(headerAccessor.getKeys());
  11.   toSend.setFlag(headerAccessor.getFlag());
  12.   for (Map.Entry<String, String> entry : headerAccessor.getUserProperties()
  13.       .entrySet()) {
  14.     toSend.putUserProperty(entry.getKey(), entry.getValue());

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

  1. public static Message convert2RMQMessage(JmsBaseMessage jmsMsg) throws Exception {
  2.   Message rocketmqMsg = new MessageExt();
  3.   rocketmqMsg.setBody(MessageConverter.getContentFromJms(jmsMsg));
  4.   rocketmqMsg.setTopic(topic);
  5.   String messageType = destination.getMessageType();
  6.   Preconditions.checkState(!messageType.contains("||"),
  7.     "'||' can not be in the destination when sending a message");
  8.   rocketmqMsg.setTags(messageType);
  9.     String value = properties.getProperty(name);
  10.     if (MessageConst.PROPERTY_KEYS.equals(name)) {
  11.       rocketmqMsg.setKeys(value);
  12.     } else if (MessageConst.PROPERTY_TAGS.equals(name)) {
  13.       rocketmqMsg.setTags(value);
  14.     } else if (MessageConst.PROPERTY_DELAY_TIME_LEVEL.equals(name)) {
  15.       rocketmqMsg.setDelayTimeLevel(Integer.parseInt(value));
  16.     } else if (MessageConst.PROPERTY_WAIT_STORE_MSG_OK.equals(name)) {
  17.       rocketmqMsg.setWaitStoreMsgOK(Boolean.parseBoolean(value));
  18.     } else if (MessageConst.PROPERTY_BUYER_ID.equals(name)) {
  19.       rocketmqMsg.setBuyerId(value);
  20.     } else {
  21.       rocketmqMsg.putUserProperty(name, value);

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

  1. @Test(expected = org.apache.rocketmq.client.exception.MQClientException.class)
  2. public void testSynSendOutOfSizeBodyMessage() throws Exception {
  3.   Message msg = new Message(topic, RandomUtils.getStringByUUID().getBytes());
  4.   msg.setBody(new byte[1024 * 1024 * 4 + 1]);
  5.   producer.send(msg);
  6. }

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

  1. public static Message decodeMessage(ByteBuffer byteBuffer) throws Exception {
  2.   Message message = new Message();
  3.   // 1 TOTALSIZE
  4.   byteBuffer.getInt();
  5.   // 2 MAGICCODE
  6.   byteBuffer.getInt();
  7.   // 3 BODYCRC
  8.   byteBuffer.getInt();
  9.   // 4 FLAG
  10.   int flag = byteBuffer.getInt();
  11.   message.setFlag(flag);
  12.   // 5 BODY
  13.   int bodyLen = byteBuffer.getInt();
  14.   byte[] body = new byte[bodyLen];
  15.   byteBuffer.get(body);
  16.   message.setBody(body);
  17.   // 6 properties
  18.   short propertiesLen = byteBuffer.getShort();
  19.   byte[] propertiesBytes = new byte[propertiesLen];
  20.   byteBuffer.get(propertiesBytes);
  21.   message.setProperties(string2messageProperties(new String(propertiesBytes, CHARSET_UTF8)));
  22.   return message;
  23. }

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

  1. byte[] prevBody = msg.getBody();
  2. try {
  3.   final String tranMsg = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
  4.   if (tranMsg != null && Boolean.parseBoolean(tranMsg)) {
  5.     sysFlag |= MessageSysFlag.TRANSACTION_PREPARED_TYPE;
  6.     context.setMessage(msg);
  7.     context.setMq(mq);
  8.     String isTrans = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
  9.     if (isTrans != null && isTrans.equals("true")) {
  10.       context.setMsgType(MessageType.Trans_Msg_Half);
  11.     if (msg.getProperty("__STARTDELIVERTIME") != null || msg.getProperty(MessageConst.PROPERTY_DELAY_TIME_LEVEL) != null) {
  12.       context.setMsgType(MessageType.Delay_Msg);
  13.   requestHeader.setTopic(msg.getTopic());
  14.   requestHeader.setDefaultTopic(this.defaultMQProducer.getCreateTopicKey());
  15.   requestHeader.setDefaultTopicQueueNums(this.defaultMQProducer.getDefaultTopicQueueNums());
  16.   requestHeader.setSysFlag(sysFlag);
  17.   requestHeader.setBornTimestamp(System.currentTimeMillis());
  18.   requestHeader.setFlag(msg.getFlag());
  19.   requestHeader.setProperties(MessageDecoder.messageProperties2String(msg.getProperties()));
  20.   requestHeader.setReconsumeTimes(0);
  21.   requestHeader.setUnitMode(this.isUnitMode());
  22.         msg.setBody(prevBody);

代码示例来源：origin: rstyro/Springboot

  1. public void sendToMQ(Object object,String tags) throws Exception {
  2.   //构造消息数据
  3.   Message message = new Message();
  4.   //主题
  5.   message.setTopic(Const.PLAN_TOPIC);
  6.   message.setTags(tags);
  7.   //key
  8.   String uuid = UUID.randomUUID().toString();
  9.   message.setKeys(uuid);
  10.   message.setBody(JSON.toJSONString(object).getBytes());
  11.   //添加参数
  12.   Map<String, Object> transactionMapArgs = new HashMap<String, Object>();
  13.   this.planProducer.sendTransactionMessage(message, this.transactionExecuterImpl, transactionMapArgs);
  14. }

代码示例来源：origin: javahongxi/whatsmars

  1. org.apache.rocketmq.common.message.Message rocketMsg = new org.apache.rocketmq.common.message.Message(topic, tags, payloads);
  2.   Object keys = headers.get(MessageConst.PROPERTY_KEYS);
  3.   if (!StringUtils.isEmpty(keys)) { // if headers has 'KEYS', set rocketMQ message key
  4.     rocketMsg.setKeys(keys.toString());
  5.   rocketMsg.setFlag(flag);
  6.   rocketMsg.setWaitStoreMsgOK(waitStoreMsgOK);
  7.       && !Objects.equals(entry.getKey(), "WAIT_STORE_MSG_OK")) // exclude "KEYS", "FLAG", "WAIT_STORE_MSG_OK"
  8.     .forEach(entry -> {
  9.       rocketMsg.putUserProperty("USERS_" + entry.getKey(), String.valueOf(entry.getValue())); // add other properties with prefix "USERS_"
  10.     });

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

  1. final Message message = new Message(topic, data.getBytes());
  2. message.setKeys(keySet);
  3. try {
  4.   Set<String> traceBrokerSet = tryGetMessageQueueBrokerSet(traceProducer.getDefaultMQProducerImpl(), topic);

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

  1. public static void main(String[] args) throws MQClientException, InterruptedException {
  2.     DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
  3.     producer.start();
  5.     try {
  6.       for (int i = 0; i < 6000000; i++) {
  7.         Message msg = new Message("TopicFilter7",
  8.           "TagA",
  9.           "OrderID001",
  10.           "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
  12.         msg.putUserProperty("SequenceId", String.valueOf(i));
  13.         SendResult sendResult = producer.send(msg);
  14.         System.out.printf("%s%n", sendResult);
  15.       }
  16.     } catch (Exception e) {
  17.       e.printStackTrace();
  18.     }
  19.     producer.shutdown();
  20.   }
  21. }

代码示例来源：origin: maihaoche/rocketmq-spring-boot-starter

  1. Message message = new Message(topic, str.getBytes(Charset.forName("utf-8")));
  2. if (!StringUtils.isEmpty(tag)) {
  3.   message.setTags(tag);
  4.   message.setKeys(messageKey);
  5.   message.setDelayTimeLevel(delayTimeLevel);

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

  1. public Object getMessageByTag(String tag) {
  2.   Object objMsg = null;
  3.   if (this instanceof RMQNormalProducer) {
  4.     org.apache.rocketmq.common.message.Message msg = new org.apache.rocketmq.common.message.Message(
  5.       topic, (RandomUtil.getStringByUUID() + "." + new Date()).getBytes());
  6.     objMsg = msg;
  7.     if (tag != null) {
  8.       msg.setTags(tag);
  9.     }
  10.   }
  11.   return objMsg;
  12. }

代码示例来源：origin: maihaoche/rocketmq-spring-boot-starter

  1. /**
  2.  * 发送数据的接口
  3.  *
  4.  * @param keySet
  5.  *            本批次包含的keyset
  6.  * @param data
  7.  *            本批次的轨迹数据
  8.  */
  9. public void sendTraceDataByMQ(Set<String> keySet, String data) {
  10.   String topic = OnsTraceConstants.traceTopic;
  11.   final Message message = new Message(topic, data.getBytes());
  12.   message.setKeys(keySet);
  13.   try {
  14.     traceProducer.send(message, new SendCallback() {
  15.       @Override
  16.       public void onSuccess(SendResult sendResult) {
  17.       }
  18.       @Override
  19.       public void onException(Throwable e) {
  20.         //todo 对于发送失败的数据，如何保存，保证所有轨迹数据都记录下来
  21.         clientlog.info("send trace data failed ,the msgidSet is"+message.getKeys());
  22.       }
  23.     }, 5000);
  24.   }
  25.   catch (Exception e) {
  26.     clientlog.info("send trace data failed ,the msgidSet is"+message.getKeys());
  27.   }
  28. }

代码示例来源：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()) {
  6.     tmpSize += entry.getKey().length() + entry.getValue().length();

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

Message

相关文章

    查看更多

    热门标签

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

    最新文章

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