不使用@Kafkalistener的Spring boot consumer

Question

我试图在不使用@Kafkalistener的情况下编写kafka consumer，下面是我用于配置侦听器的代码行：

@Configuration
    @EnableKafka
    public class KafkaConfig {

      @Value("${kafka.bootstrap-servers}")
      private String bootstrapServers;

      @Bean
      public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        // list of host:port pairs used for establishing the initial connections to the Kafka cluster
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
            bootstrapServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
            StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
            StringDeserializer.class);
        // allows a pool of processes to divide the work of consuming and processing records
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "org");
        // automatically reset the offset to the earliest offset
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        return props;
      }

      @Bean
      public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
      }

      @Bean
      public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
          ContainerProperties containerProperties=new ContainerProperties("in.t");
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());

        return factory;
      }

      @Bean
      public Consumer receiver() {
        return new Consumer();
      }
    }

在这里，我如何配置topic和listener方法，我的consumer类可以有多个方法。

另外，我想知道在将@kafkalistener与Kafka流一起使用时是否会遇到任何潜在问题。

附言：我不想使用@KafkaListener。

Answer 1

使用Spring Boot和Spring Kafka，可以创建没有@KafkaListener的消费者，并创建额外的显式Kafka bean（ConcurrentMessageListenerContainer等）：

1. 申请。yml-创建连接配置。他们将被注入Kafka的资产

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: "org.apache.kafka.common.serialization.StringSerializer"
      value-serializer: "org.springframework.kafka.support.serializer.JsonSerializer"
    listener:
      type: BATCH
    consumer:
      clientId: "applicationClientId"
      groupId: "applicationGroup"
      keyDeserializer: "org.apache.kafka.common.serialization.StringDeserializer"
      valueDeserializer: "org.springframework.kafka.support.serializer.JsonDeserializer"
      maxPollRecords: 50
      fetchMinSize: 50
      autoOffsetReset: earliest
      properties:
        spring:
          json:
            trusted:
              packages: "*"

@Configuration
@EnableKafka
@RequiredArgsConstructor //Lombok annotation
public class KafkaConsumerConfig implements KafkaListenerConfigurer {

    private final BeanFactory beanFactory;
    private final SomethingApplicationService somethingApplicationService

    @SneakyThrows //Lombok annotation
    @Override
    public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) {
        // The following may be in a loop with different services and topics
        val endpoint = new MethodKafkaListenerEndpoint<String, KafkaReceiptRequest>();

        endpoint.setBeanFactory(beanFactory);
        endpoint.setBean(somethingApplicationService);
        endpoint.setMethod(somethingApplicationService.getClass().getDeclaredMethod("processList", List.class));
        endpoint.setId("Unique id for this endpoint");
        endpoint.setTopics("topic1", ...);
        endpoint.setMessageHandlerMethodFactory(messageHandlerMethodFactory());

        registrar.registerEndpoint(endpoint);
    }

    @Bean
    public MessageHandlerMethodFactory messageHandlerMethodFactory() {
        return new DefaultMessageHandlerMethodFactory();
    }

}

@Service
public class SomethingApplicationService {

    public void processList(List<Data> list) {
        //processing logic
    }

}

@Data //Lombok annotation
public class Data {
    private String name;
    private String value;
}

Answer 2

>

@kafkalistener不适用于Kafka Streams。它适用于普通的消费者。Kafka Stream可以通过StreamsBuilderFactoryBean和特定的@Bean来管理KStream。

如果您不想使用@kafkalistener，则需要使用手册kafkalistener容器创建。Kafka的KafkaListenerContainerFactory可以用于此目的，但仅限于Spring Kafka2.2之后，而且绝对不能用于Spring Boot。

因此，您别无选择，除非手动创建一个ConnettMessageListenerContainer。这里已经通过ContainerProperties注入了目标MessageListener。对于您自定义的消费者POJO，您需要考虑将其包装到RecordMessagingMessageListenerAdapter中。并且只有最后一个必须被注入到ConnettMessageListenerContainer中。

这就是@KafkaListener的底层工作原理。