kafka.message.Message.payload()方法的使用及代码示例

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

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

Message.payload介绍

暂无

代码示例

代码示例来源：origin: apache/incubator-pinot

  1. public byte[] getMessageAtIndex(int index) {
  2.  return messageList.get(index).message().payload().array();
  3. }

代码示例来源：origin: prestodb/presto

  1. ByteBuffer message = messageAndOffset.message().payload();
  2. if (message != null) {
  3.   messageData = new byte[message.remaining()];

代码示例来源：origin: apache/incubator-pinot

  1. public int getMessageOffsetAtIndex(int index) {
  2.  return messageList.get(index).message().payload().arrayOffset();
  3. }

代码示例来源：origin: alibaba/jstorm

  1. @SuppressWarnings("unchecked")
  2. public Iterable<List<Object>> generateTuples(Message msg) {
  3.   Iterable<List<Object>> tups = null;
  4.   ByteBuffer payload = msg.payload();
  5.   if (payload == null) {
  6.     return null;
  7.   }
  8.   tups = Arrays.asList(Utils.tuple(Utils.toByteArray(payload)));
  9.   return tups;
  10. }

代码示例来源：origin: apache/incubator-gobblin

  1. @Override
  2. public byte[] getMessageBytes() {
  3.  return getBytes(this.messageAndOffset.message().payload());
  4. }

代码示例来源：origin: apache/incubator-gobblin

  1. private ByteArrayBasedKafkaRecord getMockMessageAndOffset(ByteBuffer payload) {
  2.  MessageAndOffset mockMessageAndOffset = mock(MessageAndOffset.class);
  3.  Message mockMessage = mock(Message.class);
  4.  when(mockMessage.payload()).thenReturn(payload);
  5.  when(mockMessageAndOffset.message()).thenReturn(mockMessage);
  6.  return new Kafka08ConsumerRecord(mockMessageAndOffset);
  7. }

代码示例来源：origin: Graylog2/graylog2-server

  1. final byte[] payloadBytes = ByteBufferUtils.readBytes(messageAndOffset.message().payload());
  2. if (LOG.isTraceEnabled()) {
  3.   final byte[] keyBytes = ByteBufferUtils.readBytes(messageAndOffset.message().key());

代码示例来源：origin: alibaba/jstorm

  1. public EmitState emit(SpoutOutputCollector collector) {
  2.   if (emittingMessages.isEmpty()) {
  3.     fillMessages();
  4.   }
  5.   int count = 0;
  6.   while (true) {
  7.     MessageAndOffset toEmitMsg = emittingMessages.pollFirst();
  8.     if (toEmitMsg == null) {
  9.       return EmitState.EMIT_END;
  10.     }
  11.     count ++;
  12.     Iterable<List<Object>> tups = generateTuples(toEmitMsg.message());
  13.     if (tups != null) {
  14.       for (List<Object> tuple : tups) {
  15.         LOG.debug("emit message {}", new String(Utils.toByteArray(toEmitMsg.message().payload())));
  16.         collector.emit(tuple, new KafkaMessageId(partition, toEmitMsg.offset()));
  17.       }
  18.       if(count>=config.batchSendCount) {
  19.         break;
  20.       }
  21.     } else {
  22.       ack(toEmitMsg.offset());
  23.     }
  24.   }
  25.   if (emittingMessages.isEmpty()) {
  26.     return EmitState.EMIT_END;
  27.   } else {
  28.     return EmitState.EMIT_MORE;
  29.   }
  30. }

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

  1. if (running) {
  2.   messagesInFetch++;
  3.   final ByteBuffer payload = msg.message().payload();
  4.   final long offset = msg.offset();

代码示例来源：origin: pinterest/secor

  1. ByteBuffer payload = messageAndOffset.message().payload();
  2. payloadBytes = new byte[payload.limit()];
  3. payload.get(payloadBytes);

代码示例来源：origin: edwardcapriolo/IronCount

  1. public static String getMessage(Message message) {
  2.   ByteBuffer buffer = message.payload();
  3.   byte[] bytes = new byte[buffer.remaining()];
  4.   buffer.get(bytes);
  5.   return new String(bytes);
  6.  }
  7. }

代码示例来源：origin: linkedin/camus

  1. /**
  2.  * Fetches the next Kafka message and stuffs the results into the key and
  3.  * value
  4.  *
  5.  * @param etlKey
  6.  * @return true if there exists more events
  7.  * @throws IOException
  8.  */
  9. public KafkaMessage getNext(EtlKey etlKey) throws IOException {
  10.  if (hasNext()) {
  11.   MessageAndOffset msgAndOffset = messageIter.next();
  12.   Message message = msgAndOffset.message();
  13.   byte[] payload = getBytes(message.payload());
  14.   byte[] key = getBytes(message.key());
  15.   if (payload == null) {
  16.    log.warn("Received message with null message.payload(): " + msgAndOffset);
  17.   }
  18.   etlKey.clear();
  19.   etlKey.set(kafkaRequest.getTopic(), kafkaRequest.getLeaderId(), kafkaRequest.getPartition(), currentOffset,
  20.     msgAndOffset.offset() + 1, message.checksum());
  21.   etlKey.setMessageSize(msgAndOffset.message().size());
  22.   currentOffset = msgAndOffset.offset() + 1; // increase offset
  23.   currentCount++; // increase count
  24.   return new KafkaMessage(payload, key, kafkaRequest.getTopic(), kafkaRequest.getPartition(),
  25.     msgAndOffset.offset(), message.checksum());
  26.  } else {
  27.   return null;
  28.  }
  29. }

代码示例来源：origin: org.apache.storm/storm-kafka

  1. public static Iterable<List<Object>> generateTuples(MessageMetadataSchemeAsMultiScheme scheme, Message msg, Partition partition, long offset) {
  2.   ByteBuffer payload = msg.payload();
  3.   if (payload == null) {
  4.     return null;
  5.   }
  6.   return scheme.deserializeMessageWithMetadata(payload, partition, offset);
  7. }

代码示例来源：origin: fjfd/microscope

  1. private static void printMessages(ByteBufferMessageSet messageSet) throws UnsupportedEncodingException {
  2.   for (MessageAndOffset messageAndOffset : messageSet) {
  3.     ByteBuffer payload = messageAndOffset.message().payload();
  4.     byte[] bytes = new byte[payload.limit()];
  5.     payload.get(bytes);
  6.     System.out.println(new String(bytes, "UTF-8"));
  7.   }
  8. }

代码示例来源：origin: jetoile/hadoop-unit

  1. private void printMessages(ByteBufferMessageSet messageSet) throws UnsupportedEncodingException {
  2.   for (MessageAndOffset messageAndOffset : messageSet) {
  3.     ByteBuffer payload = messageAndOffset.message().payload();
  4.     byte[] bytes = new byte[payload.limit()];
  5.     payload.get(bytes);
  6.     LOG.debug(new String(bytes, "UTF-8"));
  7.   }
  8. }

代码示例来源：origin: com.butor/butor-kafka

  1. protected String mesasgeAndOffsetToString(MessageAndOffset messageAndOffset) {
  2.    ByteBuffer payload = messageAndOffset.message().payload();
  3.    byte[] bytes = new byte[payload.limit()];
  4.    payload.get(bytes);
  5.    return messageDecoder.fromBytes(bytes).toString();
  6. }

代码示例来源：origin: Allianzcortex/code_collection

  1. @SuppressWarnings("unchecked")
  2. public Iterable<List<Object>> generateTuples(Message msg) {
  3.   Iterable<List<Object>> tups = null;
  4.   ByteBuffer payload = msg.payload();
  5.   if (payload == null) {
  6.     return null;
  7.   }
  8.   tups = Arrays.asList(Utils.tuple(Utils.toByteArray(payload)));
  9.   return tups;
  10. }

代码示例来源：origin: org.apache.gobblin/gobblin-kafka-08

  1. @Override
  2. public byte[] getMessageBytes() {
  3.  return getBytes(this.messageAndOffset.message().payload());
  4. }

代码示例来源：origin: apache/apex-malhar

  1. public String getMessage(Message message)
  2. {
  3.  ByteBuffer buffer = message.payload();
  4.  byte[] bytes = new byte[buffer.remaining()];
  5.  buffer.get(bytes);
  6.  return new String(bytes);
  7. }

代码示例来源：origin: wurstmeister/storm-kafka-0.8-plus

  1. public static Iterable<List<Object>> generateTuples(KafkaConfig kafkaConfig, Message msg) {
  2.   Iterable<List<Object>> tups;
  3.   ByteBuffer payload = msg.payload();
  4.   ByteBuffer key = msg.key();
  5.   if (key != null && kafkaConfig.scheme instanceof KeyValueSchemeAsMultiScheme) {
  6.     tups = ((KeyValueSchemeAsMultiScheme) kafkaConfig.scheme).deserializeKeyAndValue(Utils.toByteArray(key), Utils.toByteArray(payload));
  7.   } else {
  8.     tups = kafkaConfig.scheme.deserialize(Utils.toByteArray(payload));
  9.   }
  10.   return tups;
  11. }

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

Message

相关文章

    查看更多

    热门标签

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

    最新文章

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