Kafka的使用

// 如果要修改分区数，只需修改配置值重启项目即可。修改分区数并不会导致数据的丢失，但是分区数只能增大不能减小
// 删除Topic机会少，可使用命令行工具或者GUI工具操作
@Bean
public NewTopic initialTopic() {
  return new NewTopic("test-topic", 8, Short.parseShort("2"));
}

// 发送消息
kafkaTemplate.send("test-topic", new UserBean("kun", 18));
// 发送消息，取key的hashcode发送到指定的partition
kafkaTemplate.send("test-topic", "key-1", new UserBean("Jack", 20));
// 发送消息，根据传入的partition发送到指定的partition
kafkaTemplate.send("test-topic", 0, "", new UserBean("Jane", 21));
// 发送消息，根据传入的partition发送到指定的partition，并添加当前时间戳作为消息头
kafkaTemplate.send("test-topic", 0, DateUtil.currentSeconds(), "", new UserBean("Lot", 33));

UserBean user = new UserBean("kun", 18);
ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send("test-topic", user);
future.addCallback(sendResult -> {
  log.info("{}", sendResult.getRecordMetadata());
  log.info("{}", sendResult.getProducerRecord());
}, throwable -> log.info("{}", throwable));

// 需要设置spring.kafka.producer.transaction-id-prefix
kafkaTemplate.setAllowNonTransactional(true);
kafkaTemplate.send("test-topic-1", "key-1", new UserBean("Jack", 20));
kafkaTemplate.send("test-topic-1", 0, "", new UserBean("Jane", 21));
kafkaTemplate.setAllowNonTransactional(false);
// 发送事务消息
kafkaTemplate.executeInTransaction(kafkaOperations -> {
  kafkaTemplate.send("test-topic-1", new UserBean("Transaction-1", 18));
  kafkaTemplate.send("test-topic-1", new UserBean("Transaction-2", 18));
  if(true){
    throw new RuntimeException("模拟异常");
  }
  kafkaTemplate.send("test-topic-1", new UserBean("Transaction-3", 18));
  return true;
});

// spring boot本身自带的日志打印
@Bean
public ProducerListener<Object, Object> kafkaProducerLoggingListener() {
  return new LoggingProducerListener<>();
}

// 自定义功能
@Bean
public ProducerListener<Object, Object> kafkaProducerListener() {
  ProducerListener<Object, Object> listener = new ProducerListener<Object, Object>(){
    private int errorCounter = 0;
    private int successCounter = 0;

    @Override
    public void onSuccess(ProducerRecord<Object, Object> producerRecord, RecordMetadata recordMetadata) {
      successCounter++;
      System.out.println("成功个数" + successCounter);
    }

    @Override
    public void onError(ProducerRecord<Object, Object> producerRecord, Exception exception) {
      errorCounter++;
      System.out.println("失败个数" + errorCounter);
    }
  };
  return listener;
}

@KafkaListener(topics = {"test-topic"}, 
               groupId = "consumer-kun-app",
               containerGroup = "consumer-kun-app",
               errorHandler = "kafkaListenerErrorHandler")
public void onMessage(ConsumerRecord<String, UserBean> record,
                      @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key) throws Exception {
  System.out.println("简单消费："+record.topic()+"-"+record.partition() + "-" +record.value());
  throw new Exception("模拟异常");
}

// 当不需要将失败的消息发送到其它队列时返回null即可
@Bean
public KafkaListenerErrorHandler kafkaListenerErrorHandler() {
  // 一下业务逻辑是重置offset，重新消费出现异常的消息，
  ConsumerAwareListenerErrorHandler kafkaListenerErrorHandler = (message, exception, consumer) -> {
    MessageHeaders headers = message.getHeaders();
    String topicName = headers.get(KafkaHeaders.RECEIVED_TOPIC, String.class)
    Integer partitionId = headers.get(KafkaHeaders.RECEIVED_PARTITION_ID, Integer.class)
    Long offset = headers.get(KafkaHeaders.OFFSET, Long.class);
    consumer.seek(new TopicPartition(topicName, partitionId), offset);
    return null;
  };
  return kafkaListenerErrorHandler;
}

@KafkaListener(
  topics = {"test-topic"},
  groupId = "consumer-kun-app",
  containerGroup = "consumer-kun-app",
  errorHandler = "kafkaListenerErrorHandler")
public void onMessage(List<ConsumerRecord<String, UserBean>> records) throws Exception {
  for (ConsumerRecord<String, UserBean> record : records) {
    System.out.println("简单消费："+record.topic()+"-"+record.partition() + "-" +record.value());
  }
  throw new Exception("模拟异常");
}

// 将批次中的所有位移都重置为批次中最小的offset
@Bean
public KafkaListenerErrorHandler kafkaListenerErrorHandler() {
  ConsumerAwareListenerErrorHandler kafkaListenerErrorHandler = (message, exception, consumer) -> {
    MessageHeaders headers = message.getHeaders();
    List<String> topics = headers.get(KafkaHeaders.RECEIVED_TOPIC, List.class);
    List<Integer> partitions = headers.get(KafkaHeaders.RECEIVED_PARTITION_ID, List.class);
    List<Long> offsets = headers.get(KafkaHeaders.OFFSET, List.class);
    Map<TopicPartition, Long> offsetsToReset = new HashMap<>();
    for (int i = 0; i < topics.size(); i++) {
      int index = i;
      offsetsToReset.compute(new TopicPartition(topics.get(i), partitions.get(i)),
                             (k, v) -> v == null ? offsets.get(index) : Math.min(v,offsets.get(index)));
    }
    offsetsToReset.forEach((k, v) -> consumer.seek(k, v));
    return null;
  };
  return kafkaListenerErrorHandler;
}

/**
 * JsonDeserializer限制了可序列化的包名称，需要继承开放权限
 * spring.kafka.consumer.value-deserializer=com.kun.kafka.AppJsonDeserializer
 *
 * 也可配置
 * spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
 * spring.kafka.consumer.properties.spring.json.trusted.packages=*
 */
public class AppJsonDeserializer extends JsonDeserializer {

  @Override
  public void configure(Map configs, boolean isKey) {
    super.configure(configs, isKey);
    getTypeMapper().addTrustedPackages("com.kun.kafka.bean");
  }
}

@KafkaListener(
  topics = {"test-topic"}, 
  groupId = "consumer-kun-app",
  containerGroup = "consumer-kun-app")
public void onMessage(@Payload UserBean record, 
                      @Header(KafkaHeaders.RECEIVED_TOPIC) String topicName) {
  System.out.println("topicName: " + topicName);
  System.out.println("record: " + record);
}

@KafkaListener(
  topics = {"test-topic"}, 
  groupId = "consumer-kun-app", 
  containerGroup = "consumer-kun-app",
  errorHandler = "kafkaListenerErrorHandler")
@SendTo("test-error-topic")
public void onMessage(@Payload UserBean record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topicName,
                      @Header(KafkaHeaders.RECEIVED_TIMESTAMP) Long time) throws Exception {
  System.out.println("topicName: " + topicName);
  System.out.println("record: " + record);
  System.out.println("time: " + time);
  throw new Exception("模拟异常");
}

@Bean
public KafkaListenerErrorHandler kafkaListenerErrorHandler() {
  ConsumerAwareListenerErrorHandler handler = (message, exception, consumer) -> message.getPayload();
  return handler;
}

@Bean
public KStream<String, String> kStream(StreamsBuilder streamsBuilder){
  KStream<String, String> stream = streamsBuilder.stream("word-topic");
  stream.flatMapValues((values) -> Arrays.asList(values.split(" "))).
    groupBy((k, v)-> v).
    count().
    toStream().
    to("result-topic", Produced.with(Serdes.String(), Serdes.Long()));
  return stream;
}