Kafka安装与使用

素颜马尾好姑娘i 2023-07-15 11:29 5阅读 0赞

### 文章目录 ###

*  安装并启动
 *  集群环境
 *  发出你的第一条kafka消息
 *   *  1、创建topic
     *  2、启动生产者以发送消息
     *  3、启动消费者以接收消息

# 安装并启动 #

官方建议使用JDK 1.8版本，因此本文使用的环境都是JDK1.8。关于JDK的安装，本文不再详述，默认Java环境已经具备。

由于Kafka依赖Zookeeper，kafka通过Zookeeper实现分布式系统的协调，用来保存集群的元数据信息和消费者信息，所以我们需要先安装Zookeeper。(Kafka发行版自带了 Zookeeper，可以直接从脚本启动，不过安装一个完整版的Zookeeper也并不费劲)  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70]  
接下来我们按照如下步骤，一步步来安装kafka:

1.  下载zookeeper，解压  
    下载地址:[https://zookeeper.apache.org/releases.html\#download][https_zookeeper.apache.org_releases.html_download]
2.  创建zookeeper配置文件  
    在zookeeper解压后的目录下找到conf文件夹，进入后，复制文件zoo\_sample.cfg，并命 名为zoo.cfg  
    zoo.cfg中一共四个配置项，可以使用默认配置。
    
        $ vi conf/zoo.cfg
        # The number of milliseconds of each tick
        tickTime=2000
        # The number of ticks that the initial
        # synchronization phase can take
        initLimit=10
        # The number of ticks that can pass between
        # sending a request and getting an acknowledgement
        syncLimit=5
        # the directory where the snapshot is stored.
        # do not use /tmp for storage, /tmp here is just
        # example sakes.
        dataDir=/tmp/zookeeper
        # the port at which the clients will connect
        clientPort=2181
3.  启动zookeeper。  
    进入zookeeper根目录执行 `bin/zkServer.sh start`
    
        [root@izuf6bsiti3f4kuichki8nz zookeeper]# bin/zkServer.sh start
        /usr/bin/java
        ZooKeeper JMX enabled by default
        Using config: /root/software/zookeeper/zookeeper/bin/../conf/zoo.cfg
        Starting zookeeper ... STARTED
    
    如上，zk启动成功。
4.  下载kafka，解压。  
    kafka 2.4版本下载地址:  
    [https://www.apache.org/dyn/closer.cgi?path=/kafka/2.4.0/kafka\_2.12-2.4.0.tgz][https_www.apache.org_dyn_closer.cgi_path_kafka_2.4.0_kafka_2.12-2.4.0.tgz]
5.  修改kafka的配置文件。  
    进入kafka根目录下的config文件夹下，打开server.properties,修改如下配置项
    
        zookeeper.connect=localhost:2181 
        broker.id=0
        log.dirs=/tmp/kafka‐logs
    
    zookeeper.connect是zookeeper的链接信息，broker.id是当前kafka实例的id，log.dirs 是kafka存储消息内容的路径。
6.  启动kafka  
    进入kafka根目录执行 `bin/kafka-server-start.sh config/server.properties`  
    此命令告诉kafka启动时使用config/server.properties配置项  
    注意这里报错如下
    
        [root@izuf6bsiti3f4kuichki8nz kafka]# bin/kafka-server-start.sh config/server.properties 
        Java HotSpot(TM) 64-Bit Server VM warning: INFO: os::commit_memory(0x00000000c0000000, 1073741824, 0) failed; error='Cannot allocate memory' (errno=12)
        #
        # There is insufficient memory for the Java Runtime Environment to continue.
        # Native memory allocation (mmap) failed to map 1073741824 bytes for committing reserved memory.
        # An error report file with more information is saved as:
        # /root/software/kafka/kafka/hs_err_pid1079.log
    
    可以发现是因为系统内存过小无法完成需要的分配，修改bin/kafka-server-start.sh里面启动分配的内存大小：
    
        vim bin/kafka-server-start.sh
    
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70 1]  
    这里我们改为512M 和 256M。  
    （后因为zookeeper版本有问题我们又换了下zookeeper 版本 在启动后执行zkCli.sh有响应就好了。）  
    再次启动，启动成功。

如何确定kafka是否启动成功了呢？可以按照下面的步骤：

1.  启动zookeeper的client  
    进入zookeeper根目录下，执行 `bin/zkCli.sh -server 127.0.0.1:2181`。启动成功后如下图
    
        Connecting to local host:2181 
        ................
        ................ 
        ................ 
        Welcome to ZooKeeper! 
        ................
        ................
        WATCHER::
        WatchedEvent state:SyncConnected type:None path:null 10 	
        [zk: localhost:2181(CONNECTED) 0]
2.  输入命令 `ls /brokers`，回车，可以看到子节点信息
    
        [zk: 127.0.0.1:2181(CONNECTED) 0] ls /brokers
        [ids, topics, seqid]
    
    这些子节点存储的就是kafka集群管理的数据。broker是kafka的一个服务单元实例
3.  输入命令 `ls /brokers/ids`，可以看到如下信息:
    
        [zk: 127.0.0.1:2181(CONNECTED) 1] ls /brokers/ids
        [0]
    
    还记得我们在配置单机环境时，修改的kafka配置项broker.id=0 吗?这里的0就是表示那个  
    kafka的实例已经加入了kafka集群。

# 集群环境 #

集群环境的搭建也很简单，在单机环境的基础上，让多个单机连接到同一个zookeeper即可。需要注意两点:

*  1、每个实例设置不同的broker.id。
 *  2、如果多个实例部署在同一台服务器，还要注意修改log.dirs为不同目录，确保消息存储时不会有冲突。

# 发出你的第一条kafka消息 #

我们通过kafka带的工具来创建一个topic，然后尝试发送和消费一个消息，直观的去感受下 kafka。

## 1、创建topic ##

Kafka提供了一个名为 kafka-topics.sh 的命令行实用程序，用于在服务器上创建主题。 打开新终端并键入以下示例。

进入kafka根目录，执行如下命令:  
`bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic Hello-Kafka`

[root@izuf6bsiti3f4kuichki8nz kafka]# bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic Hello-Kafka
    Created topic Hello-Kafka.

我们刚刚创建了一个名为 Hello-Kafka 的主题，其中包含一个分区和一个副本因子。

创建主题后，我们可以在Kafka代理终端窗口中获取通知，并在`config/server.properties`文件中的`/tmp/kafka-logs/`中指定的创建主题的日志。

**主题列表**  
要获取Kafka服务器中的主题列表，可以使用以下命令  
`bin/kafka-topics.sh --list --zookeeper localhost:2181`

[root@izuf6bsiti3f4kuichki8nz kafka]# bin/kafka-topics.sh --list --zookeeper localhost:2181
    Hello-Kafka

由于我们已经创建了一个主题，它将仅列出 Hello-Kafka 。 假设，如果创建多个主题，您 将在输出中获取主题名称。

## 2、启动生产者以发送消息 ##

语法：  
`bin/kafka‐console‐producer.sh --broker‐list localhost:9092 --topic topic‐name`  
从上面的语法，生产者命令行客户端需要两个主要参数

*  代理列表。 在这种情况下，我们只有一个代理。 `config/server.properties`文件包含代理端口ID，这里就是`localhost:9092`。
 *  主题名称。

示例：  
`bin/kafka‐console‐producer.sh --broker‐list localhost:9092 --topic Hello‐ Kafka`

生产者将等待来自stdin的输入并发布到Kafka集群。 默认情况下，每个新行都作为新消息发布，然后在 `config/producer.properties` 文件中指定默认生产者属性。现在，我们可以在终端中键入几行消息，如下所示。

[root@izuf6bsiti3f4kuichki8nz kafka]# bin/kafka‐console‐producer.sh --broker‐list localhost:9092 ‐‐topic Hello‐Kafka
    [2020‐01‐1613:50:45,931]
    WARN property topic is not valid (kafka.utils.Verifiable Properties)
    Hello
    Myfirstmessage
    Mysecondmessage

## 3、启动消费者以接收消息 ##

与生产者类似，在 `config/consumer.properties` 文件中指定了缺省使用者属性。 打开一个新终端并键入以下消息消息语法。  
语法：  
`bin/kafka‐console‐consumer.sh --zookeeper localhost:2181 --topic topic‐name --from‐beginning`  
示例：

[root@izuf6bsiti3f4kuichki8nz kafka]# bin/kafka‐console‐consumer.sh --zookeeper localhost:2181 —-topic Hello‐Kafka --from‐beginning
    Hello
    Myfirstmessage 
    Mysecondmessage

至此我们走完了一个发送消息的流程，可以看到我们经历了创建topic、启动生产者、消费者、生产者生产消息、消费者消费消息，这几个步骤。

可能会报错`zookeeper is not a recognized option`，换为使用broker信息消费即可。

[root@izuf6bsiti3f4kuichki8nz kafka]# ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first-topic
    111
    222
    333

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70]: /images/20230528/50be8cb9ec9e43d193e648ebc30c3e87.png
[https_zookeeper.apache.org_releases.html_download]: https://zookeeper.apache.org/releases.html#download
[https_www.apache.org_dyn_closer.cgi_path_kafka_2.4.0_kafka_2.12-2.4.0.tgz]: https://www.apache.org/dyn/closer.cgi?path=/kafka/2.4.0/kafka_2.12-2.4.0.tgz
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70 1]: /images/20230528/9520c99edd8e40f58e2f4664f4c34d85.png