import org.springframework.kafka.listener.config.ContainerProperties; //导入依赖的package包/类
private void createSystemConsumer(String name, MessageListener<String, String> consumeEvent) {
    log.info("Creating kafka consumer for topic {}", name);
    ContainerProperties containerProps = new ContainerProperties(name);

    Map<String, Object> props = kafkaProperties.buildConsumerProperties();
    if (name.equals(applicationProperties.getKafkaSystemTopic())) {
        props.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());
    }
    ConsumerFactory<String, String> factory = new DefaultKafkaConsumerFactory<>(props);

    ConcurrentMessageListenerContainer<String, String> container =
        new ConcurrentMessageListenerContainer<>(factory, containerProps);
    container.setupMessageListener(consumeEvent);
    container.start();
    log.info("Successfully created kafka consumer for topic {}", name);
}
 

import org.springframework.kafka.listener.config.ContainerProperties; //导入依赖的package包/类
/**
 * Create topic consumer.
 * @param tenant the kafka topic
 */
public void createKafkaConsumer(String tenant) {
    StopWatch stopWatch = StopWatch.createStarted();
    try {
        log.info("START - SETUP:CreateTenant:kafka consumer tenantKey: {}", tenant);
        ConcurrentMessageListenerContainer<String, String> container = consumers.get(tenant);
        if (container != null) {
            if (!container.isRunning()) {
                container.start();
            }
        } else {
            ContainerProperties containerProps = new ContainerProperties(tenant);
            container = new ConcurrentMessageListenerContainer<>(consumerFactory, containerProps);
            container.setupMessageListener((MessageListener<String, String>) consumer::consumeEvent);
            container.setBeanName(tenant);
            container.start();
            consumers.put(tenant, container);
        }
        log.info("STOP  - SETUP:CreateTenant:kafka consumer tenantKey: {}, result: OK, time = {} ms",
            tenant, stopWatch.getTime());
    } catch (Exception e) {
        log.error("STOP  - SETUP:CreateTenant:kafka consumer tenantKey: {}, result: FAIL, error: {}, time = {} ms",
            tenant, e.getMessage(), stopWatch.getTime(), e);
    }
}
 

import org.springframework.kafka.listener.config.ContainerProperties; //导入依赖的package包/类
@Bean
public MessageListenerContainer messageListenerContainer(
        ConsumerFactory<String, DefaultAvMessage> consumerFactory,
        MessageListener<String, AvMessage> messageListener,
        ThreadPoolTaskScheduler kafkaClientThreadPoolTaskScheduler
) {
    ContainerProperties props = new ContainerProperties(resultTopic);
    // shouldn't be necessary but the default scheduler is not destroyed after shutdown
    props.setScheduler(kafkaClientThreadPoolTaskScheduler);

    MessageListenerContainer container = new ConcurrentMessageListenerContainer<>(
            consumerFactory,
            props
    );
    container.setupMessageListener(messageListener);

    return container;
}
 

import org.springframework.kafka.listener.config.ContainerProperties; //导入依赖的package包/类
@Test
public void testAutoCommit() throws Exception {
    LOG.info("Start testAutoCommit");
    ContainerProperties containerProps = new ContainerProperties("topic3", "topic4");
    final CountDownLatch latch = new CountDownLatch(4);
    containerProps.setMessageListener((MessageListener<Integer, String>) message -> {
        LOG.info("received: " + message);
        latch.countDown();
    });
    KafkaMessageListenerContainer<Integer, String> container = createContainer(containerProps,
            IntegerDeserializer.class, StringDeserializer.class);
    container.setBeanName("testAutoCommit");
    container.start();
    Thread.sleep(5000); // wait a bit for the container to start
    KafkaTemplate<Integer, String> template = createTemplate(IntegerSerializer.class, StringSerializer.class);
    template.setDefaultTopic("topic3");
    template.sendDefault(0, "foo");
    template.sendDefault(2, "bar");
    template.sendDefault(0, "baz");
    template.sendDefault(2, "qux");
    template.flush();
    assertTrue(latch.await(60, TimeUnit.SECONDS));
    container.stop();
    LOG.info("Stop testAutoCommit");
}
 

import org.springframework.kafka.listener.config.ContainerProperties; //导入依赖的package包/类
@Bean
public MessageListenerContainer fileServerMessageListenerContainer(
        ConsumerFactory<String, DefaultAvMessage> consumerFactory,
        MessageListener<String, AvMessage> fileServerMessageListener,
        ThreadPoolTaskScheduler kafkaServerThreadPoolTaskScheduler
) {
    ContainerProperties props = new ContainerProperties(fileTopic);
    // shouldn't be necessary but the default scheduler is not destroyed after shutdown
    props.setScheduler(kafkaServerThreadPoolTaskScheduler);

    MessageListenerContainer container = new ConcurrentMessageListenerContainer<>(
            consumerFactory,
            props
    );
    container.setupMessageListener(fileServerMessageListener);

    return container;
}
 

import org.springframework.kafka.listener.config.ContainerProperties; //导入依赖的package包/类
private void createConsumer(String name) {
    log.info("Creating kafka consumer for tenant {}", name);
    ContainerProperties containerProps = new ContainerProperties(name);
    ConcurrentMessageListenerContainer<String, String> container =
        new ConcurrentMessageListenerContainer<>(consumerFactory, containerProps);
    container.setupMessageListener((MessageListener<String, String>) timelineConsumer::consumeEvent);
    container.start();
    log.info("Successfully created kafka consumer for tenant {}", name);
}
 

import org.springframework.kafka.listener.config.ContainerProperties; //导入依赖的package包/类
private void createCommandConsumer(String name) {
    log.info("Creating kafka command consumer for topic {}", name);
    ContainerProperties containerProps = new ContainerProperties(name);

    Map<String, Object> props = kafkaProperties.buildConsumerProperties();
    props.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());
    ConsumerFactory<String, String> factory = new DefaultKafkaConsumerFactory<>(props);

    ConcurrentMessageListenerContainer<String, String> container =
        new ConcurrentMessageListenerContainer<>(factory, containerProps);
    container.setupMessageListener((MessageListener<String, String>) commandConsumer::consumeEvent);
    container.start();
    log.info("Successfully created kafka command consumer for topic {}", name);
}
 

import org.springframework.kafka.listener.config.ContainerProperties; //导入依赖的package包/类
@Before
public void setup() throws Exception {
	Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testT", "false", embeddedKafka);
	DefaultKafkaConsumerFactory<String, String> cf =
		new DefaultKafkaConsumerFactory<>(consumerProps);
	ContainerProperties containerProperties = new ContainerProperties(TEST_TOPIC);
	container = new KafkaMessageListenerContainer<>(cf, containerProperties);
	final BlockingQueue<ConsumerRecord<String, String>> records = new LinkedBlockingQueue<>();
	container.setupMessageListener((MessageListener<String, String>) record -> {
           log.error("Message received: " + record);
           records.add(record);
       });
	container.start();
	ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic());
	Map<String, Object> senderProps = KafkaTestUtils.senderProps(embeddedKafka.getBrokersAsString());
	ProducerFactory<String, String> pf =
		new DefaultKafkaProducerFactory<>(senderProps);
	template = new KafkaTemplate<>(pf);
	template.setDefaultTopic(TEST_TOPIC);
}
 

import org.springframework.kafka.listener.config.ContainerProperties; //导入依赖的package包/类
private KafkaMessageListenerContainer<Integer, IndexDTO> createContainerForDto(
        ContainerProperties containerProps, Object keyDeser, Object valDeser) {
    Map<String, Object> props = consumerProps(keyDeser, valDeser);
    DefaultKafkaConsumerFactory<Integer, IndexDTO> cf =
            new DefaultKafkaConsumerFactory<>(props);
    KafkaMessageListenerContainer<Integer, IndexDTO> container =
            new KafkaMessageListenerContainer<>(cf, containerProps);
    return container;
}
 

import org.springframework.kafka.listener.config.ContainerProperties; //导入依赖的package包/类
private KafkaMessageListenerContainer<Integer, String> createContainer(
        ContainerProperties containerProps, Object keyDeser, Object valDeser) {
    Map<String, Object> props = consumerProps(keyDeser, valDeser);
    DefaultKafkaConsumerFactory<Integer, String> cf =
            new DefaultKafkaConsumerFactory<>(props);
    KafkaMessageListenerContainer<Integer, String> container =
            new KafkaMessageListenerContainer<>(cf, containerProps);
    return container;
}
 

import org.springframework.kafka.listener.config.ContainerProperties; //导入依赖的package包/类
private KafkaMessageConsumer(String bootstrapServers,
                             String[] topics,
                             MessageTypeParameter messageType,
                             OffsetResetParameter offsetReset) {
    final Map<String, Object> consumerConfig =
            consumerConfig(bootstrapServers, UUID.randomUUID().toString(), messageType, offsetReset);
    final ConsumerFactory<String, String> consumerFactory = createConsumerFactory(consumerConfig);

    final ContainerProperties containerProperties = new ContainerProperties(topics);
    listenerContainer = new KafkaMessageListenerContainer<>(consumerFactory, containerProperties);

    messageListener = createMessageListener(messageType);
    listenerContainer.setupMessageListener(messageListener);
}
 

import org.springframework.kafka.listener.config.ContainerProperties; //导入依赖的package包/类
@Bean
public KafkaMessageListenerContainer<String, String> container(
		ConsumerFactory<String, String> consumerFactory,
		ConfigProperties config) {
	ContainerProperties containerProperties = new ContainerProperties(config.getTopic());
	containerProperties.setMessageListener(listener());
	containerProperties.setAckMode(AckMode.MANUAL_IMMEDIATE);
	return new KafkaMessageListenerContainer<>(consumerFactory, containerProperties);
}
 

import org.springframework.kafka.listener.config.ContainerProperties; //导入依赖的package包/类
@Bean
public KafkaMessageListenerContainer<String, String> container(
		ConsumerFactory<String, String> consumerFactory,
		ConfigProperties config) {
	ContainerProperties containerProperties = new ContainerProperties(config.getTopic());
	containerProperties.setMessageListener(listener());
	return new KafkaMessageListenerContainer<>(consumerFactory, containerProperties);
}
 

import org.springframework.kafka.listener.config.ContainerProperties; //导入依赖的package包/类
@Test
public void testReceive() throws Exception {
  String bootstrapServers = embeddedKafka.getBrokersAsString();
  LOGGER.info("bootstrapServers='{}'", bootstrapServers);

  ContainerProperties containerProperties =
      new ContainerProperties(JAVA_TOPIC);
  Receiver receiver = new Receiver();

  KafkaMessageListenerContainer<String, String> messageListenerContainer =
      ReceiverConfig.createMessageListenerContainer(
          containerProperties, bootstrapServers);
  messageListenerContainer.setupMessageListener(receiver);
  messageListenerContainer.start();

  // wait a bit for the container to start
  Thread.sleep(1000);

  KafkaTemplate<String, String> kafkaTemplate =
      SenderConfig.createKafkaTemplate(bootstrapServers);
  kafkaTemplate.send(JAVA_TOPIC, "Hello Java!");

  receiver.getLatch().await(10000, TimeUnit.MILLISECONDS);
  assertThat(receiver.getLatch().getCount()).isEqualTo(0);

  messageListenerContainer.stop();
}
 

import org.springframework.kafka.listener.config.ContainerProperties; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
  // set up the Kafka consumer properties
  Map<String, Object> consumerProperties =
      KafkaTestUtils.consumerProps("sender", "false", embeddedKafka);

  // create a Kafka consumer factory
  DefaultKafkaConsumerFactory<String, String> consumerFactory =
      new DefaultKafkaConsumerFactory<String, String>(consumerProperties);

  // set the topic that needs to be consumed
  ContainerProperties containerProperties = new ContainerProperties(SENDER_TOPIC);

  // create a Kafka MessageListenerContainer
  container = new KafkaMessageListenerContainer<>(consumerFactory, containerProperties);

  // create a thread safe queue to store the received message
  records = new LinkedBlockingQueue<>();

  // setup a Kafka message listener
  container.setupMessageListener(new MessageListener<String, String>() {
    @Override
    public void onMessage(ConsumerRecord<String, String> record) {
      LOGGER.debug("test-listener received message='{}'", record.toString());
      records.add(record);
    }
  });

  // start the container and underlying message listener
  container.start();

  // wait until the container has the required number of assigned partitions
  ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic());
}
 

import org.springframework.kafka.listener.config.ContainerProperties; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Bean
public ConcurrentMessageListenerContainer<String, String> kafkaListenerContainer() {
  ContainerProperties containerProps = new ContainerProperties(springIntegrationKafkaTopic);

  return (ConcurrentMessageListenerContainer<String, String>) new ConcurrentMessageListenerContainer<>(
      consumerFactory(), containerProps);
}
 

import org.springframework.kafka.listener.config.ContainerProperties; //导入依赖的package包/类
@Test
public void testSendIndexDTO() throws Exception {
    LOG.info("Start testSendIndexDTO");
    ContainerProperties containerProps = new ContainerProperties("topic1", "topic2");

    List<IndexDTO> dtos = prepareIndexableDtos(rdfHandler, "/data/discos/rmd18mddcw", null)
            .collect(Collectors.toList());

    Queue<IndexDTO> expectedDtos = new ArrayDeque<>(dtos);
    Queue<ExpectedActualDTOPair> receivedDtos = new ArrayDeque<>();

    final CountDownLatch latch = new CountDownLatch(3);

    containerProps.setMessageListener((MessageListener<Integer, IndexDTO>) message -> {
        LOG.info("received: " + message);
        IndexDTO expected = expectedDtos.remove();
        LOG.debug("expected: " + expected);
        IndexDTO actual = message.value();
        LOG.debug("actual: " + actual);
        receivedDtos.add(new ExpectedActualDTOPair(expected, actual));
        LOG.debug("Decrementing latch.");
        latch.countDown();
    });

    KafkaMessageListenerContainer<Integer, IndexDTO> container = createContainerForDto(containerProps, IntegerDeserializer.class, GenericJvmObjectDeserializer.class);
    container.setBeanName("testSendIndexDTO");
    container.start();
    Thread.sleep(5000); // wait a bit for the container to start
    KafkaTemplate<Integer, IndexDTO> template = createTemplate(IntegerSerializer.class, GenericJvmObjectSerializer.class);
    template.setDefaultTopic("topic1");

    prepareIndexableDtos(rdfHandler, "/data/discos/rmd18mddcw", null)
            .peek(dto -> LOG.debug("Prepared DTO {}", dto))
            .forEach(template::sendDefault);

    // do anything with the completablefuture returned by the template?

    template.flush();

    assertTrue(latch.await(120, TimeUnit.SECONDS));
    container.stop();

    LOG.info("Stop testSendIndexDTO");

    receivedDtos.forEach(pair -> assertEquals(pair.expected, pair.actual));
}
 

import org.springframework.kafka.listener.config.ContainerProperties; //导入依赖的package包/类
public static KafkaMessageListenerContainer<String, String> createMessageListenerContainer(
    ContainerProperties containerProperties,
    String bootstrapServers) {
  return new KafkaMessageListenerContainer<>(
      createConsumerFactory(bootstrapServers), containerProperties);
}
 

import org.springframework.kafka.listener.config.ContainerProperties; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
  // set up the Kafka consumer properties
  Map<String, Object> consumerProperties =
      KafkaTestUtils.consumerProps("sender_group", "false", AllSpringKafkaTests.embeddedKafka);

  // create a Kafka consumer factory
  DefaultKafkaConsumerFactory<String, String> consumerFactory =
      new DefaultKafkaConsumerFactory<String, String>(consumerProperties);

  // set the topic that needs to be consumed
  ContainerProperties containerProperties =
      new ContainerProperties(AllSpringKafkaTests.SENDER_TOPIC);

  // create a Kafka MessageListenerContainer
  container = new KafkaMessageListenerContainer<>(consumerFactory, containerProperties);

  // create a thread safe queue to store the received message
  records = new LinkedBlockingQueue<>();

  // setup a Kafka message listener
  container.setupMessageListener(new MessageListener<String, String>() {
    @Override
    public void onMessage(ConsumerRecord<String, String> record) {
      LOGGER.debug("test-listener received message='{}'", record.toString());
      records.add(record);
    }
  });

  // start the container and underlying message listener
  container.start();
  // wait until the container has the required number of assigned partitions
  ContainerTestUtils.waitForAssignment(container,
      AllSpringKafkaTests.embeddedKafka.getPartitionsPerTopic());
}