kafka.javaapi.producer.Producer.send()方法的使用及代码示例

x33g5p2x  于2022-01-26 转载在 其他  
字(7.3k)|赞(0)|评价(0)|浏览(249)

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

Producer.send介绍

暂无

代码示例

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

  1. private void publish(JsonNode message)
  2.   throws IOException {
  3.  if (!keepIndexing) {
  4.   avroDataStream.close();
  5.   avroDataStream = null;
  6.   return;
  7.  }
  8.  KeyedMessage<String, byte[]> data =
  9.    new KeyedMessage<String, byte[]>("airlineStatsEvents", message.toString().getBytes("UTF-8"));
  10.  producer.send(data);
  11. }

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

  1. bytes);
  2. producer.send(data);

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

  1. @Override
  2. public void addResults(QueryStatusInfo statusInfo, QueryData data)
  3. {
  4.   if (types.get() == null && statusInfo.getColumns() != null) {
  5.     types.set(getTypes(statusInfo.getColumns()));
  6.   }
  7.   if (data.getData() != null) {
  8.     checkState(types.get() != null, "Data without types received!");
  9.     List<Column> columns = statusInfo.getColumns();
  10.     for (List<Object> fields : data.getData()) {
  11.       ImmutableMap.Builder<String, Object> builder = ImmutableMap.builder();
  12.       for (int i = 0; i < fields.size(); i++) {
  13.         Type type = types.get().get(i);
  14.         Object value = convertValue(fields.get(i), type);
  15.         if (value != null) {
  16.           builder.put(columns.get(i).getName(), value);
  17.         }
  18.       }
  19.       producer.send(new KeyedMessage<>(topicName, count.getAndIncrement(), builder.build()));
  20.     }
  21.   }
  22. }

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

  1. producer.send(messagesToWrite);
  2.   messagesToWrite.clear();
  3. producer.send(messagesToWrite);

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

  1. producer.send(messagesToWrite);
  2.   messagesToWrite.clear();
  3. producer.send(messagesToWrite);

代码示例来源：origin: org.apache.kafka/kafka_2.9.2

  1. public void run() {
  2.  try{
  3.   while(true) {
  4.    KeyedMessage<byte[], byte[]> data = producerDataChannel.receiveRequest();
  5.    if(!data.equals(shutdownMessage)) {
  6.     producer.send(data);
  7.     if(logger.isDebugEnabled()) logger.debug("Sending message %s".format(new String(data.message())));
  8.    }
  9.    else
  10.     break;
  11.   }
  12.   logger.info("Producer thread " + threadName + " finished running");
  13.  } catch (Throwable t){
  14.   logger.fatal("Producer thread failure due to ", t);
  15.  } finally {
  16.   shutdownComplete.countDown();
  17.  }
  18. }

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

  1. try {
  2.  KeyedMessage keyedMessage = new KeyedMessage("TrackingMonitoringEvent", message);
  3.  producer.send(keyedMessage);
  4.  break;
  5. } catch (Exception e) {

代码示例来源：origin: rakam-io/rakam

  1. @Override
  2. public void store(Event event) {
  3.   GenericDatumWriter writer = new SourceFilteredRecordWriter(event.properties().getSchema(), GenericData.get(), sourceFields);
  4.   ByteBuf buffer = Unpooled.buffer(100);
  5.   BinaryEncoder encoder = EncoderFactory.get().directBinaryEncoder(
  6.       new ByteBufOutputStream(buffer), null);
  7.   try {
  8.     writer.write(event.properties(), encoder);
  9.   } catch (Exception e) {
  10.     throw new RuntimeException("Couldn't serialize event", e);
  11.   }
  12.   try {
  13.     producer.send(new KeyedMessage<>(event.project() + "_" + event.collection(), buffer.array()));
  14.   } catch (FailedToSendMessageException e) {
  15.     throw new RuntimeException("Couldn't send event to Kafka", e);
  16.   }
  17. }

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

  1. private static List<Message> writeKafka(String topic, int numOfMessages) {
  2.  List<Message> messages = new ArrayList<Message>();
  3.  List<KeyedMessage<String, String>> kafkaMessages = new ArrayList<KeyedMessage<String, String>>();
  4.  for (int i = 0; i < numOfMessages; i++) {
  5.   Message msg = new Message(RANDOM.nextInt());
  6.   messages.add(msg);
  7.   kafkaMessages.add(new KeyedMessage<String, String>(topic, Integer.toString(i), gson.toJson(msg)));
  8.  }
  9.  Properties producerProps = cluster.getProps();
  10.  producerProps.setProperty("serializer.class", StringEncoder.class.getName());
  11.  producerProps.setProperty("key.serializer.class", StringEncoder.class.getName());
  12.  Producer<String, String> producer = new Producer<String, String>(new ProducerConfig(producerProps));
  13.  try {
  14.   producer.send(kafkaMessages);
  15.  } finally {
  16.   producer.close();
  17.  }
  18.  return messages;
  19. }

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

  1. private Producer mockProducerThirdSendSucceed() {
  2.  Producer mockProducer = EasyMock.createMock(Producer.class);
  3.  mockProducer.send((KeyedMessage) EasyMock.anyObject());
  4.  EasyMock.expectLastCall().andThrow(new RuntimeException("dummyException")).times(2);
  5.  mockProducer.send((KeyedMessage) EasyMock.anyObject());
  6.  EasyMock.expectLastCall().times(1);
  7.  mockProducer.close();
  8.  EasyMock.expectLastCall().anyTimes();
  9.  EasyMock.replay(mockProducer);
  10.  return mockProducer;
  11. }

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

  1. private Producer mockProducerSendThrowsException() {
  2.  Producer mockProducer = EasyMock.createMock(Producer.class);
  3.  mockProducer.send((KeyedMessage) EasyMock.anyObject());
  4.  EasyMock.expectLastCall().andThrow(new RuntimeException("dummyException")).anyTimes();
  5.  mockProducer.close();
  6.  EasyMock.expectLastCall().anyTimes();
  7.  EasyMock.replay(mockProducer);
  8.  return mockProducer;
  9. }

代码示例来源：origin: caskdata/cdap

  1. void publish(List<KeyedMessage<String, byte[]>> messages) {
  2.  // Clear the interrupt flag, otherwise it won't be able to publish
  3.  boolean threadInterrupted = Thread.interrupted();
  4.  try {
  5.   producer.send(messages);
  6.  } finally {
  7.   // Reset the interrupt flag if needed
  8.   if (threadInterrupted) {
  9.    Thread.currentThread().interrupt();
  10.   }
  11.  }
  12. }

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

  1. @Override
  2. protected void execute(Object key, Map event, OutputCollector collector) throws Exception {
  3.   try {
  4.     String output = new ObjectMapper().writeValueAsString(event);
  5.     // partition key may cause data skew
  6.     //producer.send(new KeyedMessage(this.topicId, key, output));
  7.     producer.send(new KeyedMessage(this.topicId, output));
  8.   } catch (Exception ex) {
  9.     LOG.error(ex.getMessage(), ex);
  10.     throw ex;
  11.   }
  12. }

代码示例来源：origin: spotify/ffwd

  1. @Override
  2.   public Void call() throws Exception {
  3.     producer.send(eventConverter.toMessage(event));
  4.     return null;
  5.   }
  6. }, executorService);

代码示例来源：origin: hengyunabc/metrics-kafka

  1. @Override
  2.   public void run() {
  3.     try {
  4.     KeyedMessage<String, String> message = new KeyedMessage<String, String>(
  5.         topic, "" + count++, mapper.writeValueAsString(result));
  6.       producer.send(message);
  7.     } catch (Exception e) {
  8.       logger.error("send metrics to kafka error!", e);
  9.     }
  10.   }
  11. });

代码示例来源：origin: ottogroup/SPQR

  1. public void run() {
  2.     try {
  3.       kafkaProducer.send(new KeyedMessage<byte[], byte[]>(topicId, jsonMapper.writeValueAsBytes(registry)));
  4.     } catch (JsonProcessingException e) {
  5.       logger.error("Failed to send message to kafka [topic="+topicId+"]. Reason: " + e.getMessage(), e);
  6.     }
  7.   }
  8. });

代码示例来源：origin: com.github.hackerwin7/jlib-utils

  1. /**
  2.  * sendKafkaMsg batch
  3.  * @param msgs
  4.  */
  5. public void sendKafkaMsg(List<KafkaMsg> msgs) {
  6.   if(msgs.size() > 0)
  7.     producer.send(toKeyedMessage(msgs));
  8. }

代码示例来源：origin: gwenshap/kafka-examples

  1. @Override
  2. public void produce(String s) {
  3.   KeyedMessage<String, String> message = new KeyedMessage<String, String>(topic, null, s);
  4.   producer.send(message);
  5. }

代码示例来源：origin: Stratio/Decision

  1. @Override
  2. public void process(Iterable<StratioStreamingMessage> messages) throws Exception {
  3.   List<KeyedMessage<String, String>> kafkaMessages = new ArrayList<>();
  4.   for (StratioStreamingMessage message : messages) {
  5.     kafkaMessages.add(new KeyedMessage<String, String>(message.getStreamName(), getSerializer().deserialize(
  6.         message)));
  7.   }
  8.   getProducer().send(kafkaMessages);
  9. }

代码示例来源：origin: olacabs/fabric

  1. @Before
  2. public void setUp() throws InitializationException {
  3.   processor = spy(new KafkaWriter());
  4.   producer = mock(Producer.class);
  5.   processor.setTopicOnJsonPath(true);
  6.   processor.setKafkaTopic("/default");
  7.   processor.setIngestionPoolSize(10);
  8.   doReturn(producer).when(processor).getProducer();
  9.   doNothing().when(producer).send(anyList());
  10. }

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

Producer

相关文章

    查看更多

    热门标签

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

    最新文章

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