有没有办法在Spring应用程序中手动启动/停止Kafka消费者？

Question

现在，我有一个Spring Boot CLI应用程序，当应用程序启动时，它会自动启动Kafka消费者。我的任务是更新提供API的应用程序，允许在特定条件下启动或停止Kafka消费者。所以，我将使用SpringBootStarterWeb创建该API。但我找不到一种方法来手动管理消费过程。我需要的是

• 在不使用消费者的情况下启动API

关于如何手动管理消费过程的任何建议？

技术细节：

• @KafkaListener用于创建侦听器

Answer 1

如果您的应用程序中配置了多个消费者，则必须使用唯一密钥来区分他们
下面的示例考虑多消费者配置。

消费者配置示例：

consumers:
      "fun-consumer-key":
        topic:  fun.topic-1
        consumerProcessOnStart: org.kafka.ProcessConsumedMessage
        consumerProperties:
          "[bootstrap.servers]": localhost:9092,localhost:9093
          // Other consumer configs

消费者工厂侦听器：

@Component
@Slf4j
public class ConsumerFactoryListener<K, V> implements Listener<K, V> {

  @Override
  public void consumerAdded(final String id, final Consumer<K, V> consumer) {
    //
  }

  @Override
  public void consumerRemoved(final String id, final Consumer<K, V> consumer) {
    //
  }

}

用于保存使用者对象的AppPropertiesConfig：

@UtilityClass
public class AppPropertiesConfig {

  private static Map<String, Object> configConsumers = new ConcurrentHashMap<>();

  public static Map<String, Object> getConfigConsumers() {
    return consumerMap;
  }
}

消息监听器：Ack

    public class AutoAckMessageListener<K, V> extends
        implements MessageListener<K, V> {
    
      private final ProcessConsumedMessage<K, V> consumeMessage;
    
      
      @Override
      public void onMessage(ConsumerRecord<K, V> consumerRecord) {
        onMessageConsume(consumerRecord, consumeMessage);
      }
    }

    public class AckMessageListener<K, V> extends
        implements AcknowledgingMessageListener<K, V> {
    
      private final ProcessConsumedMessage<K, V> consumeMessage;
    
      
      @Override
      public void onMessage(ConsumerRecord<K, V> consumerRecord, final Acknowledgment acknowledgment) {
        onMessageConsume(consumerRecord, consumeMessage);
        acknowledgment.acknowledge();
      }
    }

  // You can put this method in an abstract class and both listener classes can extend this Abstract class with onMessageConsume method
  public void onMessageConsume(ConsumerRecord<K, V> consumerRecord,
      final ProcessConsumedMessage<K, V> consumeMessage) throws InterruptedException {
    
    // Your custom processing implementation
    consumeMessage.process(consumerRecord.key(), consumerRecord.value());
  }

初始化消息侦听器：

  public MessageListener getListener(String className) {
    final ProcessConsumedMessage consumeMessage = (ProcessConsumedMessage) getClassFromName();
    MessageListener listener;
    if (isAutoCommitEnabled) {
      listener = new AutoAckMessageListener(consumeMessage);
    } else {
      listener = new AckMessageListener(consumeMessage);
    }
    return listener;
  }

启动消费者：

public void startConsumer(final String key, final String topic,
          final Object messageListener, final Map<String, Object> consumerProperties) {
        
        
        // Check already created, start and return
        ConcurrentMessageListenerContainer<K, V> container =
            (ConcurrentMessageListenerContainer<K, V>) AppPropertiesConfig
            .getConfigConsumers().get(key);  // key - "fun-consumer-key"
        if (container != null) {
          if (!container.isRunning()) {
           
            container.start();
          }
          return;
        }
        
        final DefaultKafkaConsumerFactory<K, V> factory = new DefaultKafkaConsumerFactory<>(
            consumerProperties);
        factory.addListener(consumerFactoryListener);
    
        final ConcurrentKafkaListenerContainerFactory<K, V> containerFactory
            = new ConcurrentKafkaListenerContainerFactory<>();
        containerFactory.setConsumerFactory(factory);
    
        final ContainerProperties containerProperties = containerFactory.getContainerProperties();
        containerProperties.setPollTimeout(pollTimeout);
    
        // auto-commit??
        if (!isAutoCommitEnabled) {
          containerProperties.setAckMode(AckMode.MANUAL_IMMEDIATE);
        }

        containerFactory.setErrorHandler(getErrorHandler(<some retry configurations object>));
        // create the container
        container = containerFactory.createContainer(topic);
        container.setupMessageListener(messageListener);
        // start
        container.start();
        AppPropertiesConfig.getConfigConsumers().put(key, container);
      }
    
      private SeekToCurrentErrorHandler getErrorHandler() {
          // Provide your error handler. Ex: SeekToCurrentErrorHandler
      }

阻止消费者：

  public void stopConsumer(final String key) {
    if (StringUtils.isBlank(key)) {
      return;
    }
    final ConcurrentMessageListenerContainer<K, V> container
        = (ConcurrentMessageListenerContainer<K, V>) AppPropertiesConfig
        .getConfigConsumers().get(key);
    if (container == null || !container.isRunning()) {
      throw new Exception();
    }
    try {
      container.stop();
    } catch (Exception e) {
      // log
      throw e;
    } finally {
      AppPropertiesConfig.getConfigConsumers().remove(key);
    }
  }