5分钟带你体验一把 Kafka

说在文章前面的话：

前置条件：你的电脑已经安装 Docker

主要内容：

使用 Docker 安装

使用命令行测试消息队列的功能

zookeeper和kafka可视化管理工具

Java 程序中简单使用Kafka

使用 Docker 安装搭建Kafka环境

单机版

下面使用的单机版的Kafka 来作为演示，推荐先搭建单机版的Kafka来学习。

以下使用 Docker 搭建Kafka基本环境来自开源项目：github.com/simplesteph… 。当然，你也可以按照官方提供的来：github.com/wurstmeiste… 。

新建一个名为 zk-single-kafka-single.yml 的文件，文件内容如下：

version: '2.1'

services:

zoo1:

image: zookeeper:3.4.9

hostname: zoo1

ports:

- "2181:2181"

environment:

ZOO_MY_ID: 1

ZOO_PORT: 2181

ZOO_SERVERS: server.1=zoo1:2888:3888

volumes:

- ./zk-single-kafka-single/zoo1/data:/data

- ./zk-single-kafka-single/zoo1/datalog:/datalog

kafka1:

image: confluentinc/cp-kafka:5.3.1

hostname: kafka1

ports:

- "9092:9092"

environment:

KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka1:19092,LISTENER_DOCKER_EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9092

KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXT

KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNAL

KAFKA_ZOOKEEPER_CONNECT: "zoo1:2181"

KAFKA_BROKER_ID: 1

KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"

KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

volumes:

- ./zk-single-kafka-single/kafka1/data:/var/lib/kafka/data

depends_on:

- zoo1

运行以下命令即可完成环境搭建(会自动下载并运行一个 zookeeper 和 kafka )

docker-compose -f zk-single-kafka-single.yml up

如果需要停止Kafka相关容器的话，运行以下命令即可：

docker-compose -f zk-single-kafka-single.yml down

集群版

以下使用 Docker 搭建Kafka基本环境来自开源项目：github.com/simplesteph… 。

新建一个名为 zk-single-kafka-multiple.yml 的文件，文件内容如下：

version: '2.1'

services:

zoo1:

image: zookeeper:3.4.9

hostname: zoo1

ports:

- "2181:2181"

environment:

ZOO_MY_ID: 1

ZOO_PORT: 2181

ZOO_SERVERS: server.1=zoo1:2888:3888

volumes:

- ./zk-single-kafka-multiple/zoo1/data:/data

- ./zk-single-kafka-multiple/zoo1/datalog:/datalog

kafka1:

image: confluentinc/cp-kafka:5.4.0

hostname: kafka1

ports:

- "9092:9092"

environment:

KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka1:19092,LISTENER_DOCKER_EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9092

KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXT

KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNAL

KAFKA_ZOOKEEPER_CONNECT: "zoo1:2181"

KAFKA_BROKER_ID: 1

KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"

volumes:

- ./zk-single-kafka-multiple/kafka1/data:/var/lib/kafka/data

depends_on:

- zoo1

kafka2:

image: confluentinc/cp-kafka:5.4.0

hostname: kafka2

ports:

- "9093:9093"

environment:

KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka2:19093,LISTENER_DOCKER_EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9093

KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXT

KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNAL

KAFKA_ZOOKEEPER_CONNECT: "zoo1:2181"

KAFKA_BROKER_ID: 2

KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"

volumes:

- ./zk-single-kafka-multiple/kafka2/data:/var/lib/kafka/data

depends_on:

- zoo1

kafka3:

image: confluentinc/cp-kafka:5.4.0

hostname: kafka3

ports:

- "9094:9094"

environment:

KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka3:19094,LISTENER_DOCKER_EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9094

KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXT

KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNAL

KAFKA_ZOOKEEPER_CONNECT: "zoo1:2181"

KAFKA_BROKER_ID: 3

KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"

volumes:

- ./zk-single-kafka-multiple/kafka3/data:/var/lib/kafka/data

depends_on:

- zoo1

运行以下命令即可完成 1个节点 Zookeeper+3个节点的 Kafka 的环境搭建。

docker-compose -f zk-single-kafka-multiple.yml up

如果需要停止Kafka相关容器的话，运行以下命令即可：

docker-compose -f zk-single-kafka-multiple.yml down

使用命令行测试消息的生产和消费

一般情况下我们很少会用到 Kafka 的命令行操作。

1.进入 Kafka container 内部执行 Kafka 官方自带了一些命令

docker exec -ti docker_kafka1_1 bash

2.列出所有 Topic

root@kafka1:/# kafka-topics --describe --zookeeper zoo1:2181

3.创建一个 Topic

root@kafka1:/# kafka-topics --create --topic test --partitions 3 --zookeeper zoo1:2181 --replication-factor 1

Created topic test.

我们创建了一个名为 test 的 Topic, partition 数为 3, replica 数为 1。

4.消费者订阅主题

root@kafka1:/# kafka-console-consumer --bootstrap-server localhost:9092 --topic test

send hello from console -producer

我们订阅了名为 test 的 Topic。

5.生产者向 Topic 发送消息

root@kafka1:/# kafka-console-producer --broker-list localhost:9092 --topic test

>send hello from console -producer

>

我们使用 kafka-console-producer 命令向名为 test 的 Topic 发送了一条消息，消息内容为：“send hello from console -producer”

这个时候，你会发现消费者成功接收到了消息：

root@kafka1:/# kafka-console-consumer --bootstrap-server localhost:9092 --topic test

send hello from console -producer

IDEA相关插件推荐

Zoolytic-Zookeeper tool

这是一款 IDEA 提供的 Zookeeper 可视化工具插件，非常好用！我们可以通过它：

可视化ZkNodes节点信息

ZkNodes节点管理-添加/删除

编辑zkNodes数据

......

实际使用效果如下：

使用方法：

打开工具：View->Tool windows->Zoolytic；

点击 “+” 号后在弹出框数据：“127.0.0.1:2181” 连接 zookeeper；

连接之后点击新创建的连接然后点击“+”号旁边的刷新按钮即可！

Kafkalytic

IDEA 提供的 Kafka 可视化管理插件。这个插件为我们提供了下面这写功能：

多个集群支持

主题管理:创建/删除/更改分区

使用正则表达式搜索主题

发布字符串/字节序列化的消息

使用不同的策略消费消息

实际使用效果如下：

使用方法：

打开工具：View->Tool windows->kafkalytic；

点击 “+” 号后在弹出框数据：“127.0.0.1:9092” 连接；

Java 程序中简单使用Kafka

代码地址：github.com/Snailclimb/…

Step 1:新建一个Maven项目

Step2: pom.xml 中添加相关依赖

<dependency>

<groupId>org.apache.kafka</groupId>

<artifactId>kafka-clients</artifactId>

<version>2.2.0</version>

</dependency>

Step 3：初始化消费者和生产者

KafkaConstants常量类中定义了Kafka一些常用配置常量。

public class KafkaConstants {

public static final String BROKER_LIST = "localhost:9092";

public static final String CLIENT_ID = "client1";

public static String GROUP_ID_CONFIG="consumerGroup1";

private KafkaConstants() {

}

}

ProducerCreator 中有一个 createProducer() 方法方法用于返回一个 KafkaProducer对象

import org.apache.kafka.clients.producer.KafkaProducer;

import org.apache.kafka.clients.producer.Producer;

import org.apache.kafka.clients.producer.ProducerConfig;

import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**

* @author shuang.kou

*/

public class ProducerCreator {

public static Producer<String, String> createProducer() {

Properties properties = new Properties();

properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConstants.BROKER_LIST);

properties.put(ProducerConfig.CLIENT_ID_CONFIG, KafkaConstants.CLIENT_ID);

properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

return new KafkaProducer<>(properties);

}

}

ConsumerCreator 中有一个createConsumer() 方法方法用于返回一个 KafkaConsumer 对象

import org.apache.kafka.clients.consumer.Consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;

import org.apache.kafka.clients.consumer.KafkaConsumer;

import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Properties;

public class ConsumerCreator {

public static Consumer<String, String> createConsumer() {

Properties properties = new Properties();

properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConstants.BROKER_LIST);

properties.put(ConsumerConfig.GROUP_ID_CONFIG, KafkaConstants.GROUP_ID_CONFIG);

properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

return new KafkaConsumer<>(properties);

}

}

Step 4：发送和消费消息

生产者发送消息：

private static final String TOPIC = "test-topic";

Producer<String, String> producer = ProducerCreator.createProducer();

ProducerRecord<String, String> record =

new ProducerRecord<>(TOPIC, "hello, Kafka!");

try {

//send message

RecordMetadata metadata = producer.send(record).get();

System.out.println("Record sent to partition " + metadata.partition()

+ " with offset " + metadata.offset());

} catch (ExecutionException | InterruptedException e) {

System.out.println("Error in sending record");

e.printStackTrace();

}

producer.close();

消费者消费消息：

Consumer<String, String> consumer = ConsumerCreator.createConsumer();

// 循环消费消息

while (true) {

//subscribe topic and consume message

consumer.subscribe(Collections.singletonList(TOPIC));

ConsumerRecords<String, String> consumerRecords =

consumer.poll(Duration.ofMillis(1000));

for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {

System.out.println("Consumer consume message:" + consumerRecord.value());

}

}

Step 5：测试

运行程序控制台打印出：

Record sent to partition 0 with offset 20

Consumer consume message:hello, Kafka!

最后：

面都是自己整理好的！我就把资料贡献出来给有需要的人！顺便求一波关注，哈哈~各位小伙伴关注我后私信【Java】就可以免费领取哒

作者：SnailClimb

链接：https://juejin.im/post/5e589059e51d4527107c914d

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 215,874评论 6赞 498
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 92,102评论 3赞 391
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 161,676评论 0赞 351
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 57,911评论 1赞 290
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 66,937评论 6赞 388
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 50,935评论 1赞 295
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 39,860评论 3赞 416
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 38,660评论 0赞 271
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 45,113评论 1赞 308
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 37,363评论 2赞 331
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 39,506评论 1赞 346
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 35,238评论 5赞 341
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 40,861评论 3赞 325
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 31,486评论 0赞 21
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 32,674评论 1赞 268
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 47,513评论 2赞 368
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 44,426评论 2赞 352