feat: kafka生产者示例

TomShiDi · Mar 13, 2024 · 9a931d1 · 9a931d1
1 parent c42e657
commit 9a931d1
Show file tree

Hide file tree

Showing 6 changed files with 196 additions and 0 deletions.
diff --git a/kafka-demo/pom.xml b/kafka-demo/pom.xml
@@ -8,4 +8,12 @@
     </parent>
     <artifactId>kafka-demo</artifactId>
     <name>kafka-demo</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+            <version>3.0.0</version>
+        </dependency>
+    </dependencies>
 </project>
diff --git a/kafka-demo/src/main/java/com/tomshidi/kafka/partitioner/CustomPartitioner.java b/kafka-demo/src/main/java/com/tomshidi/kafka/partitioner/CustomPartitioner.java
@@ -0,0 +1,31 @@
+package com.tomshidi.kafka.partitioner;
+
+import org.apache.kafka.clients.producer.Partitioner;
+import org.apache.kafka.common.Cluster;
+
+import java.util.Map;
+
+/**
+ * @author TomShiDi
+ * @since 2024/3/13 14:39
+ */
+public class CustomPartitioner implements Partitioner {
+    @Override
+    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
+        // 1.计算分区号
+        String keyStr = key.toString();
+        int keyHash = keyStr.hashCode();
+        Integer partitionCount = cluster.partitionCountForTopic(topic);
+        return keyHash % partitionCount;
+    }
+
+    @Override
+    public void close() {
+
+    }
+
+    @Override
+    public void configure(Map<String, ?> configs) {
+
+    }
+}
diff --git a/kafka-demo/src/main/java/com/tomshidi/kafka/producer/CustomCallbackProducer.java b/kafka-demo/src/main/java/com/tomshidi/kafka/producer/CustomCallbackProducer.java
@@ -0,0 +1,38 @@
+package com.tomshidi.kafka.producer;
+
+import org.apache.kafka.clients.producer.*;
+import org.apache.kafka.common.serialization.StringSerializer;
+
+import java.util.Properties;
+
+/**
+ * @author TomShiDi
+ * @since 2024/3/13 10:01
+ */
+public class CustomCallbackProducer {
+    public static void main(String[] args) {
+        // 1.创建配置对象
+        Properties properties = new Properties();
+        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092,node2:9092");
+        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.tomshidi.kafka.partitioner.CustomPartitioner");
+        // 2.创建kafka生产者对象
+        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
+        // 3.发送数据
+        for (int i = 1; i < 11; i++) {
+            String message = String.format("第%d号服务员", i);
+            ProducerRecord<String, String> producerRecord = new ProducerRecord<>("first", i + "", message);
+            kafkaProducer.send(producerRecord,
+                    (recordMetadata, ex) -> {
+                if (ex == null) {
+                    int partition = recordMetadata.partition();
+                    long offset = recordMetadata.offset();
+                    System.out.printf("数据：%s\n分区：%d\n偏移量：%d\n", message, partition, offset);
+                }
+            });
+        }
+        // 4.关闭资源
+        kafkaProducer.close();
+    }
+}
diff --git a/kafka-demo/src/main/java/com/tomshidi/kafka/producer/CustomProducer.java b/kafka-demo/src/main/java/com/tomshidi/kafka/producer/CustomProducer.java
@@ -0,0 +1,32 @@
+package com.tomshidi.kafka.producer;
+
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringSerializer;
+
+import java.util.Properties;
+
+/**
+ * @author TomShiDi
+ * @since 2024/3/13 10:01
+ */
+public class CustomProducer {
+    public static void main(String[] args) {
+        // 1.创建配置对象
+        Properties properties = new Properties();
+        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092,node2:9092");
+        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+        // 2.创建kafka生产者对象
+        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
+        // 3.发送数据
+        for (int i = 1; i < 11; i++) {
+            String message = String.format("第%d号服务员", i);
+            ProducerRecord<String, String> producerRecord = new ProducerRecord<>("first", message);
+            kafkaProducer.send(producerRecord);
+        }
+        // 4.关闭资源
+        kafkaProducer.close();
+    }
+}
diff --git a/kafka-demo/src/main/java/com/tomshidi/kafka/producer/CustomSyncProducer.java b/kafka-demo/src/main/java/com/tomshidi/kafka/producer/CustomSyncProducer.java
@@ -0,0 +1,36 @@
+package com.tomshidi.kafka.producer;
+
+import org.apache.kafka.clients.producer.*;
+import org.apache.kafka.common.serialization.StringSerializer;
+
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+
+/**
+ * 生产者同步发送
+ * @author TomShiDi
+ * @since 2024/3/13 10:42
+ */
+public class CustomSyncProducer {
+    public static void main(String[] args) throws ExecutionException, InterruptedException {
+        // 1.创建配置
+        Properties properties = new Properties();
+        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092");
+        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+        // 2.创建生产者对象
+        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
+        // 3. 循环发送
+        for (int i = 1; i < 11; i++) {
+            String message = String.format("第%d号服务员", i);
+            ProducerRecord<String, String> record = new ProducerRecord<>("first", message);
+            producer.send(record, (recordMetadata, ex) -> {
+                if (ex == null) {
+                    int partition = recordMetadata.partition();
+                    long offset = recordMetadata.offset();
+                    System.out.printf("数据：%s\n分区：%d\n偏移量：%d\n", message, partition, offset);
+                }
+            }).get();
+        }
+    }
+}
diff --git a/kafka-demo/src/main/java/com/tomshidi/kafka/producer/CustomThroughputProducer.java b/kafka-demo/src/main/java/com/tomshidi/kafka/producer/CustomThroughputProducer.java
@@ -0,0 +1,51 @@
+package com.tomshidi.kafka.producer;
+
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringSerializer;
+
+import java.util.Properties;
+
+/**
+ * @author TomShiDi
+ * @since 2024/3/13 10:01
+ */
+public class CustomThroughputProducer {
+    public static void main(String[] args) {
+        // 1.创建配置对象
+        Properties properties = new Properties();
+        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092,node2:9092");
+        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.tomshidi.kafka.partitioner.CustomPartitioner");
+        // 性能优化配置
+        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
+        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
+        properties.put(ProducerConfig.LINGER_MS_CONFIG, 10);
+        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
+        // ack应答机制配置
+        properties.put(ProducerConfig.ACKS_CONFIG, "all");
+        properties.put(ProducerConfig.RETRIES_CONFIG, 3);
+        // 幂等性
+        properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
+
+        // 2.创建kafka生产者对象
+        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
+        // 3.发送数据
+        for (int i = 1; i < 11; i++) {
+            String message = String.format("第%d号服务员", i);
+            ProducerRecord<String, String> producerRecord = new ProducerRecord<>("first", i + "", message);
+            kafkaProducer.send(producerRecord,
+                    (recordMetadata, ex) -> {
+                if (ex == null) {
+                    int partition = recordMetadata.partition();
+                    long offset = recordMetadata.offset();
+                    System.out.printf("数据：%s\n分区：%d\n偏移量：%d\n", message, partition, offset);
+                }
+            });
+        }
+        // 4.关闭资源
+        kafkaProducer.close();
+    }
+}