Apache Kafka is an open-source distributed event streaming platform used by thousands of companies for high-performance data pipelines, streaming analytics, data integration, and mission-critical applications.

Apache Kafka 是一个开源分布式事件流平台，已被数千家公司使用，作为高性能数据通道、数据流分析、数据采集和关键任务应用。

搭建环境是基于Docker，Docker Compose。

1. 创建目录。

  1. mkdir -p kafka/zoo1/data
  2. mkdir -p kafka/zoo1/datalog
  3. mkdir -p kafka/zoo1/logs
  5. mkdir -p kafka/zoo2/data
  6. mkdir -p kafka/zoo2/datalog
  7. mkdir -p kafka/zoo2/logs
  9. mkdir -p kafka/zoo3/data
  10. mkdir -p kafka/zoo3/datalog
  11. mkdir -p kafka/zoo3/logs
  13. mkdir -p kafka/kafka1
  14. mkdir -p kafka/kafka2
  15. mkdir -p kafka/kafka3

Kafka目录如下：

  1. .
  2. ├── docker-compose.yml
  3. ├── kafka1
  4. ├── kafka2
  5. ├── kafka3
  6. ├── zoo1
  7. │   ├── data
  8. │   ├── datalog
  9. │   └── logs
  10. ├── zoo2
  11. │   ├── data
  12. │   ├── datalog
  13. │   └── logs
  14. └── zoo3
  15.     ├── data
  16.     ├── datalog
  17.     └── logs

2. 创建docker-compose.yml文件并放在根目录。

  1. version: '3.8'
  3. networks:
  4.   app-tier:
  5.     driver: bridge
  7. services:
  8.   zoo1:
  9.     container_name: zoo1
  10.     hostname: zoo1
  11.     image: zookeeper:3.7
  12.     restart: always
  13.     ports:
  14.       - 2181:2181
  15.     environment:
  16.       ZOO_MY_ID: 1
  17.       ZOO_SERVERS: server.1=zoo1:2888:3888;2181 server.2=zoo2:2888:3888;2181 server.3=zoo3:2888:3888;2181
  18.       ZOO_LOG4J_PROP: INFO,ROLLINGFILE
  19.     volumes:
  20.       - ./zoo1/data:/data
  21.       - ./zoo1/datalog:/datalog
  22.       - ./zoo1/logs:/logs
  23.     networks:
  24.       - app-tier
  26.   zoo2:
  27.     container_name: zoo2
  28.     hostname: zoo2
  29.     image: zookeeper:3.7
  30.     restart: always
  31.     ports:
  32.       - 2182:2181
  33.     environment:
  34.       ZOO_MY_ID: 2
  35.       ZOO_SERVERS: server.1=zoo1:2888:3888;2181 server.2=zoo2:2888:3888;2181 server.3=zoo3:2888:3888;2181
  36.       ZOO_LOG4J_PROP: INFO,ROLLINGFILE
  37.     volumes:
  38.       - ./zoo2/data:/data
  39.       - ./zoo2/datalog:/datalog
  40.       - ./zoo2/logs:/logs
  41.     networks:
  42.       - app-tier
  44.   zoo3:
  45.     container_name: zoo3
  46.     hostname: zoo3
  47.     image: zookeeper:3.7
  48.     restart: always
  49.     ports:
  50.       - 2183:2181
  51.     environment:
  52.       ZOO_MY_ID: 3
  53.       ZOO_SERVERS: server.1=zoo1:2888:3888;2181 server.2=zoo2:2888:3888;2181 server.3=zoo3:2888:3888;2181
  54.       ZOO_LOG4J_PROP: INFO,ROLLINGFILE
  55.     volumes:
  56.       - ./zoo3/data:/data
  57.       - ./zoo3/datalog:/datalog
  58.       - ./zoo3/logs:/logs
  59.     networks:
  60.       - app-tier
  62.   kafka1:
  63.     container_name: kafka1
  64.     hostname: kafka1
  65.     image: bitnami/kafka:2.8.1
  66.     restart: always
  67.     ports:
  68.       - '9091:9091'
  69.     volumes:
  70.       - ./kafka1:/bitnami/kafka
  71.     environment:
  72.       - KAFKA_BROKER_ID=1
  73.       - ALLOW_PLAINTEXT_LISTENER=yes
  74.       - KAFKA_CFG_LISTENERS=PLAINTEXT://:9091
  75.       - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://192.168.3.77:9091
  76.       - KAFKA_CFG_ZOOKEEPER_CONNECT=zoo1:2181,zoo2:2182,zoo3:2183
  77.       - KAFKA_HEAP_OPTS=-Xmx512m -Xms512m
  78.     networks:
  79.       - app-tier
  80.     depends_on:
  81.       - zoo1
  82.       - zoo2
  83.       - zoo3
  85.   kafka2:
  86.     container_name: kafka2
  87.     hostname: kafka2
  88.     image: bitnami/kafka:2.8.1
  89.     restart: always
  90.     ports:
  91.       - '9092:9092'
  92.     volumes:
  93.       - ./kafka2:/bitnami/kafka
  94.     environment:
  95.       - KAFKA_BROKER_ID=2
  96.       - ALLOW_PLAINTEXT_LISTENER=yes
  97.       - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092
  98.       - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://192.168.3.77:9092
  99.       - KAFKA_CFG_ZOOKEEPER_CONNECT=zoo1:2181,zoo2:2182,zoo3:2183
  100.       - KAFKA_HEAP_OPTS=-Xmx512m -Xms512m
  101.     networks:
  102.       - app-tier
  103.     depends_on:
  104.       - zoo1
  105.       - zoo2
  106.       - zoo3
  108.   kafka3:
  109.     container_name: kafka3
  110.     hostname: kafka3
  111.     image: bitnami/kafka:2.8.1
  112.     restart: always
  113.     ports:
  114.       - '9093:9093'
  115.     volumes:
  116.       - ./kafka3:/bitnami/kafka
  117.     environment:
  118.       - KAFKA_BROKER_ID=3
  119.       - ALLOW_PLAINTEXT_LISTENER=yes
  120.       - KAFKA_CFG_LISTENERS=PLAINTEXT://:9093
  121.       - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://192.168.3.77:9093
  122.       - KAFKA_CFG_ZOOKEEPER_CONNECT=zoo1:2181,zoo2:2182,zoo3:2183
  123.       - KAFKA_HEAP_OPTS=-Xmx512m -Xms512m
  124.     networks:
  125.       - app-tier
  126.     depends_on:
  127.       - zoo1
  128.       - zoo2
  129.       - zoo3
  131.   kafdrop:
  132.     container_name: kafdrop
  133.     hostname: kafdrop
  134.     image: obsidiandynamics/kafdrop:3.27.0
  135.     restart: always
  136.     ports:
  137.       - 9094:9000
  138.     environment:
  139.       - KAFKA_BROKERCONNECT=kafka1:9091,kafka2:9092,kafka3:9093
  140.       - JVM_OPTS=-Xms32m -Xmx64m
  141.       - SERVER_SERVLET_CONTEXTPATH=/
  142.     networks:
  143.       - app-tier
  144.     depends_on:
  145.       - kafka1
  146.       - kafka2
  147.       - kafka3

3. 以上准备工作都做好后，开始实操。

3.1 启动Kafka。

  1. docker-compose up

3.2 kafdrop - Kafka可视化UI Web。

docker-compose.yml的最后1个引用就是kafdrop。
打开浏览器并访问：http://192.168.3.77:9094/
image.png

3.3 Offset Explorer - Kafka 的终极 UI 工具。

The Ultimate UI Tool for Kafka. 下载地址详见附录。
image.png

4. Spring项目集成。

Topic 主题
Partition 分区
Kafka中每个主题可以有多个分区，每个分区只能有一个消费者，保证消息消费是有序的。
如果需要将消息发送到指定分区，可以单独定义分发规则，自定义方法实现 Partitioner 接口。

4.1 pom.xml 依赖引用

  1. <kafka.version>2.4.1</kafka.version>
  2. <jackson.version>2.12.5</jackson.version>
  4. <dependency>
  5.   <groupId>org.springframework.kafka</groupId>
  6.   <artifactId>spring-kafka</artifactId>
  7.   <exclusions>
  8.     <exclusion>
  9.       <artifactId>kafka-clients</artifactId>
  10.       <groupId>org.apache.kafka</groupId>
  11.     </exclusion>
  12.   </exclusions>
  13. </dependency>
  14. <dependency>
  15.   <groupId>org.apache.kafka</groupId>
  16.   <artifactId>kafka-clients</artifactId>
  17.   <version>${kafka.version}</version>
  18. </dependency>
  20. <dependency>
  21.   <groupId>com.fasterxml.jackson.core</groupId>
  22.   <artifactId>jackson-databind</artifactId>
  23.   <version>${jackson.version}</version>
  24. </dependency>

4.2 自定义Kafka解析器

4.2.1 自定义对象序列化工具 KafkaJsonSerializer

  1. import com.fasterxml.jackson.databind.ObjectMapper;
  2. import lombok.SneakyThrows;
  3. import org.apache.kafka.common.header.Headers;
  4. import org.apache.kafka.common.serialization.Serializer;
  6. import java.util.Map;
  8. /**
  9.  * 自定义对象序列化工具
  10.  *
  11.  * @author menglt
  12.  * @since 2021/11/20 15:52
  13.  */
  14. public class KafkaJsonSerializer implements Serializer<Object> {
  16.   @Override
  17.   @SneakyThrows
  18.   public byte[] serialize(String s, Object o) {
  19.     ObjectMapper objectMapper = new ObjectMapper();
  20.     return objectMapper.writeValueAsString(o).getBytes();
  21.   }
  23.   @Override
  24.   public void configure(Map<String, ?> configs, boolean isKey) {
  25.     Serializer.super.configure(configs, isKey);
  26.   }
  28.   @Override
  29.   public byte[] serialize(String topic, Headers headers, Object data) {
  30.     return Serializer.super.serialize(topic, headers, data);
  31.   }
  33.   @Override
  34.   public void close() {
  35.     Serializer.super.close();
  36.   }
  37. }

4.2.2 自定义对象反序列化工具 KafkaJsonDeserializer

  1. import com.fasterxml.jackson.databind.JsonNode;
  2. import com.fasterxml.jackson.databind.ObjectMapper;
  3. import lombok.SneakyThrows;
  4. import org.apache.kafka.common.header.Headers;
  5. import org.apache.kafka.common.serialization.Deserializer;
  7. import java.util.Map;
  9. /**
  10.  * 自定义对象反序列化工具
  11.  *
  12.  * @author menglt
  13.  * @since 2021/11/20 15:52
  14.  */
  15. public class KafkaJsonDeserializer implements Deserializer<JsonNode> {
  17.   @Override
  18.   @SneakyThrows
  19.   public JsonNode deserialize(String s, byte[] bytes) {
  20.     ObjectMapper objectMapper = new ObjectMapper();
  21.     return objectMapper.readTree(bytes);
  22.   }
  24.   @Override
  25.   public void configure(Map<String, ?> configs, boolean isKey) {
  26.     Deserializer.super.configure(configs, isKey);
  27.   }
  29.   @Override
  30.   public JsonNode deserialize(String topic, Headers headers, byte[] data) {
  31.     return Deserializer.super.deserialize(topic, headers, data);
  32.   }
  34.   @Override
  35.   public void close() {
  36.     Deserializer.super.close();
  37.   }
  38. }

4.3 application.properties

4.3.1 生产者 producer

  1. # kafka
  2. spring.kafka.bootstrap-servers=192.168.3.77:9091,192.168.3.77:9092,192.168.3.77:9093
  3. spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
  4. spring.kafka.producer.value-serializer=com.your.project.kafka.KafkaJsonSerializer

4.3.2 消费者 consumer

  1. # kafka
  2. spring.kafka.bootstrap-servers=192.168.3.77:9091,192.168.3.77:9092,192.168.3.77:9093
  3. spring.kafka.consumer.group-id=your-project-group-id
  4. spring.kafka.consumer.auto-offset-reset=earliest
  5. spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
  6. spring.kafka.consumer.value-deserializer=com.your.project.kafka.KafkaJsonDeserializer
  7. spring.kafka.listener.missing-topics-fatal=false

4.4 SpringBootApplication

  1. // 启动kafka
  2. @EnableKafka
  3. @SpringBootApplication
  4. public class YourProjectApplication {
  6.   public static void main(String[] args) {
  7.     SpringApplication springApplication = new SpringApplication(YourProjectApplication.class);
  8.     springApplication.run(args);
  9.   }
  11.   // 如果没有创建好的topic，就初始化1个
  12.   @Bean
  13.   public NewTopic topic() {
  14.     return TopicBuilder.name("your-project-log-topic").partitions(10).replicas(1).build();
  15.   }
  16. }

4.5 KafkaTemplate

  1. import org.springframework.kafka.core.KafkaTemplate;
  2. import org.springframework.kafka.support.SendResult;
  3. import org.springframework.util.concurrent.ListenableFuture;
  5. @Autowired private KafkaTemplate kafkaTemplate;
  7. YourProjectLog yourProjectLog = new YourProjectLog();
  8. ListenableFuture<SendResult> listenableFuture = kafkaTemplate.send("your-project-log-topic", yourProjectLog);

4.6 @KafkaListener

  1. @KafkaListener(topics = "your-project-log-topic", groupId = "your-project-group-id", topicPartitions = {})
  2. public void processMessage(String content) {
  3.     log.info("receiving message: " + content);
  4. }

5. 附录。

5.1 Offset Explorer 下载地址。

5.2 Kafka 文档。

5.3 Logstash-Plugin 集成Kafka方案。

5.4 Spring for Apache Kafka。

5.5 spring-kafka 更多配置参数。

[

](http://kafka.apache.org/28/documentation.html)