kafka极简入门(三)--创建topic

Posted 2023-03-31

参考技术A

回顾 kafka极简入门(二)--安装

topic是kafka的生产者和消费者最小交互的单位，我们先从topic入手，创建第一个topic.

或

所以执行上面命令将会创建一个名为mytest的topic，该topic下面有1个分区,并且该分区只有1个副本。

PS:除了手动创建主题外，还可以将代理配置为在发布不存在的主题时自动创建主题

Partition:0 表示该分区的id为0
leader: 9 表示分区的首领副本所在的broker（本例子中broker.id配置为9,所以这里显示9,具体在config/server.properties配置。这里只有一个分区，所以首领分区也就是自己）
Replicas: 9 表示分区的跟随副本所在的broker
Isr: 9 表示分区的同步副本所在的broker(同步副本可以认为跟首领副本准实时同步的副本，可以配置判断条件，后面会讲，首领副本挂掉后，服务器会从同步副本中选举新的首领)

发送三个消息，分别是hello, world和！

注意: --from-beginning 表示从最开始的offset处开始消费。如果不写表示从最新的offset处消费，那么先发送了消息再开启消费者是收不到已发送的信息的

Kafka安装（极简版）

Kafka简介

kafka是一个高吞吐量的分布式消息队列，具有高性能、持久化、多副本备份、横向扩展能力，通常用于大数据及流处理平台。消息队列里都有生产者/消费者的概念，生产者往队列里写消息，而消费者则是从队列里获取消息。一般在架构设计中起到解耦、削峰、异步处理的作用。

kafka对外使用topic的概念，生产者往topic里写消息，消费者则从topic里读消息。为了做到水平扩展，一个topic实际是由多个partition组成的，遇到瓶颈时，可以通过增加partition的数量来进行横向扩容。在单个parition内是保证消息有序。每新写一条消息，kafka就是在对应的文件append写，所以性能非常高。

kafka的总体数据流是这样的：

大概用法就是，Producers往Brokers里面的指定Topic中写消息，Consumers从Brokers里面拉去指定Topic的消息，然后进行业务处理。

图中有两个topic，topic 0有两个partition，topic 1有一个partition，三副本备份。可以看到consumer gourp 1中的consumer 2没有分到partition处理，这是有可能出现的。

kafka需要依赖zookeeper存储一些元信息，而kafka也自带了zookeeper。其中broker、topics、partitions的一些元信息用zookeeper来存储，监控和路由啥的也都会用到zookeeper。

kafka名词解释：

• producer：生产者。
• consumer：消费者。
• topic: 消息以topic为类别记录，Kafka将消息种子(Feed)分门别类，每一类的消息称之为一个主题(Topic)。
• partitions：每个Topics划分为一个或者多个partition，并且partition中的每条消息都被标记了一个sequential id ，也就是offset，并且存储的数据是可配置存储时间的
• broker：以集群的方式运行，可以由一个或多个服务组成，每个服务叫做一个broker，消费者可以订阅一个或多个主题(topic)，并从Broker拉数据，从而消费这些已发布的消息。

在kafka中每个消息（也叫作record记录，也被称为消息）通常是由一个key，一个value和时间戳构成。

kafka有四个核心的API：

• 应用程序使用Producer API发布消息到1个或多个topic中。
• 应用程序使用Consumer API来订阅一个或多个topic，并处理产生的消息。
• 应用程序使用Streams API充当一个流处理器，从1个或多个topic消费输入流,并产生一个输出流到1个或多个topic，有效地将输入流转换到输出流。
• Connector API允许构建或运行可重复使用的生产者或消费者，将topic链接到现有的应用程序或数据系统。

kafka就先介绍到这，网络上有很多相关的理论文章，所以这里不过多赘述了，也可以直接查看官方文档。官方文档地址如下：

http://kafka.apache.org/intro.html

---

单实例安装

本小节我们来在CentOS7上安装Kafka，由于kafka是由Scala和Java语言编写的，所以前提需要准备好java运行环境，我这里java环境是1.8的，由于jdk的安装配置都比较简单，这里就不演示jdk的安装过程了，直接安装Kafka。

到官网上复制下载地址，使用wget命令进行下载并解压：

[[email protected] ~]# cd /usr/local/src/
[[email protected] /usr/local/src]# wget http://mirrors.hust.edu.cn/apache/kafka/2.0.0/kafka_2.11-2.0.0.tgz
[[email protected] /usr/local/src]# tar -zxvf kafka_2.11-2.0.0.tgz
[[email protected] /usr/local/src]# mv kafka_2.11-2.0.0 /usr/local/kafka
[[email protected] /usr/local/src]# cd !$

没有特殊要求的话，我们使用默认的kafka配置即可。若你希望kafka能够被外部机器访问，则需要配置一下你机器的外网ip地址及端口，如下：

[[email protected] /usr/local/kafka]# vim ./config/server.properties
listeners=PLAINTEXT://192.168.190.129:9092
advertised.listeners=PLAINTEXT://192.168.190.129:9092
[[email protected] /usr/local/kafka]# 

现在我们就可以使用kafka了，由于kafka依赖zookeeper，所以我们在启动kafka前需要先启动kafka自带的zookeeper服务：

[[email protected] /usr/local/kafka]# nohup ./bin/zookeeper-server-start.sh ./config/zookeeper.properties > zookeeper.out &

zookeeper服务启动成功后，启动kafka：

[[email protected] /usr/local/kafka]# nohup ./bin/kafka-server-start.sh ./config/server.properties > kafka.out &

两个服务都启动成功后，监听的端口如下：

[[email protected] ~]# netstat -lntp |grep java
tcp6       0      0 :::38031                :::*                    LISTEN      3629/java
tcp6       0      0 :::33620                :::*                    LISTEN      3945/java
tcp6       0      0 :::9092                 :::*                    LISTEN      3945/java
tcp6       0      0 :::2181                 :::*                    LISTEN      3629/java
[[email protected] ~]# 

---

接下来我们测试一下kafka是否正常可用，首先创建一个topic，命令如下：

[[email protected] /usr/local/kafka]# ./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic hello
Created topic "hello".
[[email protected] /usr/local/kafka]# 

测试是否能获取topic列表：

[[email protected] /usr/local/kafka]# ./bin/kafka-topics.sh --list --zookeeper localhost:2181
hello
[[email protected] /usr/local/kafka]# 

测试往一个topic上发送消息：

[[email protected] /usr/local/kafka]# ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic hello>hello world
>hello kafka 
>

测试从一个topic上消费消息：

[[email protected] /usr/local/kafka]# ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic hello --from-beginning
hello world
hello kafka

通过以上测试，可以看到，kafka能够正常的创建topic进行发送/接收消息，那么就代表我们安装成功了。