什么是Kafka

<code>Apache Kafka® is a distributed streaming platformA streaming platform has three key capabilities:Publish and subscribe to streams of records, similar to a message queue or enterprise messaging system.Store streams of records in a fault-tolerant durable way.Process streams of records as they occur.Kafka is generally used for two broad classes of applications:Building real-time streaming data pipelines that reliably get data between systems or applicationsBuilding real-time streaming applications that transform or react to the streams of dataTo understand how Kafka does these things, let's dive in and explore Kafka's capabilities from the bottom up.First a few concepts:Kafka is run as a cluster on one or more servers that can span multiple datacenters.The Kafka cluster stores streams of records in categories called topics.Each record consists of a key, a value, and a timestamp./<code>

Kafka是一款分布式消息发布和订阅系统，它的特点是高性能、高吞吐量。

最早设计的目的是作为LinkedIn的活动流和运营数据的处理管道。这些数据主要是用来对用户做用户画

像分析以及服务器性能数据的一些监控所以kafka一开始设计的目标就是作为一个分布式、高吞吐量的

消息系统，所以适合运用在大数据传输场景。如的开源分布式处理系统如cloudera 、Storm 、Spark、

Flink 等都支持与 Kafka 集成

Kafka 起初是由 Linkedin 公司采用 Scala 语言开发的 个多分区、多副本且基于 ZooKeeper 协调的分布式消息系统

目前 Kafka 已经定位为一个分布式流式处理平台，它以高吞吐、可持久化、可水平扩展、支持流数据处理等多种特性而被广泛使用。

• 消息系统： kafka 和传统的消息系统（也称作消息中间件〉都具备系统解稿、冗余存储、流量削峰、缓冲、异步通信、扩展性、 可恢复性等功能。与此同时， Kafka供了大多数消息系统难以实现的消息顺序性保障及回溯消费的功能

• 存储系统： Kafka 把消息持久化到磁盘，相比于其他基于内存存储的系统而言，有效地降低了数据丢失的风险 也正是得益于 Kafka 的消息持久化功能和多副本机制，我们可以把 Kafka 作为长期的数据存储系统来使用，只需要把对应的数据保留策略设置为“永久”或启用主题的日志压缩功能即可

• 流式处理平台： Kafka 不仅为每个流行的流式处理框架提供了可靠的数据来源，还供了一个完整的流式处理类库，比如窗口、连接、变换和聚合等各类操

Kafka的企业应用场景

由于kafka具有更好的吞吐量、内置分区、冗余及容错性的优点(kafka每秒可以处理几十万消息)，让

kafka成为了一个很好的大规模消息处理应用的解决方案。所以在企业级应用长，主要会应用于如下几

个方面

行为跟踪：kafka可以用于跟踪用户浏览页面、搜索及其他行为。通过发布-订阅模式实时记录到对应的

topic中，通过后端大数据平台接入处理分析，并做更进一步的实时处理和监控

日志收集：日志收集方面，有很多比较优秀的产品，比如Apache Flume，很多公司使用kafka代理日志

聚合。日志聚合表示从服务器上收集日志文件，然后放到一个集中的平台（文件服务器）进行处理。在

实际应用开发中，我们应用程序的log都会输出到本地的磁盘上，排查问题的话通过linux命令来搞定，

如果应用程序组成了负载均衡集群，并且集群的机器有几十台以上，那么想通过日志快速定位到问题，

就是很麻烦的事情了。所以一般都会做一个日志统一收集平台管理log日志用来快速查询重要应用的问

题。所以很多公司的套路都是把应用日志集中到kafka上，然后分别导入到es和hdfs上，用来做实时检

索分析和离线统计数据备份等。而另一方面，kafka本身又提供了很好的api来集成日志并且做日志收集

在这里插入图片描述

kafka架构

一个典型的kafka集群包含若干Producer（可以是应用节点产生的消息，也可以是通过Flume收集日志产生的事件），若干个Broker（kafka支持水平扩展）、若干个Consumer Group，以及一个zookeeper集群。kafka通过zookeeper管理集群配置及服务协同。Producer使用push模式将消息发布到broker，consumer通过监听使用pull模式从broker订阅并消费消息。

多个broker协同工作，producer和consumer部署在各个业务逻辑中。三者通过zookeeper管理协调请求和转发。这样就组成了一个高性能的分布式消息发布和订阅系统。

图上有一个细节是和其他mq中间件不同的点，producer 发送消息到broker的过程是push，而consumer从broker消费消息的过程是pull，主动去拉数据。而不是broker把数据主动发送给consumer

名词解释

• 1）Broker 服务代理节点

Kafka集群包含一个或多个服务器，这种服务器被称为broker。broker端不维护数据的消费状态，提升

了性能。直接使用磁盘进行存储，线性读写，速度快：避免了数据在JVM内存和系统内存之间的复制，

减少耗性能的创建对象和垃圾回收。

• 2）Producer

负责发布消息到Kafka broker

• 3）Consumer

消息消费者，向Kafka broker读取消息的客户端，consumer从broker拉取(pull)数据并进行处理。

• 4）Topic

每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。（物理上不同Topic的消息分开存

储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消

费数据而不必关心数据存于何处）

• 5）Partition

Parition是物理上的概念，每个Topic包含一个或多个Partition.

• 6）Consumer Group

每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定

group name则属于默认的group）

• 7）Topic & Partition

Topic在逻辑上可以被认为是一个queue，每条消费都必须指定它的Topic，可以简单理解为必须指明把

这条消息放进哪个queue里。为了使得Kafka的吞吐率可以线性提高，物理上把Topic分成一个或多个

Partition，每个Partition在物理上对应一个文件夹，该文件夹下存储这个Partition的所有消息和索引文

件。若创建topic1和topic2两个topic，且分别有13个和19个分区，则整个集群上会相应会生成共32个

文件夹（本文所用集群共8个节点，此处topic1和topic2 replication-factor均为1）。

Kafka安装部署

安装kakfa之前需要安装jdk和zookeeper

版本：

jdk8

zookeeper 3.4.12

kafka 2.11-2.0.0.tgz

1 jdk安装

kafka是scale语言编写，JVM系语言，所以需要安装jdk

<code># 下载jdk-8u181-linux-x64.tar.gz# 解压到/usr/javatar -zxvf jdk-8u181-linux-x64.tar.gz# 环境变量 vim /etc/profileexport JAVA_HOME=/usr/java/jdk1.8.0_181export JRE_HOME=$JAVA_HOME/jreexport PATH=$PATH:$JAVA_HOME/binexport CLASSPATH=./://$JAVA_HOME/bin:$JRE_HOME/lib# 配置生效source /etc/profile# 验证java -version/<code>

2 zookeper安装

Zoo Keeper 是安装 Kafka 集群的必要组件， afka 通过 ZooKeeper 来实施对元数据信息，包括集群 broker 、主题、 分区等内容。

ZooKeeper 是一个开源的分布式协调服务，是 google Chubby 个开源实现。分布式应用程序可基ZooKeeper 实现诸如数据发布／订阅 、负载均衡、 命名服务、分布式协调／通知，集群管理、 Master 举、配置维护等功能。在 ZooKeeper 中共有3个角色： leader follower和observer,同一时刻 ZooKeeper 集群中只会有一个 leader ，其他的都 follower和observer。observer不参与投票，默认情况下 ZooKeeper 中只有 leader和follewer两个角色

<code># 下载 zookeeper-3.4.12.tar.gz 到 /opt# 解压tar -zxvf zookeeper-3.4.12.tar.gz# 进入zookeepr目录cd zookeeper-3.4.12# 配置环境变量 /etc/profileexport ZOOKEEPER_HOME=/opt/zookeeper-3.4.12export PATH=$PATH:$ZOOKEEPER_HOME/bin# 使其生效source /etc/profile# 配置文件 进入$ZOOKEEPER_HOME/confcd confcp zoo_sample.cfg zoo.cfg# 配置zoo.cfg#####################zoo.cfg###################### Zookeeper服务器心跳时间,mstickTime=2000# 投票选举新leader的初始化时间initLimit=10# leader于follower心跳检测最大容忍时间，响应超过syncLimit*ticeTime，leader认为# follower “死掉",从服务器列表中删除followersyncLimit=5# 数据目录dataDir=/tmp/zookeeper/data# 日志目录dataLogDir=/tmp/zookeeper/log# Zookeeper对外服务端口clientPort=2181################################################## 创建数据目录和日志目录mkdir -p /tmp/zookeeper/datamkdir -p /tmp/zookeeper/log# 在${dataDir}目录(/tmp/zookeeper/data)下创建一个myid文件，并写入# 一个数值，比如0。myid文件里存放的是服务器的编号# 启动服务zkServer.sh start# 查看服务状态 zkServer.sh status# zk集群 生成环境都是使用集群# /etc/hosts 192.168.0.111 node1192.168.0.112 node2192.168.0.113 node3# 在3台机器的zoo.cfg中添加以下配置server.0=192.168.0.111:2888:3888server.1=192.168.0.112:2888:3888server.2=192.168.0.112:2888:3888/<code>

3 kafka安装

<code># 下载 kafka_2.11-2.0.0.tgz# 解压到/opttar -zxvf kafka_2.11-2.0.0.tgzcd kafka_2.11-2.0.0# Kafka的根目录$KAFKA_HOME,即/opt/kafka_2.11-2.0.0 # 环境配置 /etc/profileexport KAFKA_HOME=/opt/kafka_2.11-2.0.0export PATH=$PATH:$KAFKA_HOME/bin# 使其生效source /etc/profile# 修改配置文件 $KAFKA_HOME/conf/server.properties###################conf/server.properties################# broker的编号，如果有多个broker，则每个broker的编号需要设置不同broker.id=0# broker对外提供的服务入口地址listeners=PLAINTEXT://localhost:9092# 存放消息日志文件的地址log.dirs=/tmp/kafka-logs# kafka所需要的Zooper集群地址,单机为了实验kafka和zk安装本机上zookeeper.connect=localhost:2181/kafka########################################################## 启动kafka，在$KAFKA_HOME目录bin/kafka-server-start.sh config/server.properties# 后台启动bin/kafka-server-start.sh -daemon config/server.properties # 或bin/kafka-server-start.sh config/server.properties & # 停止kafkabin/kafka-server-stop.sh -daemon config/server.properties ########################集群##########################192.168.0.111192.168.0.112192.168.0.113# 三台环境都如上面下操作kafka# 分别修改三台机器的server.properties,在同一个集群中每台机器id必须唯一broker.id=0broker.id=1broker.id=2# 修改zooper.connect 验证zk安装在111机器上zookeeper.connect=192.168.0.111:2181# 修改listeners配置# 如果配置了listeners,那么消息生产者和消费者都会使用listeners的配置来进行消息的收发，否则# 会使用localhost# PLAINTEXT表示协议，默认明文，可以选择其他加密协议listeners=PLAINTEXT://192.168.0.111:9092# 分别启动三台服务器sh kafka-server-start.sh -daemon ../config/server.properties/<code> 

kafka的基本操作

创建topic

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

Replication-factor 表示该topic需要在不同的broker中保存几份，这里设置成1，表示在两个broker中保存两份

Partitions 分区数

查看topic

<code>sh kafka-topics.sh --describe --zookeeper localhost:2181/kafka --topic first_topic/<code>

消费消息

<code>sh kafka-console-consumer.sh --bootstrap-server 192.168.13.106:9092 --topic test --from-beginning/<code>

发送消息

<code>sh kafka-console-producer.sh --broker-list 192.168.244.128:9092 --topic first_topic/<code>

————————————————

原文链接：https://blog.csdn.net/liulong1010/article/details/103842244

閱讀更多 學java的小亂 的文章

關鍵字: 架构师 Clo 设计