大纲：

  1. kafka是什么？

  2. JMS规范是什么？

  3. 为什么需要消息队列？

  4. Kafka核心组件

  5. Kafka安装部署

  6. Kafka生产者Java API

  7. Kafka消费者Java API

什么是消息系统？

消息系统负责将数据从一个应用程序传输到另一个应用程序，因此应用程序可以专注于数据，但不担心如何共享它。 分布式消息传递基于可靠消息队列的概念。 消息在客户端应用程序和消息传递系统之间异步排队。 有两种类型的消息模式可用 - 一种是点对点，另一种是发布 - 订阅(pub-sub)消息系统。 大多数消息模式遵循 pub-sub 。

点对点消息系统

在点对点系统中，消息被保留在队列中。 一个或多个消费者可以消耗队列中的消息，但是特定消息只能由最多一个消费者消费。 一旦消费者读取队列中的消息，它就从该队列中消失。 该系统的典型示例是订单处理系统，其中每个订单将由一个订单处理器处理，但多个订单处理器也可以同时工作。 下图描述了结构。
Kafka - 图1

发布 - 订阅消息系统

在发布 - 订阅系统中，消息被保留在主题中。 与点对点系统不同，消费者可以订阅一个或多个主题并使用该主题中的所有消息。 在发布 - 订阅系统中，消息生产者称为发布者，消息使用者称为订阅者。 一个现实生活的例子是Dish电视，它发布不同的渠道，如运动，电影，音乐等，任何人都可以订阅自己的频道集，并获得他们订阅的频道时可用。
Kafka - 图2

1、Kafka是什么

在流式计算中，Kafka一般用来缓存数据，Storm通过消费Kafka的数据进行计算。 KAFKA + STORM +REDIS 中国版kafka ——> metaQ
Apache Kafka是一个开源消息系统，由Scala写成。是由Apache软件基金会开发的一个开源消息系统项目，专为分布式高吞吐量系统而设计。类似于JMS，可以主动拉取数据。
Kafka最初是由LinkedIn开发，并于2011年初开源。2012年10月从Apache Incubator毕业。该项目的目标是为处理实时数据提供一个统一、高通量、低等待的平台。
Kafka，是一个分布式发布 - 订阅消息系统和一个强大的队列（分布式消息队列：生产者、消费者的功能），它提供了**类似于JMS的特性，但是在设计实现上完全不同，此外它并不是JMS规范的实现，可以处理大量的数据，并使您能够将消息从一个端点传递到另一个端点。
Kafka适合离线和在线消息消费。 Kafka消息保留在磁盘上，并在群集内复制以防止数据丢失。 Kafka构建在ZooKeeper同步服务之上。 它与Apache Storm和Spark非常好地集成，用于实时流式数据分析。
Kafka对消息保存时根据Topic进行归类，发送消息者称为Producer,消息接受者称为Consumer,此外kafka集群有多个kafka实例组成，每个实例(server)成为broker。
无论是kafka集群，还是producer和consumer都依赖于zookeeper**集群保存一些meta信息，来保证系统可用性
Kafka内部通信：NIO
Kafka引入解决的问题;
1. JMS 1—> to—> 1 消息只有一份，消费者主动拉取
2. JMS 1—> to—> N 消息也只有一份，广播给订阅者
Kafka - 图3

2、JMS是什么

2.1、JMS的基础

JMS是什么：JMS是Java提供的一套技术规范，做到了服务器客户端之间的解耦（RPC，JMS），
RPC客户端调用服务端后，服务端需要给客户端返回结果。
JMS客户端只放消息，服务端不会给予会应。

JMS干什么用：用来异构系统 集成通信，缓解系统瓶颈，提高系统的伸缩性增强系统用户体验，使得系统模块化和组件化变得可行并更加灵活
通过什么方式：生产消费者模式（生产者、服务器、消费者）
Kafka - 图4
jdk，kafka，activemq……

2.2、JMS消息传输模型

点对点模式（一对一，消费者主动**拉取**数据，消息收到后消息清除）
点对点模型通常是一个基于拉取或者轮询的消息传送模型，这种模型从队列中请求信息，而不是将消息推送到客户端。这个模型的特点是发送到队列的消息被一个且只有一个接收者（消费者）接收处理，即使有多个消息监听者也是如此。即一个或多个消费者可以消耗队列中的消息，但是特定消息只能由最多一个消费者消费。一旦消费者读取队列中的消息，它就从该队列中消失。
该系统的典型示例是订单处理系统，其中每个订单将由一个订单处理器处理，但多个订单处理器也可以同时工作。 下图描述了结构。
发布/订阅模式（一对多，数据生产后，**推送**给所有订阅者）
发布订阅模型则是一个基于推送的消息传送模型。发布订阅模型可以有多种不同的订阅者，临时订阅者只在主动监听主题时才接收消息，而持久订阅者则监听主题的所有消息，即时当前订阅者不可用，处于离线状态。
一个现实生活的例子是Dish电视，它发布不同的渠道，如运动，电影，音乐等，任何人都可以订阅自己的频道集，并获得他们订阅的频道时可用。
Kafka - 图5
queue.put（object） 数据生产
queue.take(object) 数据消费

2.3、JMS核心组件

Destination：消息发送的目的地，也就是前面说的Queue和Topic。
Message ：从字面上就可以看出是被发送的消息。
Producer： 消息的生产者，要发送一个消息，必须通过这个生产者来发送。
MessageConsumer： 与生产者相对应，这是消息的消费者或接收者，通过它来接收一个消息。

Kafka - 图6
通过与ConnectionFactory可以获得一个connection
通过connection可以获得一个session会话。

2.4、常见的类JMS消息服务器

2.4.1、JMS消息服务器 ActiveMQ

ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的。
主要特点：

  • 多种语言和协议编写客户端。语言: Java, C, C++, C#, Ruby, Perl, Python, PHP。应用协议: OpenWire,Stomp REST,WS Notification,XMPP,AMQP
  • 完全支持JMS1.1和J2EE 1.4规范 (持久化,XA消息,事务)
  • 对Spring的支持,ActiveMQ可以很容易内嵌到使用Spring的系统里面去,而且也支持Spring2.0的特性
  • 通过了常见J2EE服务器(如 Geronimo,JBoss 4, GlassFish,WebLogic)的测试,其中通过JCA 1.5 resource adaptors的配置,可以让ActiveMQ可以自动的部署到任何兼容J2EE 1.4 商业服务器上
  • 支持多种传送协议:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA
  • 支持通过JDBC和journal提供高速的消息持久化
  • 从设计上保证了高性能的集群,客户端-服务器,点对点
  • 支持Ajax
  • 支持与Axis的整合

  • 可以很容易得调用内嵌JMS provider,进行测试

    2.4.2、分布式消息中间件 Metamorphosis

    Metamorphosis (MetaQ) 是一个高性能、高可用、可扩展的分布式消息中间件，类似于LinkedIn的Kafka，具有消息存储顺序写、吞吐量大和支持本地和XA事务等特性，适用于大吞吐量、顺序消息、广播和日志数据传输等场景，在淘宝和支付宝有着广泛的应用，现已开源。
    主要特点：

  • 生产者、服务器和消费者都可分布

  • 消息存储顺序写
  • 性能极高,吞吐量大
  • 支持消息顺序
  • 支持本地和XA事务
  • 客户端pull，随机读,利用sendfile系统调用，zero-copy ,批量拉数据
  • 支持消费端事务
  • 支持消息广播模式
  • 支持异步发送消息
  • 支持http协议
  • 支持消息重试和recover
  • 数据迁移、扩容对用户透明
  • 消费状态保存在客户端
  • 支持同步和异步复制两种HA

  • 支持group commit

    2.4.3、分布式消息中间件 RocketMQ

    RocketMQ 是一款分布式、队列模型的消息中间件，具有以下特点：

  • 能够保证严格的消息顺序

  • 提供丰富的消息拉取模式
  • 高效的订阅者水平扩展能力
  • 实时的消息订阅机制
  • 亿级消息堆积能力

Metaq3.0 版本改名，产品名称改为RocketMQ

2.4.4、其他MQ

.NET消息中间件 DotNetMQ
基于HBase的消息队列 HQueue
Go 的 MQ 框架 KiteQ
AMQP消息服务器 RabbitMQ
MemcacheQ 是一个基于 MemcacheDB 的消息队列服务器。

3、为什么需要消息队列（重要）

消息系统的核心作用就是三点：解耦，异步和并行

以用户注册的案列来说明消息系统的作用

3.1、用户注册的一般流程

Kafka - 图7
问题：随着后端流程越来越多，每步流程都需要额外的耗费很多时间，从而会导致用户更长的等待延迟。

3.2、用户注册的并行执行

Kafka - 图8
问题：系统并行的发起了4个请求，4个请求中，如果某一个环节执行1分钟，其他环节再快，用户也需要等待1分钟。如果其中一个环节异常之后，整个服务挂掉了。
Kafka - 图9

3.3、用户注册的最终一致

Kafka - 图10

  1. 保证主流程的正常执行、执行成功之后，发送MQ消息出去。

  2. 需要这个destination的其他系统通过消费数据再执行，最终一致。

Kafka - 图11

4、Kafka组件和说明

Topic ：消息根据Topic进行归类
属于特定类别的消息流称为主题。 数据存储在主题中。主题被拆分成分区。 对于每个主题，Kafka保存一个分区的迷你妈妈。 每个这样的分区包含不可变有序序列的消息。 分区被实现为具有相等大小的一组分段文件。
Producer：发送消息者
生产者是发送给一个或多个Kafka主题的消息的发布者。 生产者向Kafka经纪人发送数据。 每当生产者将消息发布给代理时，代理只需将消息附加到最后一个段文件。实际上，该消息将被附加到分区。 生产者还可以向他们选择的分区发送消息。
Consumer：消息接受者
Consumers从经纪人处读取数据。 消费者订阅一个或多个主题，并通过从代理中提取数据来使用已发布的消息。
broker：代理，每个kafka实例(server)
代理是负责维护发布数据的简单系统。 每个代理可以代理每个主题具有零个或多个分区。
Kafka - 图12
Zookeeper：依赖集群保存meta信息。

Kafka - 图13

Partition（分区）
主题可能有许多分区，因此它可以处理任意数量的数据。
Partition offset（分区偏移）
每个分区消息具有称为 offset 的唯一序列标识。
Replicas of partition（分区备份）
副本只是一个分区的备份。 副本从不读取或写入数据。 它们用于防止数据丢失。
**Kafka Cluster（Kafka集群）

Kafka有多个代理被称为Kafka集群。 可以扩展Kafka集群，无需停机。 这些集群用于管理消息数据的持久性和复制。 | | Leader（领导者） Leader 是负责给定分区的所有读取和写入的节点。 每个分区都有一个服务器充当Leader 。 | | Follower（追随者）
跟随领导者指令的节点被称为Follower。 如果领导失败，一个追随者将自动成为新的领导者。 跟随者作为正常消费者，拉取消息并更新其自己的数据存储。 | | ZooKeeper **ZooKeeper用于管理和协调Kafka代理。 ZooKeeper服务主要用于通知生产者和消费者Kafka系统中存在任何新代理或Kafka系统中代理失败。 根据Zookeeper接收到关于代理的存在或失败的通知，然后产品和消费者采取决定并开始与某些其他代理协调他们的任务。 |

5、Kafka整体结构图

Kafka - 图14

Kafka组成

Kafka - 图15

5.1名词解释和工作方式

  1. Producer ：消息生产者，就是向kafka broker发消息的客户端。

  2. Consumer ：消息消费者，向kafka broker取消息的客户端

  3. Topic ：咋们可以理解为一个队列。

  4. Consumer Group （CG）（消费者组）：这是kafka用来实现一个topic消息的广播（发给所有的consumer）和单播（发给任意一个consumer）的手段。一个topic可以有多个CG。topic的消息会复制（不是真的复制，是概念上的）到所有的CG，但每个partion只会把消息发给该CG中的一个consumer。如果需要实现广播，只要每个consumer有一个独立的CG就可以了。要实现单播只要所有的consumer在同一个CG。用CG还可以将consumer进行自由的分组而不需要多次发送消息到不同的topic。消费组（CG）目的：提高并发消费（消费吞吐量）。

  5. Broker ：一台kafka服务器就是一个broker。一个集群由多个broker组成。一个broker可以容纳多个topic。

  6. Partition：为了实现扩展性，一个非常大的topic可以分布到多个broker（即服务器）上，一个topic可以分为多个partition，每个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id（offset）。kafka只保证按一个partition中的顺序将消息发给consumer，不保证一个topic的整体（多个partition间）的顺序。

  7. Offset：kafka的存储文件都是按照offset.kafka来命名，用offset做名字的好处是方便查找。例如你想找位于2049的位置，只要找到2048.kafka的文件即可。当然the first offset就是00000000000.kafka。

5.2Kafka中zookeeper的作用：

  1. brocker的选举

  2. brocker的动态上下线

  3. 保存元数据信息（topic/partition）

  4. offset 偏移量

Apache Kafka的一个关键依赖是Apache Zookeeper，它是一个分布式配置和同步服务。 Zookeeper是Kafka代理和消费者之间的协调接口。 Kafka服务器通过Zookeeper集群共享信息。 Kafka在Zookeeper中存储基本元数据，例如关于主题，代理，消费者偏移(队列读取器)等的信息。
由于所有关键信息存储在Zookeeper中，并且它通常在其整体上复制此数据（备份），因此Kafka代理/ Zookeeper的故障不会影响Kafka集群的状态。 Kafka将恢复状态，一旦Zookeeper重新启动。 这为Kafka带来了零停机时间。Kafka代理之间的领导者选举也通过使用Zookeeper在领导者失败的情况下完成。
Kafka - 图16

5.3 Consumer与topic的关系

本质上kafka只支持Topic；

  1. 每个group（CG）中可以有多个consumer，每个consumer属于一个consumer group； 通常情况下，一个group中会包含多个consumer，这样不仅可以提高topic中消息的并发消费能力，而且还能提高”故障容错”性，如果group中的某个consumer失效那么其消费的partitions将会有其他consumer自动接管。

  2. 对于Topic中的一条特定的消息，只会被订阅此Topic的每个group中的其中一个consumer消费，此消息不会发送给一个group的多个consumer；那么一个group中所有的consumer将会交错的消费整个Topic，每个group中consumer消息消费互相独立，我们可以认为一个group是一个”订阅”者。

  3. 在kafka中,一个partition中的消息只会被group中的一个consumer消费(同一时刻)；

  4. 一个Topic中的每个partitions，只会被一个”订阅者”中的一个consumer消费，不过一个consumer可以同时消费多个partitions中的消息。

  5. kafka的设计原理决定,对于一个topic，同一个group中不能有多于partitions个数的consumer同时消费，否则将意味着某些consumer将无法得到消息。

kafka只能保证一个partition中的消息被某个consumer消费时是顺序的；事实上，从Topic角度来说,当有多个partitions时,消息仍不是全局有序的。

5.4 Kafka消息的发布

Producer客户端负责消息的分发

  • kafka集群中的任何一个broker都可以向producer提供metadata信息,这些metadata中包含”集群中存活的servers列表”/“partitions leader列表”等信息；

  • 当producer获取到metadata信息之后, producer将会和Topic下所有partition leader保持socket连接；

  • 消息由producer直接通过socket发送到broker，中间不会经过任何”路由层”，事实上，消息被路由到哪个partition上由producer客户端决定；比如可以采用”random””key-hash””轮询”等,如果一个topic中有多个partitions,那么在producer端实现”消息均衡分发”是必要的。

  • 在producer端的配置文件中,开发者可以指定partition路由的方式。

Producer消息发送的应答机制（生产者如何保证数据的完整性？）

设置发送数据是否需要服务端的反馈,有三个值0,1,-1
0: producer不会等待broker发送ack
1: 当leader接收到消息之后发送ack
-1: 当所有的follower都同步消息成功后发送ack
request.required.acks=0
Kafka - 图17

5.4.1Kafka工作流程

Apache Kafka - WorkFlow

到目前为止，我们讨论了Kafka的核心概念。 让我们现在来看一下Kafka的工作流程。
Kafka只是分为一个或多个分区的主题的集合。 Kafka分区是消息的线性有序序列，其中每个消息由它们的索引(称为偏移)来标识。 Kafka集群中的所有数据都是不相连的分区联合。 传入消息写在分区的末尾，消息由消费者顺序读取。 通过将消息复制到不同的代理提供持久性。
Kafka以快速，可靠，持久，容错和零停机的方式提供基于pub-sub和队列的消息系统。 在这两种情况下，生产者只需将消息发送到主题，消费者可以根据自己的需要选择任何一种类型的消息传递系统。 让我们按照下一节中的步骤来了解消费者如何选择他们选择的消息系统。

发布 - 订阅消息的工作流程

以下是Pub-Sub消息的逐步工作流程 -

  1. 生产者定期向主题发送消息。

  2. Kafka代理存储为该特定主题配置的分区中的所有消息。 它确保消息在分区之间平等共享。 如果生产者发送两个消息并且有两个分区，Kafka将在第一分区中存储一个消息，在第二分区中存储第二消息。

  3. 消费者订阅特定主题。

  4. 一旦消费者订阅主题，Kafka将向消费者提供主题的当前偏移，并且还将偏移保存在Zookeeper系综中。

  5. 消费者将定期请求Kafka(如100 Ms)新消息。

  6. 一旦Kafka收到来自生产者的消息，它将这些消息转发给消费者。

  7. 消费者将收到消息并进行处理。

  8. 一旦消息被处理，消费者将向Kafka代理发送确认。

  9. 一旦Kafka收到确认，它将偏移更改为新值，并在Zookeeper中更新它。 由于偏移在Zookeeper中维护，消费者可以正确地读取下一封邮件，即使在服务器暴力期间。

  10. 以上流程将重复，直到消费者停止请求。

  11. 消费者可以随时回退/跳到所需的主题偏移量，并阅读所有后续消息。

点对点消息系统工作流

  1. 生产者以固定间隔向某个主题发送消息。

  2. Kafka存储在为该特定主题配置的分区中的所有消息，类似于前面的方案。

  3. 单个消费者订阅特定主题，假设 Topic-01 为 Group ID 为 Group-1 。

  4. Kafka以与发布 - 订阅消息相同的方式与消费者交互，直到新消费者以相同的组ID 订阅相同主题 Topic-01 1 。

  5. 一旦新消费者到达，Kafka将其操作切换到共享模式，并在两个消费者之间共享数据。 此共享将继续，直到用户数达到为该特定主题配置的分区数。

  6. 一旦消费者的数量超过分区的数量，新消费者将不会接收任何进一步的消息，直到现有消费者取消订阅任何一个消费者。 出现这种情况是因为Kafka中的每个消费者将被分配至少一个分区，并且一旦所有分区被分配给现有消费者，新消费者将必须等待。

5.5 Comsumer的负载均衡

Kafka - 图18
C1 相当于C0，C2相当于C1
当一个group中,有consumer加入或者离开时,会触发partitions均衡.均衡的最终目的,是提升topic的并发消费能力，步骤如下：
1、 假如topic1,具有如下partitions: P0,P1,P2,P3
2、 加入group中,有如下consumer: C1,C2
3、 首先根据partition索引号对partitions排序: P0,P1,P2,P3
4、 根据consumer.id排序: C0,C1
5、 计算倍数: M = [P0,P1,P2,P3].size / [C0,C1].size,本例值M=2(向上取整)
6、 然后依次分配partitions: C0 = [P0,P1],C1=[P2,P3],即Ci = [P(i M),P((i + 1) M -1)]
尽量保证CG中的消费者数量和topic中的parition数量一致，如果消费者太多，会出现闲置的情况。
Kafka - 图19

  • Consumer在消费这个partition里面的message时，顺序读取，维护下offset的信息，一个consume消费一个partition的消息

  • 同一个消费组之间，各个消费者互相排斥，不同消费组之间相互独立，互不影响。

  • 同一个CG的consumer不能处理用一个partition

  • 不同的CG可以处理同一个topic。

5.6 Kafka文件存储机制

5.6.1、Kafka文件存储基本结构

  • 在Kafka文件存储中，同一个topic下有多个不同partition，每个partition为一个目录，partiton命名规则为topic名称+有序序号，第一个partiton序号从0开始，序号最大值为partitions数量减1。

  • 每个partion(目录)相当于一个巨型文件被平均分配到多个大小相等segment(段)数据文件中。但每个段segment file消息数量不一定相等，这种特性方便old segment file快速被删除。默认保留7天的数据。

Kafka - 图20

  • 每个partiton只需要支持顺序读写就行了，segment文件生命周期由服务端配置参数决定。（什么时候创建，什么时候删除）

Kafka - 图21
数据有序的讨论？
一个partition的数据是否是有序的？ 间隔性有序，不连续
针对一个topic里面的数据，只能做到partition内部有序，不能做到全局有序。
特别加入消费者的场景后，如何保证消费者消费的数据全局有序的？伪命题。
只有一种情况下才能保证全局有序？就是只有一个partition。

5.6.2、Kafka Partition Segment

  • Segment file组成：由2大部分组成，分别为index file和data file，此2个文件一一对应，成对出现，后缀”.index”和“.log”分别表示为segment索引文件、数据文件。

Kafka - 图22

  • Segment文件命名规则：partion全局的第一个segment从0开始，后续每个segment文件名为上一个segment文件最后一条消息的offset值。数值最大为64位long大小，19位数字字符长度，没有数字用0填充。

  • 索引文件存储大量元数据，数据文件存储大量消息，索引文件中元数据指向对应数据文件中message的物理偏移地址。

Kafka - 图23

  • 497：当前log文件中的第几条信息，存放在磁盘上的那个地方

上述图中索引文件存储大量元数据，数据文件存储大量消息，索引文件中元数据指向对应数据文件中message的物理偏移地址。
其中以索引文件中元数据3,497为例，依次在数据文件中表示第3个message(在全局partiton表示第368772个message)、以及该消息的物理偏移地址为497。

5.6.3、查找segment file

00000000000000000000.index表示最开始的文件，起始偏移量(offset)为0
00000000000000368769.index的消息量起始偏移量为368770 = 368769 + 1
00000000000000737337.index的起始偏移量为737338=737337 + 1
其他后续文件依次类推。
以起始偏移量命名并排序这些文件，只要根据offset 二分查找文件列表，就可以快速定位到具体文件。当offset=368776时定位到00000000000000368769.index和对应log文件。

5.6.4、通过segment file查找message

当offset=368776时，依次定位到00000000000000368769.index的元数据物理位置和00000000000000368769.log的物理偏移地址
然后再通过00000000000000368769.log顺序查找直到offset=368776为止。

6、Kafka常用操作命令

6.1Kafka启动命令

  1. [root@zhiyou01 kafka_2.12-1.1.0]# ./bin/kafka-server-start.sh -daemon config/server.properties &
  2. 创建topic-test （cdh1上）
  4. ./bin/kafka-topics.sh --create --zookeeper cdh1:2181,Mini1:2181,Mini2:2181 --replication-factor 3 --partitions 3 --topic test
  6. 列出已创建的topic列表（cdh11上）
  7. [root@zhiyou01 kafka_2.12-1.1.0]# ./bin/kafka-topics.sh --list --zookeeper cdh1:2181
  8. test
  9. [root@zhiyou01 kafka_2.12-1.1.0]#
  11. 启动控制台生产者 (cdh1上)
  12. ./bin/kafka-console-producer.sh --broker-list cdh1:9092, Mini1:9092, Mini2:9092 --topic test
  13. [root@zhiyou01 kafka_2.12-1.1.0]# ./bin/kafka-console-producer.sh --broker-list cdh1:9092, Mini1:9092, Mini2:9092 --topic test
  14. 此处可以输入内容
  15. 启动控制台消费者  （其他的节点上面）
  17. [root@zhiyou02 kafka_2.12-1.1.0]# ./bin/kafka-console-consumer.sh --bootstrap-server cdh1:9092 --from-beginning --topic test（新版本使用bootstrap-server）
  18. ./bin/kafka-console-consumer.sh --zookeeper cdh1:2181 --from-beginning --topic test
  20. 删除主题：
  21. ./bin/kafka-topics.sh --delete  --zookeeper localhost:2181  --topic test

6.2kafka基本命令操作

  1. 查看当前服务器中的所有topic 

    bin/kafka-topics.sh --list --zookeeper  cdh1:2181

  2. 创建topic 

    ./kafka-topics.sh --create --zookeeper mini1:2181 --replication-factor 1 --partitions 3 --topic first

  3. 删除topic 

    sh ./bin/kafka-topics.sh --delete --zookeeper cdh1:2181 --topic test
//需要server.properties中设置delete.topic.enable=true否则只是标记删除或者直接重启。

  4. 通过shell命令发送消息 （生产者） 

    kafka-console-producer.sh --broker-list cdh1:9092 --topic first

  5. 通过shell消费消息 

    sh bin/kafka-console-consumer.sh --zookeeper cdh1:2181 --from-beginning --topic test1

  6. 查看消费位置 

    sh kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper cdh1:2181 --group testGroup

  7. 查看某个Topic的详情 

    sh kafka-topics.sh --topic test --describe --zookeeper cdh1:2181

6.3KAFKA启动过程中常见错误

bin/kafka-topics.sh —create —zookeeper cdh1:2181 —replication-factor 3 —partition 3 —topic test
启动kafka时，三个节点上的kafka未全部启动
解决方法：
1.启动三个节点上的kafka,
2.将—replication-factor 3 中的3改为启动的kafka的数量
Kafka - 图24

启动kafka后，创建topic,topic已存在
解决方法：更改topic名字
Kafka - 图25
Kafka - 图26

7、Kafka生产者Java API

Kafka - 图27

8、Kafka消费者Java API

Kafka - 图28