mac 本地 docker 安装 kafka

方案一：传统模式（需Zookeeper）
‌ 1. 拉取镜像‌

docker pull wurstmeister/zookeeper  # Zookeeper镜像‌:ml-citation{ref="3" data="citationList"}
docker pull wurstmeister/kafka      # Kafka镜像‌:ml-citation{ref="3" data="citationList"}

2. 启动 zookeeper

docker run -d --name zookeeper \
-p 2181:2181 \
-v /etc/localtime:/etc/localtime \
wurstmeister/zookeeper

3. 启动 kafka

docker run -d --name kafka \
-p 9092:9092 \
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
-e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 \
-e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 \
--link zookeeper \
wurstmeister/kafka

4.创建队列

注意 容器里面 kafka 的启动脚本的路径可能存在差异。需要具体找到此路径

/opt/kafka_2.13-2.8.1/bin/kafka-topics.sh

docker exec -it kafka bash
/opt/kafka_2.13-2.8.1/bin/kafka-topics.sh --create \
--bootstrap-server localhost:9092 \
--replication-factor 1 \
--partitions 1 \
--topic test-topic

5.查看队列

/opt/kafka_2.13-2.8.1/bin/kafka-topics.sh --list \
--bootstrap-server localhost:9092

快速Java 代码验证是否创建成功和链接成功

maven

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

生产者

import org.apache.kafka.clients.producer.*;public class KafkaProducerDemo {public static void main(String[] args) {Properties props = new Properties();props.put("bootstrap.servers", "localhost:9092");props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");Producer<String, String> producer = new KafkaProducer<>(props);ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", "key", "Hello Kafka");producer.send(record);producer.close();}
}

消费者

import org.apache.kafka.clients.consumer.*;public class KafkaConsumerDemo {public static void main(String[] args) {Properties props = new Properties();props.put("bootstrap.servers", "localhost:9092");props.put("group.id", "test-group");props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");Consumer<String, String> consumer = new KafkaConsumer<>(props);consumer.subscribe(Collections.singletonList("test-topic"));while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));for (ConsumerRecord<String, String> record : records) {System.out.printf("Received: key=%s, value=%s%n", record.key(), record.value());}}}
}

继承到 springboot 敬请期待后续