Kafka - 4 Spring集成Kafka

@RestController
public class KafkaProducer {
  @Autowired
  private KafkaTemplate<String, Object> kafkaTemplate;
  // 发送消息
  @GetMapping("/kafka/normal/{message}")
  public void sendMessage1(@PathVariable("message") String normalMessage) {
      kafkaTemplate.send("topic1", normalMessage);
  }
}

@Component
public class KafkaConsumer {
  // 消费监听
  @KafkaListener(topics = {"topic1"})
  public void onMessage1(ConsumerRecord<?, ?> record){
      // 消费的哪个topic、partition的消息,打印出消息内容
      System.out.println("简单消费："+record.topic()+"-"+record.partition()+"-"+record.value());
  }
}

public KafkaTemplate(ProducerFactory<K, V> producerFactory) {
  this(producerFactory, false);
}

public KafkaTemplate(ProducerFactory<K, V> producerFactory, boolean autoFlush) {
  this(producerFactory, autoFlush, (Map)null);
}

public KafkaTemplate(ProducerFactory<K, V> producerFactory, boolean autoFlush, @Nullable Map<String, Object> configOverrides) {
  this.logger = new LogAccessor(LogFactory.getLog(this.getClass()));
  this.producers = new ConcurrentHashMap();
  this.micrometerTags = new HashMap();
  this.clusterIdLock = new ReentrantLock();
  this.beanName = "kafkaTemplate";
  this.messageConverter = new MessagingMessageConverter();
  this.producerListener = new LoggingProducerListener();
  this.closeTimeout = ProducerFactoryUtils.DEFAULT_CLOSE_TIMEOUT;
  this.micrometerEnabled = true;
  this.observationRegistry = ObservationRegistry.NOOP;
  Assert.notNull(producerFactory, "'producerFactory' cannot be null");
  this.autoFlush = autoFlush;
  this.micrometerEnabled = KafkaUtils.MICROMETER_PRESENT;
  this.customProducerFactory = !CollectionUtils.isEmpty(configOverrides);
  if (this.customProducerFactory) {
      this.producerFactory = producerFactory.copyWithConfigurationOverride(configOverrides);
  } else {
      this.producerFactory = producerFactory;
  }

  this.transactional = this.producerFactory.transactionCapable();
}

@Bean
public ProducerFactory producerFactory() {
    return new DefaultKafkaProducerFactory<>(producerConfigs());
}

@Bean
public Map producerConfigs() {
    Map props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    // See https://kafka.apache.org/41/documentation/#producerconfigs for more properties
    return props;
}

@Bean
public KafkaTemplate kafkaTemplate() {
    return new KafkaTemplate(producerFactory());
}

@Bean
public KafkaTemplate<String, Object> reliableTemplate(ProducerFactory<String, Object> pf) {
  Map<String, Object> overrides = new HashMap<>();
  overrides.put(ProducerConfig.ACKS_CONFIG, "all");
  overrides.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
  overrides.put(ProducerConfig.RETRIES_CONFIG, 3);

  return new KafkaTemplate<>(pf, overrides);
}

CompletableFuture<SendResult<K, V>> sendDefault(V data);

CompletableFuture<SendResult<K, V>> sendDefault(K key, V data);

CompletableFuture<SendResult<K, V>> sendDefault(Integer partition, K key, V data);

CompletableFuture<SendResult<K, V>> sendDefault(Integer partition, Long timestamp, K key, V data);

CompletableFuture<SendResult<K, V>> send(String topic, V data);

CompletableFuture<SendResult<K, V>> send(String topic, K key, V data);

CompletableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, V data);

CompletableFuture<SendResult<K, V>> send(String topic, Integer partition, Long timestamp, K key, V data);

CompletableFuture<SendResult<K, V>> send(ProducerRecord<K, V> record);

CompletableFuture<SendResult<K, V>> send(Message<?> message);

Map<MetricName, ? extends Metric> metrics();

List<PartitionInfo> partitionsFor(String topic);

<T> T execute(ProducerCallback<K, V, T> callback);

<T> T executeInTransaction(OperationsCallback<K, V, T> callback);

// Flush the producer.
void flush();

interface ProducerCallback<K, V, T> {

    T doInKafka(Producer<K, V> producer);

}

interface OperationsCallback<K, V, T> {

    T doInOperations(KafkaOperations<K, V> operations);

}

public CompletableFuture<SendResult<K, V>> sendDefault(@Nullable V data) {
  return this.send(this.defaultTopic, data);
}

import org.springframework.messaging.support.MessageBuilder;
import org.springframework.kafka.support.KafkaHeaders;

Message<FileProcessingTask> message = MessageBuilder
        .withPayload(task)
        .setHeader(KafkaHeaders.TOPIC, kafkaConfig.getFileProcessingTopic())
        .setHeader(KafkaHeaders.KEY, request.fileMd5())
        .setHeader(KafkaHeaders.TIMESTAMP, System.currentTimeMillis())
        .build();

kafkaTemplate.send(message);

public void sendToKafka(final MyOutputData data) {
  final ProducerRecord<String, String> record = createRecord(data);

  CompletableFuture<SendResult<String, String>> future = template.send(record);
  future.whenComplete((result, ex) -> {
      if (ex == null) {
          handleSuccess(data);
      }
      else {
          handleFailure(data, record, ex);
      }
  });
}

public void sendToKafka(final MyOutputData data) {
  final ProducerRecord<String, String> record = createRecord(data);

  try {
      template.send(record).get(10, TimeUnit.SECONDS);
      handleSuccess(data);
  }
  catch (ExecutionException e) {
      handleFailure(data, record, e.getCause());
  }
  catch (TimeoutException | InterruptedException e) {
      handleFailure(data, record, e);
  }
}

public interface ProducerListener<K, V> {
  default void onSuccess(ProducerRecord<K, V> producerRecord, RecordMetadata recordMetadata) {
  }

    default void onError(ProducerRecord<K, V> producerRecord, RecordMetadata recordMetadata, Exception exception) {
  }
}

@Component
public class KafkaProducerLogListener implements ProducerListener<String, Object> {

    @Override
    public void onSuccess(ProducerRecord<String, Object> record, RecordMetadata metadata) {
        log.info("sent ok topic={}, partition={}, offset={}, key={}",
                metadata.topic(), metadata.partition(), metadata.offset(), record.key());
    }

    @Override
    public void onError(ProducerRecord<String, Object> record,
                        RecordMetadata metadata,
                        Exception exception) {
        String meta = (metadata == null)
                ? "meta=null"
                : String.format("partition=%d, offset=%d", metadata.partition(), metadata.offset());

        log.error("send failed topic={}, key={}, {}",
                record.topic(), record.key(), meta, exception);
    }
}

@Bean
public KafkaTemplate<String, Object> kafkaTemplate(
  ProducerFactory<String, Object> pf,
  ProducerListener<String, Object> producerListener) {
  KafkaTemplate<String, Object> template = new KafkaTemplate<>(pf);
  template.setProducerListener(producerListener);
  return template;
}

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class TestTopicListener {
    @KafkaListener(topics = "test", groupId = "test-group")
    public void onMessage(String msg) {
        System.out.println("收到 test topic 消息: " + msg);
    }
}

public interface MessageListener<K, V> {
  void onMessage(ConsumerRecord<K, V> data);
}

public interface AcknowledgingMessageListener<K, V> {
  void onMessage(ConsumerRecord<K, V> data, Acknowledgment acknowledgment);
}

public interface ConsumerAwareMessageListener<K, V> extends MessageListener<K, V> {
  void onMessage(ConsumerRecord<K, V> data, Consumer<?, ?> consumer);
}

public interface AcknowledgingConsumerAwareMessageListener<K, V> extends MessageListener<K, V> {
  void onMessage(ConsumerRecord<K, V> data, Acknowledgment acknowledgment, Consumer<?, ?> consumer);
}

public interface BatchMessageListener<K, V> {
  void onMessage(List<ConsumerRecord<K, V>> data);
}

public interface BatchAcknowledgingMessageListener<K, V> {
  void onMessage(List<ConsumerRecord<K, V>> data, Acknowledgment acknowledgment);
}

public interface BatchConsumerAwareMessageListener<K, V> extends BatchMessageListener<K, V> {
  void onMessage(List<ConsumerRecord<K, V>> data, Consumer<?, ?> consumer);
}

public interface BatchAcknowledgingConsumerAwareMessageListener<K, V> extends BatchMessageListener<K, V> {
  void onMessage(List<ConsumerRecord<K, V>> data, Acknowledgment acknowledgment, Consumer<?, ?> consumer);
}