org.apache.kafka.clients.consumer.Consumer.subscribe()方法的使用及代码示例

x33g5p2x  于2022-01-18 转载在 其他  
字(11.7k)|赞(0)|评价(0)|浏览(424)

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

Consumer.subscribe介绍

暂无

代码示例

代码示例来源：origin: openzipkin/brave

@Override public void subscribe(Collection<String> topics, ConsumerRebalanceListener callback) {
 delegate.subscribe(topics, callback);
}

代码示例来源：origin: openzipkin/brave

public void subscribe(Pattern pattern) {
 delegate.subscribe(pattern, NO_OP_CONSUMER_REBALANCE_LISTENER);
}

代码示例来源：origin: openzipkin/brave

@Override public void subscribe(Collection<String> topics) {
 delegate.subscribe(topics);
}

代码示例来源：origin: openzipkin/brave

@Override public void subscribe(Pattern pattern, ConsumerRebalanceListener callback) {
 delegate.subscribe(pattern, callback);
}

代码示例来源：origin: azkaban/azkaban

/**
 * Dynamically tune subscription only for the topic that dependencies need.
 */
@VisibleForTesting
synchronized void consumerSubscriptionRebalance() {
 log.debug("Subscribed Topics " + this.consumer.subscription());
 if (!this.subscribedTopics.isEmpty()) {
  final Iterator<String> iter = this.subscribedTopics.iterator();
  final List<String> topics = new ArrayList<>();
  while (iter.hasNext()) {
   topics.add(iter.next());
  }
  this.subscribedTopics.clear();
  //re-subscribe topics that are needed
  this.consumer.subscribe(topics);
 }
}

代码示例来源：origin: azkaban/azkaban

public KafkaEventMonitor(final DependencyPluginConfig pluginConfig) {
 this.initKafkaClient(pluginConfig);
 this.consumer.subscribe(Arrays.asList("AzEvent_Init_Topic"));
 if (!this.subscribedTopics.isEmpty()) {
  this.consumerSubscriptionRebalance();
 }
 this.depInstances = new KafkaDepInstanceCollection();
}

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

consumer.subscribe(topics, lease);
} else {
 consumer.subscribe(topicPattern, lease);

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

consumer.subscribe(topics, lease);
} else {
 consumer.subscribe(topicPattern, lease);

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

consumer.subscribe(topics, lease);
} else {
 consumer.subscribe(topicPattern, lease);

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

consumer.subscribe(topics, lease);
} else {
 consumer.subscribe(topicPattern, lease);

代码示例来源：origin: linkedin/cruise-control

consumerProps.setProperty(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, Integer.toString(Integer.MAX_VALUE));
_metricConsumer = new KafkaConsumer<>(consumerProps);
_metricConsumer.subscribe(Pattern.compile(metricReporterTopic), new ConsumerRebalanceListener() {
 @Override
 public void onPartitionsRevoked(Collection<TopicPartition> collection) {

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

consumer.subscribe(topics, lease);

代码示例来源：origin: linkedin/cruise-control

Consumer<String, CruiseControlMetric> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList(TOPIC));
long startMs = System.currentTimeMillis();
HashSet<Integer> expectedMetricTypes = new HashSet<>(Arrays.asList((int) ALL_TOPIC_BYTES_IN.id(),

代码示例来源：origin: spring-projects/spring-kafka

@SuppressWarnings("unchecked")
@Test
public void stopContainerAfterException() throws Exception {
  assertThat(this.config.deliveryLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.pollLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.errorLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.closeLatch.await(10, TimeUnit.SECONDS)).isTrue();
  MessageListenerContainer container = this.registry.getListenerContainer(CONTAINER_ID);
  assertThat(container.isRunning()).isFalse();
  InOrder inOrder = inOrder(this.consumer);
  inOrder.verify(this.consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class));
  inOrder.verify(this.consumer).poll(Duration.ofMillis(ContainerProperties.DEFAULT_POLL_TIMEOUT));
  inOrder.verify(this.consumer).wakeup();
  inOrder.verify(this.consumer).unsubscribe();
  inOrder.verify(this.consumer).close();
  inOrder.verifyNoMoreInteractions();
}

代码示例来源：origin: spring-projects/spring-kafka

@SuppressWarnings("unchecked")
@Test
public void remaingRecordsReceived() throws Exception {
  assertThat(this.config.deliveryLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.commitLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.pollLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.errorLatch.await(10, TimeUnit.SECONDS)).isTrue();
  InOrder inOrder = inOrder(this.consumer);
  inOrder.verify(this.consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class));
  inOrder.verify(this.consumer).poll(Duration.ofMillis(ContainerProperties.DEFAULT_POLL_TIMEOUT));
  inOrder.verify(this.consumer).commitSync(
      Collections.singletonMap(new TopicPartition("foo", 0), new OffsetAndMetadata(1L)));
  inOrder.verify(this.consumer).commitSync(
      Collections.singletonMap(new TopicPartition("foo", 0), new OffsetAndMetadata(2L)));
  inOrder.verify(this.consumer).commitSync(
      Collections.singletonMap(new TopicPartition("foo", 1), new OffsetAndMetadata(1L)));
  assertThat(this.config.count).isEqualTo(4);
  assertThat(this.config.contents).containsExactly("foo", "bar", "baz", "qux");
  assertThat(this.config.remaining).containsExactly("qux", "fiz", "buz");
}

代码示例来源：origin: spring-projects/spring-kafka

@SuppressWarnings("unchecked")
@Test
public void stopContainerAfterException() throws Exception {
  assertThat(this.config.deliveryLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.pollLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.errorLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.closeLatch.await(10, TimeUnit.SECONDS)).isTrue();
  MessageListenerContainer container = this.registry.getListenerContainer(CONTAINER_ID);
  assertThat(container.isRunning()).isFalse();
  InOrder inOrder = inOrder(this.consumer);
  inOrder.verify(this.consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class));
  inOrder.verify(this.consumer).poll(Duration.ofMillis(ContainerProperties.DEFAULT_POLL_TIMEOUT));
  inOrder.verify(this.consumer).wakeup();
  inOrder.verify(this.consumer).unsubscribe();
  inOrder.verify(this.consumer).close();
  inOrder.verifyNoMoreInteractions();
  assertThat(this.config.count).isEqualTo(4);
  assertThat(this.config.contents.toArray()).isEqualTo(new String[]
      { "foo", "bar", "baz", "qux" });
}

代码示例来源：origin: spring-projects/spring-kafka

@SuppressWarnings("unchecked")
@Test
public void discardRemainingRecordsFromPollAndSeek() throws Exception {
  assertThat(this.config.deliveryLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.commitLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.pollLatch.await(10, TimeUnit.SECONDS)).isTrue();
  this.registry.stop();
  assertThat(this.config.closeLatch.await(10, TimeUnit.SECONDS)).isTrue();
  InOrder inOrder = inOrder(this.consumer);
  inOrder.verify(this.consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class));
  inOrder.verify(this.consumer).poll(Duration.ofMillis(ContainerProperties.DEFAULT_POLL_TIMEOUT));
  Map<TopicPartition, OffsetAndMetadata> offsets = new LinkedHashMap<>();
  offsets.put(new TopicPartition("foo", 0), new OffsetAndMetadata(2L));
  offsets.put(new TopicPartition("foo", 1), new OffsetAndMetadata(1L));
  inOrder.verify(this.consumer).commitSync(offsets);
  inOrder.verify(this.consumer).seek(new TopicPartition("foo", 1), 1L);
  inOrder.verify(this.consumer).seek(new TopicPartition("foo", 2), 0L);
  inOrder.verify(this.consumer).poll(Duration.ofMillis(ContainerProperties.DEFAULT_POLL_TIMEOUT));
  offsets = new LinkedHashMap<>();
  offsets.put(new TopicPartition("foo", 1), new OffsetAndMetadata(2L));
  offsets.put(new TopicPartition("foo", 2), new OffsetAndMetadata(2L));
  inOrder.verify(this.consumer).commitSync(offsets);
  inOrder.verify(this.consumer).poll(Duration.ofMillis(ContainerProperties.DEFAULT_POLL_TIMEOUT));
  assertThat(this.config.count).isEqualTo(7);
  assertThat(this.config.contents.toArray()).isEqualTo(new String[]
      { "foo", "bar", "baz", "qux", "qux", "fiz", "buz" });
}

代码示例来源：origin: spring-projects/spring-kafka

@SuppressWarnings("unchecked")
@Test
public void stopContainerAfterException() throws Exception {
  assertThat(this.config.deliveryLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.commitLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.pollLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.errorLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.closeLatch.await(10, TimeUnit.SECONDS)).isTrue();
  MessageListenerContainer container = this.registry.getListenerContainer(CONTAINER_ID);
  assertThat(container.isRunning()).isFalse();
  InOrder inOrder = inOrder(this.consumer);
  inOrder.verify(this.consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class));
  inOrder.verify(this.consumer).poll(Duration.ofMillis(ContainerProperties.DEFAULT_POLL_TIMEOUT));
  inOrder.verify(this.consumer).commitSync(
      Collections.singletonMap(new TopicPartition("foo", 0), new OffsetAndMetadata(1L)));
  inOrder.verify(this.consumer).commitSync(
      Collections.singletonMap(new TopicPartition("foo", 0), new OffsetAndMetadata(2L)));
  inOrder.verify(this.consumer).commitSync(
      Collections.singletonMap(new TopicPartition("foo", 1), new OffsetAndMetadata(1L)));
  inOrder.verify(this.consumer).wakeup();
  inOrder.verify(this.consumer).unsubscribe();
  inOrder.verify(this.consumer).close();
  inOrder.verifyNoMoreInteractions();
  assertThat(this.config.count).isEqualTo(4);
  assertThat(this.config.contents.toArray()).isEqualTo(new String[]
      { "foo", "bar", "baz", "qux" });
}

代码示例来源：origin: spring-projects/spring-kafka

@SuppressWarnings("unchecked")
@Test
public void discardRemainingRecordsFromPollAndSeek() throws Exception {
  assertThat(this.config.deliveryLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.pollLatch.await(10, TimeUnit.SECONDS)).isTrue();
  this.registry.stop();
  assertThat(this.config.closeLatch.await(10, TimeUnit.SECONDS)).isTrue();
  InOrder inOrder = inOrder(this.consumer, this.producer);
  inOrder.verify(this.consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class));
  inOrder.verify(this.consumer).poll(Duration.ofMillis(ContainerProperties.DEFAULT_POLL_TIMEOUT));
  inOrder.verify(this.producer).beginTransaction();
  inOrder.verify(this.consumer).seek(new TopicPartition("foo", 0), 0L);
  inOrder.verify(this.consumer).seek(new TopicPartition("foo", 1), 0L);
  inOrder.verify(this.consumer).seek(new TopicPartition("foo", 2), 0L);
  inOrder.verify(this.producer).abortTransaction();
  inOrder.verify(this.consumer).poll(Duration.ofMillis(ContainerProperties.DEFAULT_POLL_TIMEOUT));
  inOrder.verify(this.producer).beginTransaction();
  Map<TopicPartition, OffsetAndMetadata> offsets = new LinkedHashMap<>();
  offsets.put(new TopicPartition("foo", 0), new OffsetAndMetadata(2L));
  offsets.put(new TopicPartition("foo", 1), new OffsetAndMetadata(2L));
  offsets.put(new TopicPartition("foo", 2), new OffsetAndMetadata(2L));
  inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID);
  inOrder.verify(this.producer).commitTransaction();
  assertThat(this.config.ehException).isInstanceOf(ListenerExecutionFailedException.class);
  assertThat(((ListenerExecutionFailedException) this.config.ehException).getGroupId()).isEqualTo(CONTAINER_ID);
}

代码示例来源：origin: spring-projects/spring-kafka

assertThat(this.config.closeLatch.await(10, TimeUnit.SECONDS)).isTrue();
InOrder inOrder = inOrder(this.consumer);
inOrder.verify(this.consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class));
inOrder.verify(this.consumer).poll(Duration.ofMillis(ContainerProperties.DEFAULT_POLL_TIMEOUT));
inOrder.verify(this.consumer).commitSync(

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

Consumer

相关文章

    查看更多

    热门标签

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

    最新文章

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