Kafka安装及使用

博主： smalleyes
发布时间：2018 年 07 月 24 日
223次浏览
1 条评论
18177字数
分类：大数据

### **一、Kafka介绍**

**Apache开源。**

**Scala主要编程语言。**

**一种MQ。**

**详见官网：**[http://kafka.apache.org/](http://kafka.apache.org/)

**中文翻译:  **[http://kafka.apachecn.org/](http://kafka.apachecn.org/)

### **二、集群搭建**

#### **2.1.集群配置**

| **服务器**      | **配置** |
| ----------------- | ---------- |
| **192.168.0.3** | **2U8G** |
| **192.168.0.4** | **2U8G** |
| **192.168.0.5** | **2U8G** |

#### **2.2.集群准备**

* **JDK环境：本文当基于1.8**
* **SSH集群间免密**
* **zookeeper（可自行搭建，也可使用kafka自带的zk，本文档依赖于自行搭建的zk集群，服务器同上）**

#### **2.3.下载kafka安装包**

**下载地址：**[http://kafka.apache.org/downloads](http://kafka.apache.org/downloads)

**本文档记录下载版本：kafka_2.12-2.2.1.tgz**

**版本文件名称解释：**

* **2.1.2：scala版本**
* **2.2.1：kafka版本**

**解压安装包：**

```
tar -xvf kafka_2.12-2.2.1.tgz
```

**解压后的目录结构:**

****-bin：kafka命令文件夹****

****-config：kafka相关的配置文件夹****

**	****--consumer.properties：消费者相关配置，本文未涉及**

**	****--log4j.properties：log4j日志相关配置，本文未涉及**

**	****--producer.properties：生产者相关配置，本文未涉及**

**	******--server.properties：kafka核心配置文件****

**	****--zookeeper.properties：zookeeper相关配置，若使用kafka自带zookeeper，在此配置**

**-libs：依赖包**

**-LICENSE：APACHE开源许可证**

**-logs：日志文件夹**

**-NOTICE：notice**

**-site-docs：文档**

#### **2.4.修改配置文件**

**进入config目录下，修改server.properties文件，本文搭建时有如下属性涉及改动：**

```
#节点id,每个集群节点请设置唯一值
broker.id=0
#Socket服务器监听的地址，如果没有设置，则监听java.net.InetAddress.getCanonicalHostName()返回的地址
listeners=PLAINTEXT://192.168.0.4:9092
#broker通知到producers和consumers的主机地址和端口号。如果未设置，使用listeners的配置。
#否则，使用java.net.InetAddress#.getCanonicalHostName()返回的值。
#如果均为设置，拿到的值为localhot:9092,将造成无法连接其他机器,如需外网使用时,请设置为外网ip
advertised.listeners=PLAINTEXT://yourip:9092
#日志文件地址
log.dirs=/app/kafka_2.12-2.2.1/logs
#topic默认的partition数量
num.partitions=6
#topic默认的副本数量,本集群为3台,所以副本配置为2
default.replication.factor=2
#zookeeper地址
zookeeper.connect=192.168.0.3:2181,192.168.0.4:2181,192.168.0.5:2181
#zookeeper连接超时时间
zookeeper.connection.timeout.ms=12000
```

**以上配置为基础配置，设置后，集群已可正常启动运行，若涉及kafka调优的配置，请参考以下全部配置项，自行设置。**

```
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# see kafka.server.KafkaConfig for additional details and defaults

############################# Server Basics #############################

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=0

############################# Socket Server Settings #############################

# The address the socket server listens on. It will get the value returned from 
# java.net.InetAddress.getCanonicalHostName() if not configured.
#   FORMAT:
#     listeners = listener_name://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://192.168.0.4:9092

# Hostname and port the broker will advertise to producers and consumers. If not set, 
# it uses the value for "listeners" if configured.  Otherwise, it will use the value
# returned from java.net.InetAddress.getCanonicalHostName().
advertised.listeners=PLAINTEXT://114.67.92.113:9092

# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

# The number of threads that the server uses for receiving requests from the network and sending responses to the network
num.network.threads=3

# The number of threads that the server uses for processing requests, which may include disk I/O
num.io.threads=8

# The send buffer (SO_SNDBUF) used by the socket server
socket.send.buffer.bytes=102400

# The receive buffer (SO_RCVBUF) used by the socket server
socket.receive.buffer.bytes=102400

# The maximum size of a request that the socket server will accept (protection against OOM)
socket.request.max.bytes=104857600

############################# Log Basics #############################

# A comma separated list of directories under which to store log files
log.dirs=/app/kafka_2.12-2.2.1/logs

# The default number of log partitions per topic. More partitions allow greater
# parallelism for consumption, but this will also result in more files across
# the brokers.
num.partitions=6
default.replication.factor=2

# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
# This value is recommended to be increased for installations with data dirs located in RAID array.
num.recovery.threads.per.data.dir=1

############################# Internal Topic Settings  #############################
# The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"
# For anything other than development testing, a value greater than 1 is recommended for to ensure availability such as 3.
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

############################# Log Flush Policy #############################

# Messages are immediately written to the filesystem but by default we only fsync() to sync
# the OS cache lazily. The following configurations control the flush of data to disk.
# There are a few important trade-offs here:
#    1. Durability: Unflushed data may be lost if you are not using replication.
#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
#    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to excessive seeks.
# The settings below allow one to configure the flush policy to flush data after a period of time or
# every N messages (or both). This can be done globally and overridden on a per-topic basis.

# The number of messages to accept before forcing a flush of data to disk
#log.flush.interval.messages=10000

# The maximum amount of time a message can sit in a log before we force a flush
#log.flush.interval.ms=1000

############################# Log Retention Policy #############################

# The following configurations control the disposal of log segments. The policy can
# be set to delete segments after a period of time, or after a given size has accumulated.
# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
# from the end of the log.

# The minimum age of a log file to be eligible for deletion due to age
log.retention.hours=168

# A size-based retention policy for logs. Segments are pruned from the log unless the remaining
# segments drop below log.retention.bytes. Functions independently of log.retention.hours.
#log.retention.bytes=1073741824

# The maximum size of a log segment file. When this size is reached a new log segment will be created.
log.segment.bytes=1073741824

# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
log.retention.check.interval.ms=300000

############################# Zookeeper #############################

# Zookeeper connection string (see zookeeper docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
zookeeper.connect=hadoop01:2181,hadoop02:2181,hadoop03:2181

# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=12000

############################# Group Coordinator Settings #############################

# The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance.
# The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms.
# The default value for this is 3 seconds.
# We override this to 0 here as it makes for a better out-of-the-box experience for development and testing.
# However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup.
group.initial.rebalance.delay.ms=0
```

#### **2.5.启动集群**

**将kafka安装包scp至其他服务器，并修改以下参数，为对应服务器值：**

```
broker.id=1

listener=yourip:port

advertised.listeners=yourip:port
```

**修改好后，依次进入各服务器/${kafka_dir}/bin目录执行如下命令：**

```
./kafka-server-start.sh -daemon ../config/server.properties
#-daemon：以守护进程方式执行，避免shell窗口阻塞

#停止命令
./kafka-server-stop.sh
```

**启动后，进入logs目录查看启动日志，server.log，有如下日志，集群启动成功。可通过jps命令验证。**

```
[2020-04-03 17:55:08,643] INFO [ExpirationReaper-0-topic]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2020-04-03 17:55:08,645] INFO [ExpirationReaper-0-Heartbeat]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2020-04-03 17:55:08,651] INFO [ExpirationReaper-0-Rebalance]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2020-04-03 17:55:08,663] INFO [GroupCoordinator 0]: Starting up. (kafka.coordinator.group.GroupCoordinator)
[2020-04-03 17:55:08,670] INFO Successfully created /controller_epoch with initial epoch 0 (kafka.zk.KafkaZkClient)
[2020-04-03 17:55:08,676] INFO [GroupCoordinator 0]: Startup complete. (kafka.coordinator.group.GroupCoordinator)
[2020-04-03 17:55:08,691] INFO [GroupMetadataManager brokerId=0] Removed 0 expired offsets in 5 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
[2020-04-03 17:55:08,700] INFO [ProducerId Manager 0]: Acquired new producerId block (brokerId:0,blockStartProducerId:0,blockEndProducerId:999) by writing to Zk with path version 1 (kafka.coordinator.transaction.ProducerIdManager)
[2020-04-03 17:55:08,733] INFO [TransactionCoordinator id=0] Starting up. (kafka.coordinator.transaction.TransactionCoordinator)
[2020-04-03 17:55:08,734] INFO [TransactionCoordinator id=0] Startup complete. (kafka.coordinator.transaction.TransactionCoordinator)
[2020-04-03 17:55:08,737] INFO [Transaction Marker Channel Manager 0]: Starting (kafka.coordinator.transaction.TransactionMarkerChannelManager)
[2020-04-03 17:55:08,772] INFO [/config/changes-event-process-thread]: Starting (kafka.common.ZkNodeChangeNotificationListener$ChangeEventProcessThread)
[2020-04-03 17:55:08,807] INFO [SocketServer brokerId=0] Started data-plane processors for 1 acceptors (kafka.network.SocketServer)
[2020-04-03 17:55:08,813] INFO Kafka version: 2.2.1 (org.apache.kafka.common.utils.AppInfoParser)
[2020-04-03 17:55:08,813] INFO Kafka commitId: 55783d3133a5a49a (org.apache.kafka.common.utils.AppInfoParser)
[2020-04-03 17:55:08,814] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)
```

**启动其他服务器的kafka集群，启动命令同上。**

**本文在进行集群搭建时，未遇到明显异常，所以坑没踩够。**

### **三、集群验证**

#### **3.1.创建测试topic**

```
 ./bin/kafka-topics.sh --create --zookeeper 192.168.0.3:2181 --replication-factor 2 --partitions 6 --topic test_topic
 
#--zookeeper 配置文件中，zookeeper.connect的值
#--replication-factor 该topics的副本数
#--partitions 该topics的分区数
#--topic 名称
```

#### **3.2.查看topic**

```
#所有topic
./bin/kafka-topics.sh --list --zookeeper 192.168.0.3:2181

#__consumer_offsets主题，这个是kafka创建用来存储客户端消费消息的偏移量，0.9版本之前偏移量是放在zookeeper里，后面优化到kafka维护。

#指定topic详情
./bin/kafka-topics.sh --describe --zookeeper 192.168.0.3:2181 --topic test_topic

```

**显示详情如下：**

**略。**

**参数说明参考下表格：**

| **PartiticonCount**   | **显示分区数量一共有多少**                                                                                                                                                                                                                   |
| ----------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **ReplicationFactor** | **副本因子是多少**                                                                                                                                                                                                                           |
| **Partition**         | **分区编号**                                                                                                                                                                                                                                 |
| **Leader**            | **显示Leader副本在哪个Broker上，这里是不同分区会有不同，表示Leader在broker.id=0的服务器上。三个分区每个分区有三个副本，分区编号从0开始，所以这个Leader是说后面Replicas副本里面哪个是Leader。Leader副本提供读写，非Leader副本只做数据备份。** |
| **Replicas**          | **显示该partitions所有副本存储在哪些节点上 broker.id 这个是配置文件中设置的，它包括leader和follower节点**                                                                                                                                    |
| **Isr**               | **显示副本都已经同步的节点集合，这个集合的所有节点都是存活的，并且跟LEADER节点同步**                                                                                                                                                         |

#### **3.3.启动生产者**

```
./bin/kafka-console-producer.sh --broker-list xx.xx.xx.xx:9092 --topic test_topic
#输入发送的消息,回车发送.
```

#### **3.3.启动消费者**

```
./bin/kafka-console-consumer.sh --bootstrap-server xx.xx.xx.xx:9092 --topic test_topic --from-beginning

#--from-beginning:从偏移量初始位置开始消费(不带此参数,无法消费到消费者未启动前发送的消息)
```

**如果消费者端成功接收到生产者的消息,则验证成功.**

#### **3.4.springboot集成**

* **引入kafka依赖**

```

        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            
        </dependency>

<dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka-test</artifactId>
            <scope>test</scope>
        </dependency>

```

* **springboot配置**

```
spring: 
 #kafka
  kafka:
    # 以逗号分隔的地址列表，用于建立与 Kafka 集群的初始连接 (kafka 默认的端口号为 9092)
    bootstrap-servers: 114.67.92.113:9092,114.67.92.114:9092,114.67.92.122:9092
    producer:
      # 发生错误后，消息重发的次数。
      retries: 0
      #当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算。
      batch-size: 16384
      # 设置生产者内存缓冲区的大小。
      buffer-memory: 33554432
      # 键的序列化方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      # 值的序列化方式
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      # acks=0 ： 生产者在成功写入消息之前不会等待任何来自服务器的响应。
      # acks=1 ： 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成功响应。
      # acks=all ：只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。
      acks: 1
    consumer:
      # 自动提交的时间间隔 在 spring boot 2.X 版本中这里采用的是值的类型为 Duration 需要符合特定的格式，如 1S,1M,2H,5D
      auto-commit-interval: 1S
      # 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：
      # latest（默认值）在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的记录）
      # earliest ：在偏移量无效的情况下，消费者将从起始位置读取分区的记录
      auto-offset-reset: earliest
      # 是否自动提交偏移量，默认值是 true,为了避免出现重复数据和数据丢失，可以把它设置为 false,然后手动提交偏移量
      enable-auto-commit: true
      # 键的反序列化方式
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      # 值的反序列化方式
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    listener:
      # 在侦听器容器中运行的线程数。
      concurrency: 5
```

* **生产者代码示例**

```
@Component
public class KafkaProducer {
    @Autowired
    private KafkaTemplate kafkaTemplate;

public void sendMessage(String topic, String message) {

/*
         * 这里的 ListenableFuture 类是 spring 对 java 原生 Future 的扩展增强,是一个泛型接口,用于监听异步方法的回调
         * 而对于 kafka send 方法返回值而言，这里的泛型所代表的实际类型就是 SendResult<K, V>,而这里 K,V 的泛型实际上
         * 被用于 ProducerRecord<K, V> producerRecord,即生产者发送消息的 key,value 类型
         */
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, "", message);

future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable throwable) {
                System.err.println("发送消息失败:" + throwable.getMessage());
            }

@Override
            public void onSuccess(SendResult<String, Object> sendResult) {
                System.out.println("发送结果:" + sendResult.toString());
            }
        });
    }
}
```

* **消费者代码示例**

```
@Component
public class KafkaTestConsumer {
    //groupId:消费组的ID;topics:要监听的主题
    @KafkaListener(groupId = "testGroup", topics = "test_topic")
    public void consumer(ConsumerRecord<String, String> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
            Consumer consumer) {
        System.out.println("消费者收到消息:" + record.value() + "; topic:" + topic);
        engineService.decision(record.value());
        /*
         * 如果需要手工提交异步 consumer.commitSync();
         * 手工同步提交 consumer.commitAsync()
         */
    }
}
```

### **四、集群监控**

#### **4.1.监控对象**

* **集群运行情况**
* **主题监控：列表、主题管理**
* **消费者监控：消费情况、偏移量**
* **可视化图形**

#### **4.2.监控工具对比**

**针对4.1，做如下社区常用kafka监控工具对比**

| **工具** | **集群情况** | **主题监控** | **消费者监控** | **可视化图形** | **安装简易程度** |
| ---------- | -------------- | -------------- | ---------------- | ---------------- | ------------------ |
|          |              |              |                |                |                  |

#### **4.3.kafka-eagle安装**

### **五、附：常用命令**

最后修改：2022 年 03 月 14 日 11 : 31 AM

如果觉得我的文章对你有用，请随意赞赏

1 条评论

aweather
August 6th, 2021 at 05:22 pm

wawaawawawa~~

回复

发表评论取消回复

评论 *

私密评论

名称 *

🎲

邮箱 *

地址

Kafka安装及使用

smalleyes • 2018 年 07 月 24 日