「javakafka队列」java调用kafka接口发送数据

今天给各位分享javakafka队列的知识，其中也会对java调用kafka接口发送数据进行解释，如果能碰巧解决你现在面临的问题，别忘了关注本站，现在开始吧！

本文目录一览：

• 1、【Kafka】队列的思想
• 2、Kafka 设计详解之队列
• 3、Kafka延时队列
• 4、3分钟带你彻底搞懂 Kafka
• 5、kafka 消息队列初识

【Kafka】队列的思想

防止并发写入数据丢失，可以采取队列的模式，即发布订阅者模式。

定义一个并发的list的结构，生产者向list填入数据，消费者从list中取出数据，然后再做业务逻辑处理，这样能保证数据不丢失。

kafka中文教程

Kafka 设计详解之队列

在 上文 中我们介绍了 Kafka 的网络通信，本文打算详细分析 Kafka 的核心 — 队列 的设计和实现，来对 Kafka 进行更深一步的了解。

队列是一种先进先出（FIFO）的数据结构，它是 Kafka 中最重要的部分，负责收集生产者生产的消息，并将这些消息传递给消费者。要实现一个队列有多种方式，Kafka 作为一个消息队列中间件，在设计队列时主要要考虑两个问题：

乍一看到这个问题，我们会想，内存的读取速度远快于磁盘，如果追求性能，内存也充足的话，当然是将生产者产生的消息数据写到内存（比如用一个数组或者链表来存储队列数据），供消费者消费。真的是这样吗？

下面我们依次分析下写内存和写磁盘文件的优缺点，首先，内存的优点是读写速度非常快，但是，如果我们的目标是设计「大数据量」下的「高吞吐量」的消息队列，会有以下几个问题：

接下来我们来分析一下磁盘，写磁盘文件方式存储队列数据的优点就是能规避上述内存的缺点，但其有很严重的缺点，就是读写速度慢，如果纯依靠磁盘，那消息队列肯定做不到「高吞吐量」这个目标。

分析了内存跟磁盘的优缺点，好像我们还是只能选写内存，但我们忽视了磁盘的两个情况：一是磁盘慢是慢在随机读写，如果是顺序读写，他的速度能达到 600MB/sec（RAID-5 磁盘阵列），并不慢，如果我们尽可能地将数据的读写设计成顺序的，可以大大提升性能。二是 现代的操作系统会（尽可能地）将磁盘里的文件进行缓存 。

有了操作系统级别的文件缓存，那用磁盘存储队列数据的方式就变得有优势了。首先，磁盘文件的数据会有文件缓存，所以不必担心随机读写的性能；其次，同样是使用内存，磁盘文件使用的是操作系统级别的内存，相比于在 Java 内存堆中存储队列，它没有 GC 问题，也没有 Java 对象的额外内存开销，更可以规避应用重启后的内存 load 数据耗时的问题，而且，文件缓存是操作系统提供的，因为我们只要简单的写磁盘文件，系统复杂性大大降低。

因此，Kafka 直接使用磁盘来存储消息队列的数据。

刚才我们已经决定用磁盘文件来存储队列数据，那么要如何选择数据结构呢？一般情况下，如果需要查找数据并随机访问，我们会用 B+ 树来存储数据，但其时间复杂度是 O(log N)，由于我们设计的是消息队列，我们可以完全顺序的写收到的生产者消息，消费者消费时，只要记录下消费者当前消费的位置，往后消费就可以了，这样可以对文件尽可能的进行顺序读写，同时，时间复杂度是O(1)。其实，这跟我们写日志的方式很像，每条日志顺序 append 到日志文件。

之前我们已经确定采用直接顺序写磁盘文件的方式来存储队列数据，下面我们来剖析下具体的实现细节。

在 Kafka 中，用一个文件夹存储一条消息队列，成为一个 Log，每条消息队列由多个文件组成，每个文件称为一个 LogSegment，每当一个 LogSegment 的大小到达阈值，系统就会重新生成一个 LogSegment；当旧的 LogSegment 过期需要清理时（虽然磁盘空间相对于内存会宽裕很多，我们可以保存更长时间的消息数据，比如一周，以供消费者更灵活的使用，但还是需要定期清理太老的数据），系统会根据清理策略删除这些文件。

现在我们知道一个队列（Log）是由多个队列段文件（LogSegment）组成的，那么 Kafka 是如何将这些文件逻辑上连接从而组成一条有序队列的呢？在生成每个队列段文件时，Kafka 用该段的初始位移来对其命名，如在新建一个队列时，会初始化第一个队列段文件，那么其文件名就是0，假设每个段的大小是固定值 L，那么第二个段文件名就是 L，第 N 个就是 （N - 1）* L。这样，我们就可以根据文件名对段文件进行排序，排序后的顺序就是整个队列的逻辑顺序。

了解了队列的基本实现，下面我们就来分析下队列的核心操作—读和写。

写操作发生在生产者向队列生产消息时，在上篇文章讲网络通信时我们已经说到，所有的客户端请求会根据协议转到一个 Handler 来具体处理，负责写操作的 Handler 叫 ProducerHandler，整个写请求的流程如下：

之前我们说过，如果是顺序写，由于省掉了磁头寻址的时间，磁盘的性能还是很高的，我们看到 Kakfa 队列是以顺序方式写的，所以性能很高。但是，如果一台 Kafka 服务器有很多个队列，而硬盘的磁头是有限的，所以还是得在不同的队列直接来回切换寻址，性能会有所下降。

队列的读操作发送在消费者消费队列数据时，由于队列是线性的，只需要记录消费者上次消费到了哪里（offset），接下去消费就好了。那么首先会有一个问题，由谁来记消费者到底消费到哪里了？

一般情况下，我们会想到让服务端来记录各个消费者当前的消费位置，当消费者来拉数据，根据记录的消费位置和队列的当前位置，要么返回新的待消费数据，要么返回空。让服务端记录消费位置，当遇到网络异常时会有一些问题，比如服务端将消息发给消费者后，如果网络异常消费者没有收到消息，那么这条消息就被「跳过」了，当然我们可以借鉴二阶段提交的思想，服务端将消息发送给消费者后，标记状态为「已发送」，等消费者消费成功后，返回一个 ack 给服务端，服务端再将其标记为「成功消费」。不过这样设计还是会有一个问题，如果消费者没有返回 ack 给服务端，此时这条消息可能在已经被消费也可能还没被消费，服务端无从得知，只能根据人为策略跳过（可能会漏消息）或者重发（可能存在重复数据）。另一个问题是，如果有很多消费者，服务端需要记录每条消息的每个消费者的消费状态，这在大数据的场景下，非常消耗性能和内存。

Kafka 将每个消费者的消费状态记录在消费者本身（隔一段时间将最新消费状态同步到 zookeeper），每次消费者要拉数据，就给服务端传递一个 offset，告诉服务端从队列的哪个位置开始给我数据，以及一个参数 length，告诉服务端最多给我多大的数据（批量顺序读数据，更高性能），这样就能使服务端的设计复杂度大大降低。当然这解决不了一致性的问题，不过消费者可以根据自己程序特点，更灵活地处理事务。

下面就来分析整个读的流程：

分布式系统中不可避免的会遇到一致性问题，主要是两块：生产者与队列服务端之间的一致性问题、消费者与队列服务端之间的一致性问题，下面依次展开。

当生产者向服务端投递消息时，可能会由于网络或者其他问题失败，如果要保证一致性，需要生产者在失败后重试，不过重试又会导致消息重复的问题，一个解决方案是每个消息给一个唯一的 id，通过服务端的主动去重来避免重复消息的问题，不过这一机制目前 Kafka 还未实现。目前 Kafka 提供配置，供用户不同场景下选择允许漏消息（失败后不重试）还是允许重复消息（失败后重试）。

由于在消费者里我们可以自己控制消费位置，就可以更灵活的进行个性化设计。如果我们在拉取到消息后，先增加 offset，然后再进行消息的后续处理，如果在消息还未处理完消费者就挂掉，就会存在消息遗漏的问题；如果我们在拉取到消息后，先进行消息处理，处理成功后再增加 offset，那么如果消息处理一半消费者挂掉，会存在重复消息的问题。要做到完全一致，最好的办法是将 offset 的存储与消费者放一起，每消费一条数据就将 offset+1。

本文介绍了 Kafka 的队列实现以及其读写过程。Kafka 认为操作系统级别的文件缓存比 Java 的堆内存更省空间和高效，如果生产者消费者之间比较「和谐」的话，大部分的读写操作都会落在文件缓存，且在顺序读写的情况下，硬盘的速度并不慢，因此选择直接写磁盘文件的方式存储队列。在队列的读写过程中，Kafka 尽可能地使用顺序读写，并使用零拷贝来优化性能。最后，Kafka 让消费者自己控制消费位置，提供了更加灵活的数据消费方式。

Kafka延时队列

TimingWheel是kafka时间轮的实现，内部包含了⼀个TimerTaskList数组，每个数组包含了⼀些链表组成的TimerTaskEntry事件，每个TimerTaskList表示时间轮的某⼀格，这⼀格的时间跨度为tickMs，同⼀个TimerTaskList中的事件都是相差在⼀个tickMs跨度内的，整个时间轮的时间跨度为interval = tickMs * wheelSize，该时间轮能处理的时间范围在cuurentTime到currentTime + interval之间的事件。

当添加⼀个时间他的超时时间⼤于整个时间轮的跨度时， expiration = currentTime + interval，则会将该事件向上级传递，上级的tickMs是下级的interval，传递直到某⼀个时间轮满足expiration currentTime + interval，然后计算对应位于哪⼀格，然后将事件放进去，重新设置超时时间，然后放进jdk延迟队列。

SystemTimer会取出queue中的TimerTaskList，根据expiration将currentTime往前推进，然后把⾥⾯所有的事件重新放进时间轮中，因为ct推进了，所以有些事件会在第0格，表示到期了，直接返回。

else if (expiration currentTime + tickMs) {

然后将任务提交到java线程池中处理。

服务端在处理客户端的请求，针对不同的请求，可能不会⽴即返回响应结果给客户端。在处理这类请求时，服务端会为这类请求创建延迟操作对象放⼊延迟缓存队列中。

延迟缓存的数据结构类似MAP，延迟操作对象从延迟缓存队列中完成并移除有两种⽅式：

1，延迟操作对应的外部事件发⽣时，外部事件会尝试完成延迟缓存中的延迟操作 。

2，如果外部事件仍然没有完成延迟操作，超时时间达到后，会强制完成延迟的操作。

DelayedOperation接口表示延迟的操作对象。此接口的实现类包括延迟加⼊，延迟心跳，延迟生产，延迟拉取。

延迟接口相关的方法：

tryComplete：尝试完成，外部事件发⽣时会尝试完成延迟的操作。该⽅法返回值为true，表示可以完成延迟操作，会调⽤强制完成的方法（forceComplete）。返回值为false，表示不可以完成延迟操作。

onComplete：完成的回调方法。

onExpiration：超时的回调方法。

外部事件触发完成和超时完成都会调⽤forceComplete()，并调⽤onComplete()。forceComplete和onComplete只会调⽤⼀次。多线程下⽤原⼦变量来控制只有⼀个线程会调⽤onComplete和forceComplete。

延迟⽣产和延迟拉取完成时的回调⽅法，尝试完成的延迟操作副本管理器在创建延迟操作时，会把回调⽅法传给延迟操作对象。当延迟操作完成时，在onComplete⽅法中会调⽤回调⽅法，返回响应结果给客户端。

创建延迟操作对象需要提供请求对应的元数据。延迟⽣产元数据是分区的⽣产结果；延迟拉取元数据是分区的拉取信息。

创建延迟的⽣产对象之前，将消息集写⼊分区的主副本中，每个分区的⽣产结果会作为延迟⽣产的元数据。创建

延迟的拉取对象之前，从分区的主副本中读取消息集，但并不会使⽤分区的拉取结果作为延迟拉取的元数据，因为延迟⽣产返回给客户端的响应结果可以直接从分区的⽣产结果中获取，⽽延迟的拉取返回给客户端的响应结果不能直接从分区的拉取结果中获取。

元数据包含返回结果的条件是：从创建延迟操作对象到完成延迟操作对象，元数据的含义不变。对于延迟的⽣产，服务端写⼊消息集到主副本返回的结果是确定的。是因为ISR中的备份副本还没有全部发送应答给主副本，才会需要创建延迟的⽣产。服务端在处理备份副本的拉取请求时，不会改变分区的⽣产结果。最后在完成延迟⽣产的操作对象时，服务端就可以把 “创建延迟操作对象” 时传递给它的分区⽣产结果直接返回给⽣产者 。对应延迟的拉取，读取了主副本的本地⽇志，但是因为消息数量不够，才会需要创建延迟的拉取，⽽不⽤分区的拉取结果⽽是⽤分区的拉取信息作为延迟拉取的元数据，是因为在尝试完成延迟拉取操作对象时，会再次读取主副本的本地⽇志，这次的读取有可能会让消息数量达到⾜够或者超时，从⽽完成延迟拉取操作对象。这样创建前和完成时延迟拉取操作对象的返回结果是不同的。但是拉取信息不管读取多少次都是⼀样的。

延迟的⽣产的外部事件是：ISR的所有备份副本发送了拉取请求；备份副本的延迟拉取的外部事件是：追加消息集到主副本；消费者的延迟拉取的外部事件是：增加主副本的最⾼⽔位。

服务端处理⽣产者客户端的⽣产请求，将消息集追加到对应主副本的本地⽇志后，会等待ISR中所有的备份刚本都向主副本发送应答 。⽣产请求包括多个分区的消息集，每个分区都有对应的ISR集合。当所有分区的ISR副本都向对应分区的主副本发送了应答，⽣产请求才能算完成。⽣产请求中虽然有多个分区，但是延迟的⽣产操作对象只会创建⼀个。

判断分区的ISR副本是否都已经向主副本发送了应答，需要检查ISR中所有备份副本的偏移量是否到了延迟⽣产元数据的指定偏移量（延迟⽣产的元数据是分区的⽣产结果中包含有追加消息集到本地⽇志返回下⼀个偏移量）。所以ISR所有副本的偏移量只要等于元数据的偏移量，就表示备份副本向主副本发送了应答。由于当备份副本向主副本发送拉取请求，服务端读取⽇志后，会更新对应备份副本的偏移量数据。所以在具体的实现上，备份副本并不需要真正发送应答给主副本，因为主副本所在消息代理节点的分区对象已经记录了所有副本的信息，所以尝试完成延迟的⽣产时，根据副本的偏移量就可以判断备份副本是否发送了应答。进⽽检查分区是否有⾜够的副本赶上指定偏移量，只需要判断主副本的最⾼⽔位是否等于指定偏移量（最⾼⽔位的值会选择ISR中所有备份副本中最⼩的偏移量来设置，最⼩的值都等于了指定偏移量，那么就代表所有的ISR都发送了应答）。

总结：服务端创建的延迟⽣产操作对象，在尝试完成时根据主副本的最⾼⽔位是否等于延迟⽣产操作对象中元数据的指定偏移量来判断。 具体步骤：

1，服务端处理⽣产者的⽣产请求，写⼊消息集到Leader副本的本地⽇志。

2，服务端返回追加消息集的下⼀个偏移量，并且创建⼀个延迟⽣产操作对象。元数据为分区的⽣产结果（其中就

包含下⼀个偏移量的值）。

3，服务端处理备份副本的拉取请求，⾸先读取主副本的本地⽇志。

4，服务端返回给备份副本读取消息集，并更新备份副本的偏移量。

5，选择ISR备份副本中最⼩的偏移量更新主副本的最⾼⽔位。

6，如果主副本的最⾼⽔位等于指定的下⼀个偏移量的值，就完成延迟的⽣产。

服务端处理消费者或备份副本的拉取请求，如果创建了延迟的拉取操作对象，⼀般都是客户端的消费进度能够⼀直赶上主副本。⽐如备份副本同步主副本的数据，备份副本如果⼀直能赶上主副本，那么主副本有新消息写⼊，备份副本就会⻢上同步。但是针对备份副本已经消费到主副本的最新位置，⽽主副本并没有新消息写⼊时：服务端没有⽴即返回空的拉取结果给备份副本，这时会创建⼀个延迟的拉取操作对象，如果有新的消息写⼊，服务端会等到收集⾜够的消息集后，才返回拉取结果给备份副本，有新的消息写⼊，但是还没有收集到⾜够的消息集，等到延迟操作对象超时后，服务端会读取新写⼊主副本的消息后，返回拉取结果给备份副本（完成延迟的拉取时，服务端还会再读取⼀次主副本的本地⽇志，返回新读取出来的消息集）。

客户端的拉取请求包含多个分区，服务端判断拉取的消息⼤⼩时，会收集拉取请求涉及的所有分区。只要消息的总⼤⼩超过拉取请求设置的最少字节数，就会调⽤forceComplete()⽅法完成延迟的拉取。

外部事件尝试完成延迟的⽣产和拉取操作时的判断条件：

拉取偏移量是指拉取到消息⼤⼩。对于备份副本的延迟拉取，主副本的结束偏移量是它的最新偏移量（LEO）。对于消费者的拉取延迟，主副本的结束偏移量是它的最⾼⽔位（HW）。备份副本要时刻与主副本同步，消费者只能消费到主副本的最⾼⽔位。

客户端的⼀个请求包括多个分区，服务端为每个请求都会创建⼀个延迟操作对象。⽽不是为每个分区创建⼀个延迟操作对象。服务端的“延迟操作缓存”管理了所有的“延迟操作对象”，缓存的键是每⼀个分区，缓存的值是分区对应的延迟操作列表。

⼀个客户端请求对应⼀个延迟操作，⼀个延迟操作对应多个分区。在延迟缓存中，⼀个分区对应多个延迟操作。

延迟缓存中保存了分区到延迟操作的映射关系。

根据分区尝试完成延迟的操作，因为⽣产者和消费者是以分区为最⼩单位来追加消息和消费消息。虽然延迟操作的创建是针对⼀个请求，但是⼀个请求中会有多个分区，在⽣产者追加消息时，⼀个⽣产请求总的不同分区包含的消息是不⼀样的。这样追加到分区对应的主副本的本地⽇志中，有的分区就可以去完成延迟的拉取，但是有的分区有可能还达不到完成延迟拉取操作的条件。同样完成延迟的⽣产也⼀样。所以在延迟缓存中要以分区为键来存储各个延迟操作。

由于⼀个请求创建⼀个延迟操作，⼀个请求⼜会包含多个分区，所以不同的延迟操作可能会有相同的分区。在加⼊到延迟缓存时，每个分区都对应相同的延迟操作。外部事件发⽣时，服务端会以分区为粒度，尝试完成这个分区中的所有延迟操作 。 如果指定分区对应的某个延迟操作可以被完成，那么延迟操作会从这个分区的延迟操作列表中移除。但这个延迟操作还有其他分区，其他分区中已经被完成的延迟操作也需要从延迟缓存中删除。但是不会⽴即被删除，因为分区作为延迟缓存的键，在服务端的数量会很多。只要分区对应的延迟操作完成了⼀个，就要⽴即检查所有分区，对服务端的性能影响⽐较⼤。所以采⽤⼀个清理器，会负责定时地清理所有分区中已经完成的延迟操作。

副本管理器针对⽣产请求和拉取请求都分别有⼀个全局的延迟缓存。⽣产请求对应延迟缓存中存储了延迟的⽣产。拉取请求对应延迟缓存中存储了延迟的拉取。

延迟缓存提供了两个⽅法：

tryCompleteElseWatch()：尝试完成延迟的操作，如果不能完成，将延迟操作加⼊延迟缓存中。⼀旦将延迟操作加⼊延迟缓存的监控，延迟操作的每个分区都会监视该延迟操作。换句话说就是每个分区发⽣了外部事件后，都会去尝试完成延迟操作。

checkAndComplete()：参数是延迟缓存的键，外部事件调⽤该⽅法，根据指定的键尝试完成延迟缓存中的延迟操作。

延迟缓存在调⽤tryCompleteElseWatch⽅法将延迟操作加⼊延迟缓存之前，会先尝试⼀次完成延迟的操作，如果不能完成，会调⽤⽅法将延迟操作加⼊到分区对应的监视器，之后还会尝试完成⼀次延迟操作，如果还不能完成，会将延迟操作加⼊定时器。如果前⾯的加⼊过程中，可以完成延迟操作后，那么就可以不⽤加⼊到其他分区的延迟缓存了。

延迟操作不仅存在于延迟缓存中，还会被定时器监控。定时器的⽬的是在延迟操作超时后，服务端可以强制完成延迟操作返回结果给客户端。延迟缓存的⽬的是让外部事件去尝试完成延迟操作。

延迟缓存的每个键都有⼀个监视器（类似每个分区有⼀个监视器），以链表结构来管理延迟操作。当外部事件发⽣时，会根据给定的键，调⽤这个键的对应监视器的tryCompleteWatch()⽅法，尝试完成监视器中所有的延迟操作。

监视器尝试完成所有延迟操作的过程中，会调⽤每个延迟操作的tryComplete()⽅法，判断能否完成延迟的操作。如果能够完成，就从链表中删除对应的延迟操作。

清理线程的作⽤是清理所有监视器中已经完成的延迟操作。

服务端创建的延迟操作会作为⼀个定时任务，加⼊定时器的延迟队列中。当延迟操作超时后，定时器会将延迟操作从延迟队列中弹出，并调⽤延迟操作的运⾏⽅法，强制完成延迟的操作。

定时器使⽤延迟队列管理服务端创建的所有延迟操作，延迟队列的每个元素是定时任务列表，⼀个定时任务列表可以存放多个定时任务条⽬。服务端创建的延迟操作对象，会先包装成定时任务条⽬，然后加⼊延迟队列指定的⼀个定时任务列表。延迟队列是定时器中保存定时任务列表的全局数据结构，服务端创建的延迟操作不是直接加⼊定时任务列表，⽽是加⼊时间轮。

时间轮和延迟队列的关系：

1，定时器拥有⼀个全局的延迟队列和时间轮，所有时间轮公⽤⼀个计数器。

2，时间轮持有延迟队列的引⽤。

3，定时任务条⽬添加到时间轮对应的时间格（槽）（槽中是定时任务列表）中，并且把该槽表也会加⼊到延迟队列中。

4，⼀个线程会将超时的定时任务列表会从延迟队列的poll⽅法弹出。定时任务列表超时并不⼀定代表定时任务超时，将定时任务重新加⼊时间轮，如果加⼊失败，说明定时任务确实超时，提交给线程池执⾏。

5，延迟队列的poll⽅法只会弹出超时的定时任务列表，队列中的每个元素（定时任务列表）按照超时时间排序，如果第⼀个定时任务列表都没有过期，那么其他定时任务列表也⼀定不会超时。

延迟操作本身的失效时间是客户端请求设置的，延迟队列的元素（每个定时任务列表）也有失效时间，当定时任务列表中的getDelay()⽅法返回值⼩于等于0，就表示定时任务列表已经过期，需要⽴即执⾏。

如果当前的时间轮放不下加⼊的时间时，就会创建⼀个更⾼层的时间轮。定时器只持有第⼀层的时间轮的引⽤，并不会持有更⾼层的时间轮。因为第⼀层的时间轮会持有第⼆层的时间轮的引⽤，第⼆层会持有第三层的时间轮的引⽤。定时器将定时任务加⼊到当前时间轮，要判断定时任务的失效时间⾸是否在当前时间轮的范围内，如果不在当前时间轮的范围内，则要将定时任务上升到更⾼⼀层的时间轮中。时间轮包含了定时器全局的延迟队列。

时间轮中的变量：tickMs=1：表示⼀格的⻓度是1毫秒；wheelSize=20表示⼀共20格，时间轮的范围就是20毫秒，定时任务的失效时间⼩于等于20毫秒的都会加⼊到这⼀层的时间轮中；interval=tickMs*wheelSize=20，如果需要创建更⾼⼀层的时间轮，那么低⼀层的时间轮的interval的值作为⾼⼀层数据轮的tickMs值；currentTime当前时间轮的当前时间，往前移动时间轮，主要就是更新当前时间轮的当前时间，更新后重新加⼊定时任务条⽬。

3分钟带你彻底搞懂 Kafka

Kafka到底是个啥？用来干嘛的？

官方定义如下：

翻译过来，大致的意思就是，这是一个实时数据处理系统，可以横向扩展，并高可靠！

实时数据处理 ，从名字上看，很好理解，就是将数据进行实时处理，在现在流行的微服务开发中，最常用实时数据处理平台有 RabbitMQ、RocketMQ 等消息中间件。

这些中间件，最大的特点主要有两个：

在早期的 web 应用程序开发中，当请求量突然上来了时候，我们会将要处理的数据推送到一个队列通道中，然后另起一个线程来不断轮训拉取队列中的数据，从而加快程序的运行效率。

但是随着请求量不断的增大，并且队列通道的数据一致处于高负载，在这种情况下，应用程序的内存占用率会非常高，稍有不慎，会出现内存不足，造成程序内存溢出，从而导致服务不可用。

随着业务量的不断扩张，在一个应用程序内，使用这种模式已然无法满足需求，因此之后，就诞生了各种消息中间件，例如 ActiveMQ、RabbitMQ、RocketMQ等中间件。

采用这种模型，本质就是将要推送的数据，不在存放在当前应用程序的内存中，而是将数据存放到另一个专门负责数据处理的应用程序中，从而实现服务解耦。

消息中间件 ：主要的职责就是保证能接受到消息，并将消息存储到磁盘，即使其他服务都挂了，数据也不会丢失，同时还可以对数据消费情况做好监控工作。

应用程序 ：只需要将消息推送到消息中间件，然后启用一个线程来不断从消息中间件中拉取数据，进行消费确认即可！

引入消息中间件之后，整个服务开发会变得更加简单，各负其责。

Kafka 本质其实也是消息中间件的一种，Kafka 出自于 LinkedIn 公司，与 2010 年开源到 github。

LinkedIn 的开发团队，为了解决数据管道问题，起初采用了 ActiveMQ 来进行数据交换，大约是在 2010 年前后，那时的 ActiveMQ 还远远无法满足 LinkedIn 对数据传递系统的要求，经常由于各种缺陷而导致消息阻塞或者服务无法正常访问，为了能够解决这个问题，LinkedIn 决定研发自己的消息传递系统， Kafka 由此诞生 。

在 LinkedIn 公司，Kafka 可以有效地处理每天数十亿条消息的指标和用户活动跟踪，其强大的处理能力，已经被业界所认可，并成为大数据流水线的首选技术。

先来看一张图， 下面这张图就是 kafka 生产与消费的核心架构模型 ！

如果你看不懂这些概念没关系，我会带着大家一起梳理一遍！

简而言之，kafka 本质就是一个消息系统，与大多数的消息系统一样，主要的特点如下：

与 ActiveMQ、RabbitMQ、RocketMQ 不同的地方在于，它有一个**分区 Partition **的概念。

这个分区的意思就是说，如果你创建的 topic 有5个分区，当你一次性向 kafka 中推 1000 条数据时，这 1000 条数据默认会分配到 5 个分区中，其中每个分区存储 200 条数据。

这样做的目的，就是方便消费者从不同的分区拉取数据，假如你启动 5 个线程同时拉取数据，每个线程拉取一个分区，消费速度会非常非常快！

这是 kafka 与其他的消息系统最大的不同！

和其他的中间件一样，kafka 每次发送数据都是向 Leader 分区发送数据，并顺序写入到磁盘，然后 Leader 分区会将数据同步到各个从分区 Follower ，即使主分区挂了，也不会影响服务的正常运行。

那 kafka 是如何将数据写入到对应的分区呢？kafka中有以下几个原则：

与生产者一样，消费者主动的去kafka集群拉取消息时，也是从 Leader 分区去拉取数据。

这里我们需要重点了解一个名词： 消费组 ！

考虑到多个消费者的场景，kafka 在设计的时候，可以由多个消费者组成一个消费组，同一个消费组者的消费者可以消费同一个 topic 下不同分区的数据，同一个分区只会被一个消费组内的某个消费者所消费，防止出现重复消费的问题！

但是不同的组，可以消费同一个分区的数据！

你可以这样理解，一个消费组就是一个客户端，一个客户端可以由很多个消费者组成，以便加快消息的消费能力。

但是，如果一个组下的消费者数量大于分区数量，就会出现很多的消费者闲置。

如果分区数量大于一个组下的消费者数量，会出现一个消费者负责多个分区的消费，会出现消费性能不均衡的情况。

因此，在实际的应用中，建议消费者组的 consumer 的数量与 partition 的数量保持一致！

光说理论可没用，下面我们就以 centos7 为例，介绍一下 kafka 的安装和使用。

kafka 需要 zookeeper 来保存服务实例的元信息，因此在安装 kafka 之前，我们需要先安装 zookeeper。

zookeeper 安装环境依赖于 jdk，因此我们需要事先安装 jdk

下载zookeeper，并解压文件包

创建数据、日志目录

配置zookeeper

重新配置 dataDir 和 dataLogDir 的存储路径

最后，启动 Zookeeper 服务

到官网 下载想要的版本，我这里下载是最新稳定版 2.8.0 。

按需修改配置文件 server.properties （可选）

server.properties 文件内容如下：

其中有四个重要的参数：

可根据自己需求修改对应的配置！

启动 kafka 服务

创建一个名为 testTopic 的主题，它只包含一个分区，只有一个副本：

运行 list topic 命令，可以看到该主题。

输出内容：

Kafka 附带一个命令行客户端，它将从文件或标准输入中获取输入，并将其作为消息发送到 Kafka 集群。默认情况下，每行将作为单独的消息发送。

运行生产者，然后在控制台中键入一些消息以发送到服务器。

输入两条内容并回车：

Kafka 还有一个命令行使用者，它会将消息转储到标准输出。

输出结果如下：

本文主要围绕 kafka 的架构模型和安装环境做了一些初步的介绍，难免会有理解不对的地方，欢迎网友批评、吐槽。

由于篇幅原因，会在下期文章中详细介绍 java 环境下 kafka 应用场景！

kafka 消息队列初识

添加消息的依赖

dependency

groupIdorg.springframework.kafka/groupId

artifactIdspring-kafka/artifactId

/dependency

dependency

groupIdio.projectreactor.kafka/groupId

artifactIdreactor-kafka/artifactId

version1.1.0.RELEASE/version

/dependency

信息队列通信的模式---信息中间件

1》点对点模式

   点对点模式通常是基于拉取或者轮询的消息传送模式，（pull）主动的

特点：发送到队列的消息被一个且只有一个消费者进行处理。生产者将消息放入消息队列后，由消费者主动的去拉去消息进行消费。

优点：消费者拉取消息的频率可以由自己控制，消费者端需要额外的线程去监控消息是否消费。

2》发布订阅模式

发布订阅模式是一个基于消息送的消息传送模式，该模式可以由多种不同的订阅者，(push)被动接受

特点：生产者将消息放入消息队列后，队列会将消息推送给订阅过该类消息的消费者，消息队列无法感知消费者消费的速度，消息的处理速度不同，会出现自己的浪费问题

Kafka 消息：

概念：高吞吐量的分布式发布订阅消息系统，

特点：处理消费者规模的网站中的所有动作流数据，具有高性能，持久化，多副本备份，横向扩展能力，，，等

Producer :生产者，消息的入口，

kafka cluster (kafka 群)：

Broker：kafka的实例，每个服务器上有一个或多个kafka 实例，多个broker 在kafka集群中是不重复存在的，会被定义不同不得编号并使用。

Topic：消息的主题，对消息进行分类，kafka数据保存在Topic中，而且，Broker 可以创建多个Topic

Partition:Topic的分区。其作用：处理负载，提高kafka 的吞吐量，同一个Topic 的不同分区数据不重复。Partition相当于一个个的文件夹。

Replication：分区有多个副本，其作用是为了做备胎。当主分区（Leader）c出现故障时，会选择一个备胎（Follower）上位,，成为Leader，在Kafka中默认最大的副本量为10个，且副本的数量不能大于Broker 的数量，follower和Leader 绝对不能在同一个机器中，同一机器对同一分区只能存放一个副本（包括自己）

Message：发送的消息主体

Consumer：消费者，消息的出口，消费方。

Consumer Group：多个消费者组成一个消费组，同一分区的数据，被消费组中一个消费者消费，这样提高了kafka的吞吐量

Zookeeper：kafka集群依赖zookeeper，用来保存集群的元信息，确保系统的可用性。

重点来啦：

工作流程分析：

注意：消息写入leader后，follower 是主动的去 leader进行同步操作的，Producer采用push的方式将数据发布到broker 中，并对每条消息追加到分区中，依次顺序写入磁盘中，以此保证统一分区内的数据是有序的，

如上图：数据在不同的分区中，可能会问没为什么要分区，目的又是什么，

分区的主要目的：1》方便扩展，topic和partition是一对多的关系，可以使用扩展机器，轻松扩展数据量。

                           2》提高并发，以partition 为基本的读写单位，多个消费者可以同时消费数据，提高消费的处理效率。

同时也会有疑问：一个topic 有多个partition，producer 如何确认数据发向哪个partition ?

z在此提下kafka的几大原则：

1：partition 在写入的时候可以选择需要写入的partition，有指定，则写入指定的partition。

2：若没有指定，但是呢，设置可数据的key 值，则会根据key值hash 出一个partition，

3：若没有partition ，也没有key ,则会根据轮询选出一个partition 。

保证信息不丢失是一个消息队列中间件的基本保证，但如何确保呢？-----通过ACK应答机制，

在生产者向队列写入数据的时候，可以设置参数来确定是否确认kafka接收到数据，这个参数可以设置的值为0，1，all。

0：producer 往集群发送数据不需要等到集群的返回，不确保消息的发送成功。其安全性最低，但效率最高。

1：producer往集群发送数据只要leader应答就可以发送一条，只确保leader发送成功。

all：producer 往集群发送数据所有的follower 都完成从leader 的同步发送下一条，确保leader 发送成功何所有副本都完成备份，安全性高，但效率最低。

kafka的应用场景：

主要解决应用解耦，一步消息，流量消费等，实现高性能，高可用，可伸缩和最终一致性架构。存储转发

目前使用较多的消息队列，有Active MQ。BabbitMQ,ZeroMQ，kafka ,MetaMQ，rocket MQ

应用场景：

1：网站活动追踪，对消息进行实时处理，实时监测，离线处理数据仓库，或加载到Hadoop上

2：指标 用于监测数据，分布式应用程序生成的统计数据集中聚合

3：日志聚合：使用kafka代替一个日志聚合的解决方案

4：流处理：基于单个topic主题的实时数据流，

5：实践采集：是一种应用程序的设计风格，kafka支持非常大的存储日志数据的场景

6：提交日志：kafka可以作为一种分布式的外部提交日志，日志帮助节点之间复制数据。并作为失败的节点来恢复实践重新同步，kafka的日志压缩功能能制止这个用法。

k

关于javakafka队列和java调用kafka接口发送数据的介绍到此就结束了，不知道你从中找到你需要的信息了吗 ？如果你还想了解更多这方面的信息，记得收藏关注本站。