kafka的安装和配置

墨蓝 2022-04-05 05:11 215阅读 0赞

下载kafka

进入kafka官网：[http://kafka.apache.org/][http_kafka.apache.org] 。点击QUICKSTART

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjE4NTEw_size_16_color_FFFFFF_t_70][]

进去之后点击Download

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjE4NTEw_size_16_color_FFFFFF_t_70 1][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjE4NTEw_size_16_color_FFFFFF_t_70 2][]

复制链接地址到CentOS,执行

wget [http://mirrors.shu.edu.cn/apache/kafka/2.1.0/kafka\_2.11-2.1.0.tgz][http_mirrors.shu.edu.cn_apache_kafka_2.1.0_kafka_2.11-2.1.0.tgz]

再解压

tar -xzvf kafka\_2.11-2.1.0.tgz

运行kafka需要使用Zookeeper，所以你需要先启动Zookeeper，如果你没有Zookeeper，你可以使用kafka自带打包和配置好的Zookeeper。

(zookeeper安装：[https://blog.csdn.net/qq\_41618510/article/details/84946713）][https_blog.csdn.net_qq_41618510_article_details_84946713]

运行命令：bin/zookeeper-server-start.sh config/zookeeper.properties

### 启动kafka服务 ###

命令：bin/kafka-server-start.sh config/server.properties &

## 创建一个主题(topic) ##

创建一个名为“test”的Topic，只有一个分区和一个备份：

> bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

创建好之后，可以通过运行以下命令，查看已创建的topic信息：

> bin/kafka-topics.sh --list --zookeeper localhost:2181
    test

或者，除了手工创建topic外，你也可以配置你的broker，当发布一个不存在的topic时自动创建topic。

##  发送消息 ##

Kafka提供了一个命令行的工具，可以从输入文件或者命令行中读取消息并发送给Kafka集群。每一行是一条消息。  
运行producer（生产者）,然后在控制台输入几条消息到服务器。

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
    This is a message
    This is another message

## 消费消息 ##

Kafka也提供了一个消费消息的命令行工具，将存储的信息输出出来。

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
    This is a message
    This is another message

如果你有2台不同的终端上运行上述命令，那么当你在运行生产者时，消费者就能消费到生产者发送的消息。

## 设置多个broker集群 ##

到目前，我们只是单一的运行一个broker，没什么意思。对于Kafka，一个broker仅仅只是一个集群的大小，所有让我们多设几个broker。

首先为每个broker创建一个配置文件:

> cp config/server.properties config/server-1.properties 
    > cp config/server.properties config/server-2.properties

现在编辑这些新建的文件，设置以下属性：

config/server-1.properties: 
        broker.id=1 
        listeners=PLAINTEXT://:9093 
        log.dir=/tmp/kafka-logs-1
    
    config/server-2.properties: 
        broker.id=2 
        listeners=PLAINTEXT://:9094 
        log.dir=/tmp/kafka-logs-2

`broker.id`是集群中每个节点的唯一且永久的名称，我们修改端口和日志目录是因为我们现在在同一台机器上运行，我们要防止broker在同一端口上注册和覆盖对方的数据。

我们已经运行了zookeeper和刚才的一个kafka节点，所有我们只需要在启动2个新的kafka节点。

> bin/kafka-server-start.sh config/server-1.properties &
    ... 
    > bin/kafka-server-start.sh config/server-2.properties &
    ...

现在，我们创建一个新topic，把备份设置为：3

> bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic

好了，现在我们已经有了一个集群了，我们怎么知道每个集群在做什么呢？运行命令“describe topics”

> bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic
    Topic:my-replicated-topic    PartitionCount:1    ReplicationFactor:3    Configs:
    Topic: my-replicated-topic    Partition: 0    Leader: 1    Replicas: 1,2,0    Isr: 1,2,0

输出解释：第一行是所有分区的摘要，其次，每一行提供一个分区信息，因为我们只有一个分区，所以只有一行。

*  "leader"：该节点负责该分区的所有的读和写，每个节点的`leader`都是随机选择的。
 *  "replicas"：备份的节点列表，无论该节点是否是leader或者目前是否还活着，只是显示。
 *  "isr"：“同步备份”的节点列表，也就是活着的节点并且正在同步leader。

我们运行这个命令，看看一开始我们创建的那个节点：

> bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test
    Topic:test    PartitionCount:1    ReplicationFactor:1    Configs:
    Topic: test    Partition: 0    Leader: 0    Replicas: 0    Isr: 0

这并不奇怪，刚才创建的主题没有Replicas，并且在服务器“0”上，我们创建它的时候，集群中只有一个服务器，所以是“0”。

让我们来发布一些信息在新的topic上：

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic
     ...
    my test message 1
    my test message 2
    ^C

现在，消费这些消息。

> bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic my-replicated-topic
     ...
    my test message 1
    my test message 2
    ^C

我们要测试集群的容错，kill掉leader，Broker1作为当前的leader，也就是kill掉Broker1。

> ps | grep server-1.properties
    7564 ttys002    0:15.91 /System/Library/Frameworks/JavaVM.framework/Versions/1.6/Home/bin/java... 
    > kill -9 7564

备份节点之一成为新的leader，而broker1已经不在同步备份集合里了。

但是，消息仍然没丢：

> bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic my-replicated-topic
    ...
    my test message 1
    my test message 2
    ^C

原文参考：orchome.com/6

[http_kafka.apache.org]: http://kafka.apache.org/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjE4NTEw_size_16_color_FFFFFF_t_70]: /images/20220405/d7969e018e5f436fbb2dac8e74efe93a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjE4NTEw_size_16_color_FFFFFF_t_70 1]: /images/20220405/d2bce4ab3f5d4806ab312644c02def49.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjE4NTEw_size_16_color_FFFFFF_t_70 2]: /images/20220405/f517c775bdca473595806de33d7f9c23.png
[http_mirrors.shu.edu.cn_apache_kafka_2.1.0_kafka_2.11-2.1.0.tgz]: http://mirrors.shu.edu.cn/apache/kafka/2.1.0/kafka_2.11-2.1.0.tgz
[https_blog.csdn.net_qq_41618510_article_details_84946713]: https://blog.csdn.net/qq_41618510/article/details/84946713