关于kafka:kafka-系列-2搭建与实践

前言

入手实际往往比看看更重要????

单机版 Docker 搭建

version: '2' 
services:
  zookeeper:
    image: wurstmeister/zookeeper
    ports:
      - "2181:2181"
  kafka:
    image: wurstmeister/kafka
    depends_on: [zookeeper]
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: kafka
      KAFKA_ADVERTISED_PORT: 9092
      KAFKA_CREATE_TOPICS: "test:1:1"
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"

注意事项：

• 如果想要 java 客户端可能失常连贯上 kafka, 须要配置宿主机的 host

sudo vim /etc/hosts

172.20.10.6 kafka

• 如何应用 kafka 自带的 kafka-console-producer 测试发送音讯？

kafka-console-producer.sh --bootstrap-server kafka:9092 --topic test

集群版 + kafka manager

kafka 集群 docker-compose

version: '2'
services:
  zookeeper:
    image: wurstmeister/zookeeper
    ports:
      - "2181:2181"
  kafka1:
    restart: always
    image: wurstmeister/kafka
    depends_on: [zookeeper]
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://kafka1:9092"
      KAFKA_LISTENERS: "PLAINTEXT://kafka1:9092"
      KAFKA_PORT: 9092
  kafka2:
    restart: always
    image: wurstmeister/kafka
    depends_on: [zookeeper]
    ports:
      - "9093:9093"
    environment:
      KAFKA_BROKER_ID: 2
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://kafka2:9093"
      KAFKA_LISTENERS: "PLAINTEXT://kafka2:9093"
      KAFKA_PORT: 9093
  kafka3:
    restart: always
    image: wurstmeister/kafka
    depends_on: [zookeeper]
    ports:
      - "9094:9094"
    environment:
      KAFKA_BROKER_ID: 3
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://kafka3:9094"
      KAFKA_LISTENERS: "PLAINTEXT://kafka3:9094"
      KAFKA_PORT: 9094

注意事项：

• 如果想要 java 客户端可能失常连贯上 kafka, 须要配置宿主机的 host

sudo vim /etc/hosts

172.20.10.6 kafka1
172.20.10.6 kafka2
172.20.10.6 kafka3

• 如何应用 kafka 自带的 kafka-console-producer 测试发送音讯？这里假如是进入到 kafka3 容器中

kafka-topics.sh --create --zookeeper zookeeper:2181 --replication-factor 3 --partitions 3 --topic test2

kafka-console-producer.sh --bootstrap-server kafka3:9094 --topic test

kafka-manager docker-compose

version: "2"
services:
  kafka-manager:
    image: kafkamanager/kafka-manager
    container_name: kafka-manager
    ports:
      - "9000:9000"
    external_links:  # 连贯本 compose 文件以外的 container
      - kafka_kafka1_1
      - kafka_kafka2_1
      - kafka_kafka3_1
    environment:
      ZK_HOSTS: kafka_zookeeper_1:2181
networks:
  default:
    external:
      name: kafka_default

注意事项
kafka-manager、与 kafka 集群不在同一个 compose 中。因而这里须要应用 networks 连贯到 kafka 集群的网络中

基本操作

以下均在 docker 内操作

cd /opt/kafka/bin

1. 创立主题

创立了 1 个正本 1 个分区的主题

kafka-topics.sh --create --zookeeper zookeeper:2181 --replication-factor 1 --partitions 1 --topic test2

1. 查看主题

kafka-topics.sh  --zookeeper zookeeper:2181 --list

1. 查看主题详情

kafka-topics.sh  --zookeeper zookeeper:2181  --describe  --topic test2

1. 发送音讯

kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test2

1. 生产音讯

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test2 --from-beginning

springboot 连贯 kafka

maven

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.1.10.RELEASE</version>
</parent>
<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>

</dependencies>

yaml

server:
  port: 9009
spring:

  kafka:
    bootstrap-servers: 127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094

    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

    consumer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      group-id: goup1 # 生产组 

生产者

@Autowired
private KafkaTemplate<String,Object> kafkaTemplate;
kafkaTemplate.send("test2","qweqwe");

消费者

@KafkaListener(topics = "test2")
public void onMsg(String msg) {log.error("kafka {}" ,msg);
    System.out.println(msg);
}

相干文档

• Kafka 官网配置参数
• kafka 中文文档 broker 配置参数