<dependencies>
  <dependency>
     <groupId>org.apache.kafka</groupId>
     <artifactId>kafka-clients</artifactId>
     <version>3.0.0</version>
  </dependency>
</dependencies>

package com.atguigu.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;

public class CustomProducer {
     public static void main(String[] args) throws 
        InterruptedException {
         // 1. создавать kafka Объект конфигурации производителя
         Properties properties = new Properties();
         // 2. Давать kafka Добавьте информацию о конфигурации в объект конфигурации: bootstrap.servers.
         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");

         // key,value Сериализация (обязательно): key.serializer, value.serializer.
         properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
         // 3. создавать kafka продюсерский объект
        KafkaProducer<String, String> kafkaProducer = new 
        KafkaProducer<String, String>(properties);
         // 4. вызов send метод, отправить сообщение
         for (int i = 0; i < 5; i++) {
             kafkaProducer.send(new 
             ProducerRecord<>("first","yy " + i));
         }
         // 5. закрыть ресурс
         kafkaProducer.close();
  }
}

bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

// 1. создавать kafka Объект конфигурации производителя
 Properties properties = new Properties();
 // 2. Давать kafka Объект конфигурации добавляет информацию о конфигурации.
 properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 
"hadoop102:9092");
 // key,value Сериализация (обязательно): key.serializer, value.serializer.
 properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
 
 properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
 // 3. создавать kafka продюсерский объект
 KafkaProducer<String, String> kafkaProducer = new 
 KafkaProducer<String, String>(properties);
 // 4. вызов send метод, отправить сообщение
 for (int i = 0; i < 5; i++) {
     // Добавить обратный вызов
     kafkaProducer.send(new ProducerRecord<>("first", "atguigu " + i), new Callback() {
             // Методсуществовать Producer получать ack часвызов,дляасинхронныйвызов
         @Override
         public void onCompletion(RecordMetadata metadata, 
            Exception exception) {
             if (exception == null) {
               // Без исключений, выводить информацию в консоль
               System.out.println(" тема: " + 
               metadata.topic() + "->" + "Раздел:" + metadata.partition());
             } else {
               //  Происходит печать исключений
               exception.printStackTrace();
              }
            }
         });
     //  После задержки вы увидите, что данные отправляются в разные разделы.
     Thread.sleep(2);
 }
 // 5. закрыть ресурс
 kafkaProducer.close();

// Отправить синхронно
 kafkaProducer.send(new ProducerRecord<>("first","kafka" + i)).get();

/**
* 1. реализовать интерфейс Partitioner
* 2. выполнить 3 индивидуальныйметод:partition,close,configure
* 3. писать partition Метод, возвращает номер раздела
*/
public class MyPartitioner implements Partitioner {
 /**
 * Вернуть раздел, соответствующий информации
 * @param topic тема
 * @param key новости key
 * @param keyBytes новости key Сериализованный массив байтов
 * @param value новости value
 * @param valueBytes новости value Сериализованный массив байтов
 * @param cluster Метаданные кластера могут просматривать информацию о разделах.
 * @return
 */
 @Override
 public int partition(String topic, Object key, byte[] 
keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
    // Получить новости
     String msgValue = value.toString();
     // создавать partition
     int partition;
     // Определите, содержит ли сообщение atguigu
     if (msgValue.contains("xxx")){
     	partition = 0;
     }else {
    	 partition = 1;
     }
     // Вернуть номер раздела
     return partition;
 }
 // закрыть ресурс
 @Override
 public void close() {
 }
 // Метод настройки
 @Override
 public void configure(Map<String, ?> configs) {
 } 
}

// добавить в Пользовательский разделитель
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.yy.kafka.producer.MyPartitioner");

// пакет.размер: размер пакета, по умолчанию 16K
 properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
// linger.ms: время ожидания, по умолчанию 0
 properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);
// RecordAccumulator: размер буфера, по умолчанию. 32M：buffer.memory
 properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);
// compress.type: сжатие, по умолчанию нет, настраиваемое значение gzip、snappy、lz4 и zstd
 properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");

 // настраивать acks
 properties.put(ProducerConfig.ACKS_CONFIG, "all");
 // Количество повторов повторные попытки, значение по умолчанию int Максимальное значение, 2147483647
 properties.put(ProducerConfig.RETRIES_CONFIG, 3);

// 1 Инициализировать транзакцию
void initTransactions();
// 2 Открыть транзакцию
void beginTransaction() throws ProducerFencedException;
// 3 существовать Зафиксировать потребленное возмещение в рамках транзакции (в основном используется для потребителей)
void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,String consumerGroupId) throws ProducerFencedException;
// 4 совершить транзакцию
void commitTransaction() throws ProducerFencedException;
// 5 Отменить транзакцию (операция, аналогичная откату транзакции)
void abortTransaction() throws ProducerFencedException;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
public class CustomProducerTransactions {
   public static void main(String[] args) throws InterruptedException {
     // 1. создавать kafka Объект конфигурации производителя
     Properties properties = new Properties();
     // 2. Давать kafka Объект конфигурации добавляет информацию о конфигурации.
     properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                    "hadoop102:9092");
     // key,value сериализация
     properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
     // настраиватьдела идентификатор (обязательно), транзакция id Назовите все, что вам нравится
     properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transaction_id_0");
     // 3. создавать kafka продюсерский объект
     KafkaProducer<String, String> kafkaProducer = new 
     KafkaProducer<String, String>(properties);
     // Инициализировать транзакцию
     kafkaProducer.initTransactions();
     // Открыть транзакцию
     kafkaProducer.beginTransaction();
     try {
         // 4. вызов send метод, отправить сообщение
         for (int i = 0; i < 5; i++) {
         // отправить сообщение
             kafkaProducer.send(new ProducerRecord<>("first","atguigu " + i));
         }
        // int i = 1 / 0;
         // совершить транзакцию
         kafkaProducer.commitTransaction();
     } catch (Exception e) {
         // Завершить транзакцию
         kafkaProducer.abortTransaction();
     } finally {
         // 5. закрыть ресурс
         kafkaProducer.close();
     }
 } 
}