正文

kafka的安装部署及应用(代码片段)

-飞鹤-  -飞鹤-  2022-12-14  221

关键词：

1. 介绍

1.1. 简介

kafka是一款分布式、支持分区的、多副本，基于zookeeper协调的分布式消息中间件。其最大特性就是能够实时处理大量消息。

1.2. 基本概念

Broker, 消息中间件的处理节点，也即服务器。一个broker一个服务器，多个broker即构成一个kafka的服务集群。
Topic, kafka根据topic来对消息进行分类，每个发布到kafka的消息都必须指定一个topic。
Partition，为了应对大量的topic消息，可以将topic消息分为不同的partition，并且每个partition可以分布到不同上的Broker上，达到负载均衡。Partition也可以分布在一台电脑上的不同磁盘上，可以提升磁盘的I/O响应率。
Replication，每一个分区都可以有多个副本，用作备份，在主分区(Leader分区)数据异常时，可以将备份分区(Follower分区)设置为主分区(Leader)。
Producer，消息生产者，也即消息发布者。
Consumer，消息消费者，也即消费订阅者。
Consumer group，消息订阅集群，每个consumer都属于一个特定的consumer group，若未指定，则属于默认的group。一个group中只有一个comsumer能够消费消息。
Zookeeper, kafka 集群依赖 zookeeper 来保存集群的的元信息，来保证系统的可用性。Zookeeper负责管理集群中的消息负载均衡，将消息分配到不同的分区，并且判断使用哪个副本作为Leader分区等。

1.3. 应用场景

日志收集，可以用kafka收集各种服务的日志，通过已统一接口的形式开放给各种消费者。
消息系统，解耦生产和消费，并且可以基于消息系统提供事件响应框架。
分布式大数据处理，基于Kafka的分布式和高性能，可以用于分布式大数据处理。

1.4. 特点

可靠性，kafka的传输是基于tcp的自定义协议，可以设置ack机制来保证数据数据在副本中由浅完成才返回，并且如果数据异常，默认会有3次重发机制。
灵活性，kafka的消费记录是以offset来控制的，即消费者可以灵活地获取想要的消息。
可扩展性，kafka可以非常方便地扩容服务器(Broker)，而不影响原有的服务器功能。
高性能，Kafka采用顺序添加消息，并且采用页缓存和零拷贝技术，可以大幅提升磁盘I/O利用率。

1.5. 生产数据

kafka 每次把生产的数据发送向Leader分区，并顺序写入到磁盘，然后Leader分区会将数据同步到各个从分区Follower，即使主分区挂了，也不会影响服务的正常运行。

数据在写入的时候可以指定需要写入的分区，如果有指定，则写入对应的分区。
如果没有指定分区，但是设置了数据的key，则会根据key的值hash出一个分区。
如果既没指定分区，又没有设置key，则会轮询选出一个分区。

1.5.1. 三种数据发送方式

Kafka中有三种发送消息的方式：

只发不管结果（fire-and-forget）：只调用接口发送消息到Kafka服务器，但不管成功写入与否。由于Kafka是高可用的，因此大部分情况下消息都会写入，但在异常情况下会丢消息。
同步发送（Synchronous send）：调用send()方法返回一个Future对象，我们可以使用它的get()方法来判断消息发送成功与否。会阻塞当前线程，执行效率低，但可靠性最高。
异步发送（Asynchronous send）：调用send()时提供一个回调方法，不会阻塞当前线程，当接收到broker结果后回调此方法。可靠性不如同步发送，可能因为等问题不保证100%被回调。

1.5.2. 生产者常用参数

acks：用来指定分区中必须要有多少个副本收到这条消息，之后生产者才会认为这条消息是成功写入的。

acks = 1, 默认值即为1.生产者发送消息之后，只要分区的leader副本成功写入消息，那么它就会收到来自服务端的成功响应。如果消息无法写入leader副本，比如在leader副本崩溃、重新选举新的leader副本的过程中，那么生产者就会收到一个错误的响应，为了避免消息丢失，生产者可以选择重发消息。如果消息写入leader副本并返回成功响应给生产者，且在被其他follower副本拉取之前leader副本崩溃，那么此时消息还是会丢失，因为新选举的leader副本中并没有这条对应的消息。acks设置为1，是消息可靠性和吞吐量之间的折中方案。
acks = 0, 生产者发送消息之后不需要等待任何服务的相应。如果在消息从发送到写入kafka的过程中出现某些异常，导致kafka并没有收到这条消息，那么生产者也无从得知，消息也就丢失了。在其他配置环境相同的情况下，acks设置为0可以达到最大的吞吐量。
acks = -1 或acks = all。生产者在消息发送之后，需要等待ISR中的所有副本都成功写入消息之后才能够对来自服务端的成功响应。在其他配置环境相同的情况下，acks设置为-1（all）可以达到最强的可靠性。但这并不意味着消息就一定可靠，因为ISR中可能只有leader副本，这样就退化成了acks = 1 的情况。要获得更高的消息可靠性需要配合min.insync.replicas等参数的联动。
max.request.size: 用来限制生产者客户端能发送的消息的最大值，默认为1048576B，即1MB。
retries和retry.bakcoff.ms： retries参数用来配置生产者重试的次数。默认值为0，即在发生异常的时候不进行任何重试动作。retry.backoff.ms默认值为100，用来设置两次重试之间的时间间隔。避免无效的频繁重试。
compression.type：这个参数用来指定消息的压缩方式，默认值为“none”，默认情况下，消息不会被压缩。可配置为“gzip”、“snappy”、和“lz4”。
connections.max.idle.ms: 用来指定在多久之后关闭闲置的连接，默认值是540000（ms），及9分钟。
linger.ms: 用来指定生产者发送ProducerBatch之前等待更多消息（ProducerRecord）加入ProducerBatch的时间，默认值为0.生产者客户端会在ProducerBatch被填满或等待时间超过linger.ms值时发送出去。
receive.buffer.bytes: 用来设置Socket接收消息缓冲区（SO_RECBUF）的大小，默认值为131072（B），即32KB。
send.buffer.bytes: 用来设置Socket发送消息缓冲区（SO_SNDBUF）的大小，默认值为131072（B），即32KB。
request.timeout.ms: 用来配置Producer等待请求响应的最长时间，默认值为30000（ms）。请求超时之后可以选择重试。注意：这个参数需要比broker端参数raplica.lag.time.max.ms的值要大，这样可以减少因客户端重试而引起的消息重复的概率。

1.6. 消费数据

消费者主动的去kafka集群拉取消息时，也是从Leader分区去拉取数据。
多个消费者组成一个消费者组，并且同一个消费者组内的消费者不能消费同一个分区中的消息，以免偏移混乱，可以消费不同分区中的消息。
一般建议一个消费者对应一个分区，以提升响应性能。

1.6.1. 消费者参数

fetch.min.bytes
一次拉取的最小字节数(1B)
fetch.max.bytes
一次拉取的最大数据量(50M)
fetch.max.wait.ms
拉取时的最大等待时长(500ms)
max.partition.fetch.bytes
每个分区一次拉取的最大数据量(1MB)
max.poll.records
一次拉取的最大条数(500)
connections.max.idle.ms
网络连接的最大闲置时长(540000ms)
request.timeout.ms
一次请求等待响应的最大超时时间，consumer 等待请求响应的最长时间(30000ms)
metadata.max.age.ms
元数据在限定时间内没有进行更新，则会被强制更新(300000)
reconnect.backoff.ms
尝试重新连接指定主机之前的退避时间(50ms)
retry.backoff.ms
尝试重新拉取数据的重试间隔(100ms)
isolation.level
隔离级别！决定消费者能读到什么样的数据read_uncommitted：可以消费到 LSO（LastStableOffset）位置； read_committed：可以消费到 HW（High Watermark）位置
max.poll.interval.ms
超过时限没有发起 poll 操作，则消费组认为该消费者已离开消费组

2. 安装

kafka是基于Java/Scala的软件，所以在安装kafka之前需要先安装java。Linux下安装更方便，且更能利用网络的性能，建议在Linux下安装。

2.1. 安装java

直接安装默认的即可。

sudo apt install default-jre

2.2. 安装kafka

新版的kafka默认已经集成了zookeeper，所以只要下载稳定版解压即可。

mkdir kafka
cd kafka
wget https://dlcdn.apache.org/kafka/3.3.1/kafka_2.13-3.3.1.tgz
sudo tar -xzf kafka_2.13-3.3.1.tgz

3. 部署

kafka的部署分为两种方式，一种方式是基于zookeeper，另一种方式是单独部署。

3.1. 基于zookeeper

3.1.1. 启动zookeeper

zookeeper的配置文件config/zookeeper.properties，一般不需要改动，如果有端口冲突，可以配置新的端口号。

sudo ./bin/zookeeper-server-start.sh config/zookeeper.properties

3.1.2. 启动kafka

kafka配置文件config/server.properties的常用配置如下：

#broker的全局唯一编号，不能重复
broker.id=0
#删除topic功能使能
delete.topic.enable=true
#处理网络请求的线程数量
num.network.threads=3
#用来处理磁盘IO的现成数量
num.io.threads=8
#发送套接字的缓冲区大小
socket.send.buffer.bytes=102400
#接收套接字的缓冲区大小
socket.receive.buffer.bytes=102400
#请求套接字的缓冲区大小
socket.request.max.bytes=104857600
#kafka运行日志存放的路径
log.dirs=/opt/module/kafka/logs
#topic在当前broker上的分区个数
num.partitions=1
#用来恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1
#segment文件保留的最长时间，超时将被删除
log.retention.hours=168
#配置连接Zookeeper地址
zookeeper.connect=172.18.1.22:2181
# 配置连接Zookeeper集群地址，可以将一个broker分布到多个zookeeper上
# zookeeper.connect=172.18.1.22:2181,172.18.1.23:2181,172.18.1.24:2181

单机单broker部署

sudo bin/kafka-server-start.sh config/server.properties

单机多broker部署
配置多个server1.properties，server2.properties，server3.properties中的listeners为不同的端口，然后启动多个进程

sudo bin/kafka-server-start.sh config/server1.properties
sudo bin/kafka-server-start.sh config/server2.properties
sudo bin/kafka-server-start.sh config/server3.properties

多机多broker部署
在多台电脑上启动zookeeper，再配置server1.properties中的zookeeper.connect分别对应不同电脑上的zookeeper。然后启动多个kafka。。

sudo bin/kafka-server-start.sh config/server1.properties
sudo bin/kafka-server-start.sh config/server2.properties
sudo bin/kafka-server-start.sh config/server3.properties

3.2. 单一模式

3.2.1. 单broker模式

单一模式，也即KRaft模式，不需要zookeeper。这是kafka新版本提供的功能，此模式避免与zookeeper通信，可以提升管理性能。

生成UUID

KAFKA_CLUSTER_ID=“$(bin/kafka-storage.sh random-uuid)”

bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/kraft/server.properties

启动服务器

sudo ./bin/kafka-server-start.sh config/kraft/server.properties

3.2.2. 多broker模式

配置config/kraft/server.properties文件

# The role of this server. Setting this puts us in KRaft mode
process.roles=broker,controller

# The node id associated with this instance's roles
node.id=1

# @前面的序列号为node.id, roles=controller的都要配置在此处, 以考虑最终controller的投票
controller.quorum.voters=1@localhost:9093,2@172.18.11.12:9093,3@172.18.11.13:9093

4. 应用

4.1. 命令行应用

创建主题

# 默认broker,partition
bin/kafka-topics.sh --create --topic quickstart --bootstrap-server localhost:9092
# 多集群
bin/kafka-topics.sh --create --topic quickstart --bootstrap-server localhost:9092,172.18.11.12:9092,,172.18.11.13:9092
# 指定broker，partition，并且指定3个副本
kafka-topics.sh --create --zookeeper localhost:9092 --replication-factor 3 --partitions 1 --topic quickstart-events

查询主题

# KRaft模式
sudo bin/kafka-topics.sh --list --bootstrap-server localhost:9092
## 多集群
sudo bin/kafka-topics.sh --list --bootstrap-server localhost:9092,172.18.11.12:9092,,172.18.11.13:9092
sudo bin/kafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092
# zookeeper模式
sudo bin/kafka-topics.sh --list --zookeeper localhost:2181
sudo bin/kafka-topics.sh --describe --zookeeper localhost:2181

生产消息

# KRaft模式 
$ bin/kafka-console-producer.sh --topic quickstart-events --bootstrap-server localhost:9092
This is my first event
This is my second event
# 多zookeeper模式
$ bin/kafka-console-producer.sh --broker-list localhost:9092 localhost:9093 localhost:9094 --topic my-replicated-topic

消费消息

# KRaft模式
$ bin/kafka-console-consumer.sh --topic quickstart-events --from-beginning --bootstrap-server localhost:9092
This is my first event
This is my second event
# zookeeper模式
kafka-console-consumer.sh --zookeeper locahost:2181 --topic my-replicated-topic

删除主题
先要确保配置文件config/kraft/server.properties中配置了可删除。

auto.create.topics.enable = false
delete.topic.enable=true

删除指令

# KRaft模式
bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic quickstart-events

点对点消费
点对点消费，即发一个生产者对应一个消费者，订阅同一个主题的消费者的group id必须一样，即对一个主题而言，同时只有一个消费者在消费。
订阅发布
订阅发布，即一个主题有多个消费者，每个消费者的Group ID不一样，即可以使多个消费者同时消费一个主题。
删除消息
消除消息，只能通过配置服务的log.retention.hours=48或log.retention.bytes=1073741824来自动删除消息。

4.2. Python

kafka默认提供的是Java接口，但是C++，Python的接口也都有提供。

安装kafka的Python客户端接口库

pip install kafka-python

示例

from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import kafka_errors
import traceback
import json


def producer_demo():
    # 假设生产的消息为键值对（不是一定要键值对），且序列化方式为json
    producer = KafkaProducer(
        bootstrap_servers=['localhost:9092'], 
        key_serializer=lambda k: json.dumps(k).encode(),
        value_serializer=lambda v: json.dumps(v).encode())
    # 发送三条消息
    for i in range(0, 3):
        future = producer.send(
            'kafka_demo',
            key='count_num',  # 同一个key值，会被送至同一个分区
            value=str(i),
            partition=1)  # 向分区1发送消息
        print("send ".format(str(i)))
        try:
        	# kafka默认是异步的, 消费会先发送到消息缓存中,根据配置的时间来定时将消息缓存发送到服务器
        	# 默认立即发往服务器
            future.get(timeout=10) # 监控是否发送成功           
        except kafka_errors:  # 发送失败抛出kafka_errors
            traceback.format_exc()


def consumer_demo():
    consumer = KafkaConsumer(
        'kafka_demo', 
        bootstrap_servers=':9092',
        group_id='test'
    )
    for message in consumer:
        print("receive, key: , value: ".format(
            json.loads(message.key.decode()),
            json.loads(message.value.decode())
            )
        )
        
consumer_demo()
producer_demo()

4.3. C/C++

安装
参见：https://gitcode.net/mirrors/edenhill/librdkafka?utm_source=csdn_github_accelerator
示例
参见：https://developer.confluent.io/get-started/c/

kafka安装及部署(代码片段)

阅读目录一、环境配置二、操作过程Kafka介绍安装及部署回到顶部一、环境配置操作系统：Cent OS 7Kafka版本：0.9.0.0Kafka官网下载：请点击JDK版本：1.7.0_51SSHSecureShell版本：XShell5回到顶部二、操作过程1、下载Kafka并解压... 查看详情

kafka的安装部署及应用(代码片段)

1.介绍1.1.简介kafka是一款分布式、支持分区的、多副本，基于zookeeper协调的分布式消息中间件。其最大特性就是能够实时处理大量消息。1.2.基本概念Broker,消息中间件的处理节点，也即服务器。一个broker一个服务器，... 查看详情

win10安装virtualbox及部署kafka踩坑记录(代码片段)

...要部署一套kafka环境进行调试，所以考虑使用virtualbox安装centos来部署kafka。windows下载安装virtualbox很简单，不清楚的可以从网上查询相关资料。环境virtualbox: VirtualBox-6.1.34a-150636-Wincentos: CentOS-7-x 查看详情

win10安装virtualbox及部署kafka踩坑记录(代码片段)

kafka集群的部署及kafka监控工具(代码片段)

...习记录(术语)3.kafka集群部署(直接部署在了一台机器上)3.1安装多节点zookeeper集群3.2配置kafka集群3.3启动集群3.4设置zookeeper开机自启动3.4.1法一(简单)3.4.2法二(将zookeeper做成服务)3.5配置kafka开机自启动4.验证部署4.1测试topic的创建和删... 查看详情

kafka单节点部署及使用(代码片段)

...测试需要，我下载的是kafka_2.11-0.9.0.0.tgz2.上传解压将安装包上传到/opt/software目录，查看详情

kafka(代码片段)

...2消息队列的两种模式1.3Kafka基础架构第2章Kafka快速入门2.1安装部署2.1.1集群规划2.1.3集群部署2.2Kafka命令行操作第3章Kafka架构深入3.1Kafka工作流程及文件存储机制3.2Kafka生产者3.2.1分区策略（封装成ProducerRecord对象&#x 查看详情

kafka(代码片段)

...1.2.2消息队列的两种模式1.3Kafka基础架构2Kafka快速入门2.1安装部署2.1.1集群规划2.1.2jar包下载2.1.3集群部署2.1.4编写群起脚本2.2Kafka命令行操作3Kafka架构深入3.1Kafka工作流程及文件存储机制3.2Kafka生产者3.2.1分区策略3.2.2数据可靠性保... 查看详情

kafka(代码片段)

基于cdh的kafka配置及部署（详细，成功运行）(代码片段)

一、下载http://archive.cloudera.com/kafka/parcels/2.2.0/wgethttp://archive.cloudera.com/kafka/parcels/2.2.0/KAFKA-2.2.0-1.2.2.0.p0.68-el6.parcelwgethttp://archive.cloudera.com/kafka/parcels/2.2.0/KAFKA-2. 查看详情

消息中间件--kafka安装部署(代码片段)

一、概念1、什么是kafkaKafka是由Apache软件基金会开发的一个开源的分布式流处理平台，由Scala和Java编写。Kafka是一个分布式、分区的、多副本的、多订阅者，基于zookeeper协调的分布式日志系统（也可以当做MQ系统），它可以处理消... 查看详情

解开kafka神秘的面纱：kafka单机部署和集群部署(代码片段)

文章目录一、前言二、kafka单机安装2.1下载压缩包2.2zookeeper安装2.3kafka安装三、kafka集群安装3.1zookeeper安装3.2kafka的安装3.3两种Topic四、尾声一、前言本文主要介绍kafka的单机和集群模式部署。二、kafka单机安装2.1下载压缩包需要先... 查看详情

win10安装virtualbox及部署kafka踩坑记录(代码片段)

kafka入门概述部署与api的简单使用(代码片段)

...Partition分区副本SegmentOffset对比Kafka集群架构部署第一台机安装分发及修改其余节点配置启动与关闭封装脚本启动脚查看详情

kafka安装部署(代码片段)

kafka安装教程 1.下载下载地址：http://kafka.apache.org/downloads以键头所指版本为例 2.安装下载的版本已经编译，直接解压到想要的目录就算安装好了tar-zxfkafka_2.11-0.11.0.1.tgz-C/usr/myapp 3.配置单节点3.1配置单节点zookeeper我们使... 查看详情

kafka的入门及安装使用(代码片段)

...fka学习之路（三）Kafka的高可用Kafka学习之路（四）Kafka的安装Kafka学习之路（五）Kafka在zookeeper中的存储二：Kafka架构原理（一）Kafka应用场景Kafka是最初由Linkedin公司开发，是一个分布式、分区的、多副本的、多订阅者，基于zookee... 查看详情

docker及常用软件的安装部署(代码片段)

Docker内容简介1.安装与运行2.常用命令3.镜像相关命令4.创建与启动容器5.部署应用内容简介本篇文章介绍Docker安装部署和在Docker安装一些常用软件。首先介绍一下什么是Docker摘自百度百科：Docker是一个开源的应用容器引擎࿰... 查看详情

kafka的简单了解和安装部署(代码片段)

这里写目录标题一、kafka概述二、部署kafka1、下载jdk2、安装zookeeper3、安装kafka一、kafka概述Kafka是一种消息队列，主要用来处理大量数据状态下的消息队列，一般用来做日志的处理。既然是消息队列，那么Kafka也就拥有... 查看详情