websphere mq 查看队列中是否有数据_全网最全的 “消息队列”

消息队列的使用场景

以下介绍消息队列在实际应用常用的使用场景。异步处理、应用解耦、流量削锋和消息通讯四个场景。

1】异步处理：场景说明：用户注册后，需要发注册邮件和注册短信。

引入消息队列后架构如下：用户的响应时间=注册信息写入数据库的时间，例如50毫秒。发注册邮箱、发注册短信写入消息队列后，直接返回客户端，因写入消息队列的速度很快，基本可以忽略，因此用户的响应时间可能是50毫秒。按照传统的做法：
①、串行方式，将注册信息写入数据库成功后，发注册邮件，再发送注册短信，以上三个成功后，返回客户端。可能需要150毫秒，这样使用消息队列提高了3倍。
②、并行方式，将注册信息写入数据库成功后，发送注册邮件，同时发送注册短信。也可能需要100毫秒，这样使用消息队列提高了2倍。
2】应用解耦：场景说明：用户下单后，订单系统需要通知库存系统。如下图：

传统模式的缺点：①、库存系统无法访问时，则订单减库存业务将会失败，从而导致订单失败；②、订单系统与库存系统耦合；

引入消息队列：①、用户下单后，订单系统完成持久化处理，将消息写入消息队列，返回用户订单下单成功。②、库存系统：订阅下单的消息，采用拉/推的方式，获取下单信息，库存系统根据下单信息，进行库存操作。

☛ 当库存系统不能正常使用时，也不会影响正常下单，因为下单后，订单系统写入消息队列就不再关心其他的后续操作了。实现订单系统与库存系统的解耦。
3】流量削锋：场景说明：秒杀或团抢活动中使用广泛。秒杀活动，一般会因为流量过大，导致流量暴增，应用挂掉。一般需要在应用前端加入消息队列。用户请求：服务器接受后，首先写入消息队列。当消息队列长度超出最大数量，则直接抛弃用户请求或跳转至错误页面。秒杀业务处理：根据消息队列中的请求信息，再做后续处理。
▁▂▃ 这样可以有效的控制活动人数和有效缓解短时间内的高流量冲击，防止压垮应用系统。
4】日志处理：指将消息队列用在日志处理中，比如 Kafka 的应用，解决大量日志传输的问题。

▷ 日志采集客户端：负责日志数据采集，定时写入 Kafka队列。

▷ kafka消息队列：负责日志数据的接收，存储和转发。

▷ 日志处理应用：订阅并消费 kafka 队列中的日志数据。

5】消息通信：消息队列一般都内置了高效的通信机制，因此也可以用纯消息通信。比如实现点对点消息队列，或者聊天室。

①、点对点通讯：客户端A和客户端B使用同一队列，进行消息通讯
②、聊天室通讯(发布订阅模式)：客户端A，客户端B，客户端N订阅同一主题，进行消息发布和接收。实现类似聊天室效果。

消息中间件的工作流程

1、发送端 MQ-Product (消息生产者)将消息发送给 MQ-server；

2、MQ-server 将消息落地，持久化到数据库等；

3、MQ-server 回 ACK 给 MQ-Producer；

4、MQ-server 将消息发送给消息接收端 MQ-Consumer (消息消费者)；

5、MQ-Consumer 消费接收到消息后发送 ACK 给 MQ-server；

6、MQ-server 将落地消息删除；

消息的重发，补发策略

为了保证消息必达，MQ使用了消息超时、重传、确认机制。使得消息可能被重复发送，当消息生产者收不到 MQ-server 的ACK，重复向 MQ-server发送消息。MQ-server 收不到消息消费者的 ACK，重复向消息消费者发消息。

【消息重发】：【1】如果消息接收者在处理消息过程中没有对MOM(消息中间键)进行应答，则消息将由 MOM重发。

【2】如果队列中设置了预读参数(consumer.perfetchSize)，如果消息接收者在处理第一条消息时(没有向MOM进行确认)就宕机了，则预读数量的所有消息将被重发。

【3】如果 Session 是事务的，则只要消息接收者有一条消息没有确认，或消息发送期间 MOM 或客户端某一方突然宕机了，则该事务范围中的所有消息 MOM 都将重发。

▷ ActiveMQ 消息服务器怎么知道客户端到底是消息正在处理中还是已处理完成没应答MOM或者宕机等等情况？其实是所有的客户端机器，都运行着一套客户端的 ActiveMQ 环境，该环境缓存发来的消息，维持着和 ActiveMQ服务器的消息通讯，负责失效转移(fail-over)等，所有的判断和处理都是由这套客户端环境来完成的。

【补发策略】：前提，Broker 根据自己的规则，通过 BrokerInfo 命令包和客户端建立连接，向客户端传送缺省发送策略(发送：同步和异步，策略：持久化消息和非持久化消息)。但是客户端可以使用 ActiveMQConnect.getRedeliveryPolicy() 方法覆盖该策略设置。

RedeliveryPolicy policy = connection.getRedeliveryPolicy();  policy.setInitialRedeliveryDelay(500);  policy.setBackOffMultiplier(2);  policy.setUseExponentialBackOff(true);  policy.setMaximumRedeliveries(2);

★ 一旦消息重发尝试超过重发策略中配置的 maximumRedeliveries(默认=6)会给 Broker 发送一个“Poison ack”通知它，这个消息被认为是 a poison pill，接着 Broker会将这个消息发送给 DLQ(Dead Letter Queue)，以便后续处理。

【策略】：【1】 缺省死信队列(Dead Letter Queue)叫做Active.DLQ；所有的未送达消息将发送到这个队列，导致非常难于管理。此时就可以通过设置 activemq.xml 文件中的 destination policy map 的 “individualDeadLetterStrategy” 属性来修改。

<broker...>    <destinationPolicy>    <policyMap>      <policyEntries>            <policyEntry queue=">">        <deadLetterStrategy>                <individualDeadLetterStrategy          queuePrefix="DLQ." useQueueForQueueMessages="true" />        deadLetterStrategy>      policyEntry>      policyEntries>    policyMap>    destinationPolicy>    ...  broker>

【2】自动丢弃过期消息(Expired Messages)：一些应用可能只是简单的丢弃过期消息，而不是将它们放到 DLQ。在dead letter strategy死信策略上配置 processExpired 属性为 false，可以实现这个功能。

<broker...>    <destinationPolicy>     <policyMap>     <policyEntries>              <policyEntry queue=">">              <deadLetterStrategy>         <sharedDeadLetterStrategy processExpired="false" />       deadLetterStrategy>       policyEntry>     policyEntries>     policyMap>    destinationPolicy>  ...  broker>

【3】将非持久信息(non-persistent messages)放入死信队列 ActiveMQ 缺省不会将未发送到的非持久信息放入死信队列。如果一个应用程序并不想将消息 message 设置为持久的，那么记录下来的那些未发送到的消息对它来说往往也就没有价值。不过如果想实现这个功能，可以在 dead-letter-strategy 死信策略上设置 processNonPersistent="true"。

<broker...>    <destinationPolicy>     <policyMap>     <policyEntries>              <policyEntry queue=">">              <deadLetterStrategy>         <sharedDeadLetterStrategy processNonPersistent="true" />       deadLetterStrategy>       policyEntry>     policyEntries>     policyMap>    destinationPolicy>  ...  broker>

消息重复发送产生的后果

对于非幂等性的服务而言，如果重复发送消息就会产生严重的问题。譬如：银行取钱，上游支付系统负责给用户扣款，下游系统负责给用户发钱，通过MQ异步通知。不管是上游的ACK丢失，导致 MQ收到重复的消息，还是下半场 ACK丢失，导致系统收到重复的出钱通知，都可能出现，上游扣了一次钱，下游发了多次钱。消息队列的异步操作，通常用于幂等性的服务，非幂等性的服务时不适用中间件进行通信的。更多的是建立长连接 Socket 进行通信的。或者通过如下方式改造。

MQ内部如何做到幂等性的

对于每条消息，MQ内部生成一个全局唯一、与业务无关的消息ID：inner-msg-id。当 MQ-server 接收到消息时，先根据 inner-msg-id 判断消息是否重复发送，再决定是否将消息落地到 DB中。这样，有了这个 inner-msg-id 作为去重的依据就能保证一条消息只能一次落地到 DB。

消息消费者应当如何做到幂等性

【1】对于非幂等性业务且要求实现幂等性业务：生成一个唯一ID标记每一条消息，将消息处理成功和去重日志通过事物的形式写入去重表。
【2】对于非幂等性业务可不实现幂等性的业务：权衡去重所花的代价决定是否需要实现幂等性，如：购物会员卡成功，向用户发送通知短信，发送一次或者多次影响不大。不做幂等性可以省掉写去重日志的操作。

如何保证消息的有序性

【Active 中有两种方式保证消息消费的顺序性】：【1】通过高级特性 consumer 独有的消费者(exclusive consumer)。如果一个 queue 设置为 exclusive，broker 会挑选一个 consumer，并且将所有的消息都发给这个 consumer。如果这个 consumer挂了，broker 会自动挑选另外一个 consumer。

queue = new ActiveMQQueue("TEST.QUEUE?consumer.exclusive=true"); consumer = session.createConsumer(queue);

【2】利用 Activemq 的高级特性：MessageGroups。Message Groups 特性是一种负载均衡的机制。在一个消息被分发到consumer 之前，broker 首先检查消息 JMSXGroupID 属性。如果存在，那么 broker 会检查是否有某个 consumer 拥有这个message group。如果没有，那么 broker 会选择一个 consumer，并将它关联到这个 message group。此后，这个 consumer 会接收这个 message group 的所有消息，直到：
①、Consumer 被关闭。
②、Message group 被关闭，通过发送一个消息，并设置这个消息的 JMSXGroupSeq 为 -1。

消费者实际上根据两个维度排序了，一个是消费者的 Priority，即消费者的优先级。还有一个是消费者的指定的消息组的个数 AssignedGroupCount。这个顺序直接影响到下一条消息是谁来接收。

protected boolean assignMessageGroup(Subscription subscription, QueueMessageReference node) throws Exception {  boolean result = true;  // 保持消息组在一起。  String groupId = node.getGroupID();  int sequence = node.getGroupSequence();  if (groupId != null) {    // 先查找该queue存储的一个groupId，和consumerId的一个map    MessageGroupMap messageGroupOwners = getMessageGroupOwners();    // 如果是该组的第一条消息。则指定该consumer消费该消息组    if (sequence == 1) {      assignGroup(subscription, messageGroupOwners, node, groupId);    } else {      // 确保前一个所有者仍然有效，否则就生成新的主人。      ConsumerId groupOwner;      groupOwner = messageGroupOwners.get(groupId);      if (groupOwner == null) {        assignGroup(subscription, messageGroupOwners, node, groupId);      } else {        if (groupOwner.equals(subscription.getConsumerInfo().getConsumerId())) {          // 一个组中的 sequence < 1 表示改组消息已经消费完了          if (sequence < 0) {            messageGroupOwners.removeGroup(groupId);            subscription.getConsumerInfo().decrementAssignedGroupCount(destination);          }        } else {          // 说明该消费者不能消费该消息组          result = false;        }      }    }  }  return result;}

【RabbitMQ 保证消息队列的顺序性】：造成顺序错乱的场景：RabbitMQ 中有一个 Queue，多个 Consumer。生产者向 RabbitMQ 里发送了三条数据，顺序依次是 data1、data2、data3，放入RabbitMQ 的一个内存队列。有三个消费者分别从 MQ 中消费这三条数据中的一条，可能消费者2先执行完操作，把 data2 存入数据库，然后是 data1、data3。导致顺序错乱。

【解决方案】：RabbitMQ 将上面的一个 Queue 拆分为三个 Queue，每个 Queue 对应一个 Consumer，就是多一些 Queue 而已，确实是麻烦点；然后这个 Consumer 内部用内存队列做排队，然后分发给底层不同的 worker 来处理。如下，将消息放入一个队列，由一个消费者消费即可保证顺序。

【Kafka 保证消息队列的顺序性】: 建了一个 Topic，有三个 Partition。生产者在写的时候，其实可以指定一个 key，比如说指定了某个订单 id 作为 key，那么这个订单相关的数据，一定会被分发到同一个 Partition 中去，而且这个 Partition 中的数据一定是有顺序的。消费者从 Partition 中取出来数据的时候，也一定是有顺序的。接着，消费者里可能会搞多个线程来并发处理消息。因为如果消费者用单线程时，处理比较耗时。而多线程并发处理时，顺序可能就乱序。
【解决方案】：①、一个 topic，一个 partition，一个 consumer，内部单线程消费，单线程吞吐量太低，一般不会用这个。
②、写 N 个内存 queue，具有相同 key 的数据都到同一个内存 queue；然后对于 N 个线程，每个线程分别消费一个内存 queue 即可，这样就能保证顺序性。

用过哪些MQ，和其他 MQ比较有什么优缺点

【1】Kafka 是 LinkedIn 开发的一个高性能、分布式的消息系统，广泛用于日志收集、流式数据处理、在线和离线消息分发等场景。虽然不是作为传统的 MQ来设计，但在大部分情况下，Kafka 也可以代替原有 ActiveMQ 等传统的消息系统。
【2】Kafka 将消息流按 Topic 组织，保存消息的服务器称为 Broker，消费者可以订阅一个或者多个 Topic。为了均衡负载，一个Topic 的消息又可以划分到多个分区(Partition)，分区越多，Kafka 并行能力和吞吐量越高。
【3】Kafka 集群需要 Zookeeper 支持来实现集群，Kafka 发行包中已经包含了 Zookeeper，部署的时候可以在一台服务器上同时启动一个 Zookeeper Server 和一个 Kafka Server，也可以使用已有的其他 Zookeeper 集群。
【4】和传统的 MQ 不同，消费者需要自己保留一个 offset，从 Kafka 获取消息时，只拉取当前 offset 以后的消息。Kafka 的scala/java 版的 Client 已经实现了这部分的逻辑，将 offset 保存到 zookeeper 上。每个消费者可以选择一个 id，同样 id 的消费者对于同一条消息只会收到一次。一个 Topic 的消费者如果都使用相同的id，就是传统的 Queue；如果每个消费者都使用不同的id，就是传统的 pub-sub。

【如果在 MQ 的场景下，将 Kafka 和 ActiveMQ 相比，Kafka 的优点】：
【1】分布式、高可扩展：Kafka 集群可以透明的扩展，增加新的服务器进集群。
【2】高性能：Kafka 的性能大大超过传统的 ActiveMQ、RabbitMQ 等 MQ 实现，尤其是 Kafka 还支持 batch 操作。
【3】容错：Kafka 每个 Partition 的数据都会复制到几台服务器上。当某个 Broker 故障失效时，ZooKeeper 服务将通知生产者和消费者，生产者和消费者转而使用其它 Broker。
【4】高吞吐：在一台普通的服务器上既可以达到 10W/s 的吞吐速率。
【5】完全的分布式系统：Broker、Producer、Consumer都原生自动支持分布式，自动实现负载均衡。
【6】快速持久化：可以在 O(1) 的系统开销下进行消息持久化。
【7】游标位置：ActiveMQ 游标由 AMQ来管理，无法读取历史数据。Kafka 客户端自己管理游标，可以重读数据。

【Kafka 的缺点】：
【1】重复消息：Kafka 只保证每个消息至少会送达一次，虽然几率很小，但一条消息有可能会被送达多次。
【2】消息乱序：虽然一个 Partition 内部的消息是保证有序的，但是如果一个Topic 有多个Partition，Partition 之间的消息送达不保证有序。
【3】复杂性：Kafka 需要 zookeeper 集群的支持，Topic 通常需要人工来创建，部署和维护较一般消息队列成本更高。
☞ MQ 是非线程安全的【Kafka 架构】：【1】Producers(生产者)：生产者是发送一个或多个主题 Topic 的发布者。生产者向 Kafka 代理发送数据。每当生产者将消息发布给代理时，代理只需要将消息附加到最后一个段文件。实际上，该消息将被附加到分区。生产者也可以向指定的分区发送消息。

【2】Brokers：代理(经纪人)负责维护发布数据的简单系统。
【3】Topic：主题属于特定类别的信息流称为主题。数组存储在主题中。Topic 相当于 Queue。主题被拆分成分区。分区被实现为具有大小相等的一组分段文件。
【4】Partition(分区)：每个 Partition 内部消息有序，其中每个消息都有一个 offset 序号。一个 Partition 值对应一个 Broker，一个 Broker 可以管理多个 Partition。
【5】Segment：Partition 物理上由多个 Segment组成。每个 Partion 目录相当于一个巨型文件被平均分配到多个大小相等segment 段数据文件中。但每个段 segment file消息数量不一定相等
【6】Partition offset(分区偏移)：每个 Partition 都由一系列有序的、不可变的消息组成，这些消息被连续的追加到 Partition中。Partition 中的每个消息都有一个连续的序列号叫做 offset，用于 Partition唯一标识一条消息。
【7】Replicas of partition(分区备份)：副本只是一个分区备份：不读取和写入数据，主要用于防止数据丢失。
【8】Kafka Cluster(Kafka 集群)：Kafka 有多个代理被称为 Kafka集群。可以扩展 Kafka集群，无需停机。这些集群用于管理消息数据的持久性和复制。
【9】Consumers(消费者)：Consumers 从 MQ读取数据。消费者订阅一个或多个主题，并通过从代理中提取数据来使用已发布的消息。Consumer 自己维护消费到哪个 offset。
【每个Consumer 都有对应的 group】：【1】group 内是 queue 消费模型：各个 Consumer 消费不同的 Partition，因此一个消息在 group 内只消费一次。
【2】group 间是 publish-subscribe 消费模型：各个 group 各自独立消费，互不影响，因此一个消息被每个 group 消费一次。

MQ 系统的数据如何保证不丢失

【Producer 数据丢失的原因】：【1】使用同步模式的时候，有 3种状态保证消息被安全生产，当配置 ack=1时(只保证写入Leader成功)的话，如果刚好 Leader partition 挂了，数据就会丢失。

ack 机制：broker 表示发来的数据已确认接收无误，表示数据已经保存到磁盘。
0：不等待 broker 返回确认消息
1：等待 topic 中某个 partition leader 保存成功的状态反馈
-1/all：等待 topic 中某个 partition 所有副本都保存成功的状态反馈

【2】使用异步模式时，当缓冲区满了，如果配置=0(还没有收到确认的数据，数据就立即被丢弃掉)。
【解决办法】：只要能避免以上两种情况就可以保证消息不会被丢失。如下：
【1】当同步模式时，确认机制设置为-1，就是让消息写入 Leader 和所有副本。
【2】当异步模式时，消息发出，还没收到确认的时候，缓冲区也满了。在配置文件中设置成不限制阻塞超时的时间，也就是说让生产者一直阻塞，这样就能保证数据不会丢失。

producer.type = asyncrequest.required.acks=1queue.buffering.max.ms=5000  #异步发送的时候 发送时间间隔 单位是毫秒queue.buffering.max.messages=10000queue.enqueue.timeout.ms = -1batch.num.messages=200 #异步发送 每次批量发送的条目

【Kafka弄丢了数据】：Kafka 的某个 Broker宕机了，然后重新选举Broker 上的 Partition 的 Leader时。如果此时 Follower还没来得及同步数据，Leader就挂了，然后某个 Follower成为了 Leader，他就少了一部分数据。
【解决办法】：一般要求设置 4个参数来保证消息不丢失：
【1】给 Topic设置 replication.factor 参数：这个值必须大于1，表示要求每个 Partition必须至少有2个副本。
【2】在 Kafka服务端设置 min.isync.replicas参数：这个值必须大于1，表示要求一个 Leader至少感知到有至少一个 Follower在跟自己保持联系正常同步数据，这样才能保证 Leader挂了之后还有一个 Follower。
【3】在生产者端设置 acks= -1：要求每条数据，必须是写入所有 Replica 副本之后，才能认为是写入成功了。
【4】在生产者端设置 retries=MAX(很大的一个值，表示无限重试)：表示消息一旦写入事变，就无限重试。
【Consumer 数据丢失的原因】：当你消费到了这个消息，然后消费者那边自动提交了offset，让 kafka 以为你已经消费好了这个消息，其实你刚准备处理这个消息，你还没处理，你自己就挂了，此时这条消息就丢了。
【解决办法】：【1】Kafka 会自动提交 offset，使用 Kafka高级API，如果将自动提交 offset 改为手动提交(当数据入库之后进行偏移量的更新)，就可以保证数据不会丢。但是可能导致重复消费，比如你刚处理完，还没有提交 offset，结果自己挂了，此时肯定会重复消费一次，自己保证幂等性就好了。

RabbitMQ 如何实现集群高可用

【镜像模式】：队列的数据都镜像了一份到所有的节点上。这样任何一个节点失效，不会影响到整个集群的使用。在实现上 mirror queue 内部有一套选举算法，会选出一个 master 和若干的 slaver。master 和 slaver 通过相互之间不断的发送心跳来检查是否连接断开。可以通过指定 net_ticktime 来控制心跳检查频率。注意一个单位时间 net_ticktime 实际上做了4次交互，故当超过net_ticktime (± 25%) 秒没有响应的话则认为节点挂掉。另外注意修改 net_ticktime 时需要所有节点都一致。配置举例：

{rabbit, [{tcp_listeners, [5672]}]},{kernel, [{net_ticktime,  120}]}

【Consumer】：任意连接一个节点，若连上的不是 Master，请求会转发给 Master，为了保证消息的可靠性，Consumer 回复 Ack 给 Master 后，Master 删除消息并广播所有的 Slaver 去删除；
【Publisher】：任意连接一个节点，若连上的不是 Master，则转发给 Master，由 Master存储并转发给其他的 Slaver存储；
【如果 Slaver 挂掉】：则集群的节点状态没有任何变化。只要 Client 没有连到这个节点上，也不会给 Client 发送失败的通知。在检测到 Slaver 挂掉的期间 Publish 消息会有延迟。如果配置了高可用策略是自动同步，当 Slaver 起来后，队列中有大量的消息需要同步，将会整个集群阻塞长时间的不能读写直到同步结束；
【RabbitMQ 实现了一种镜像队列(mirrored queue)的算法提供HA】：创建队列时可以通过传入“x-ha-policy”参数设置队列为镜像队列，镜像队列会存储在多个 Rabbit MQ 节点上，并配置成一主多从的结构，可以通过“x-ha-policy-params”参数来具体指定master 节点和 slave节点的列表。所有发送到镜像队列上的操作，比如消息的发送和删除，都会先在 master节点上执行，再通过一种叫 GM(Guaranteed Multicast)的原子广播(atomic broadcast)算法同步到各 slave节点。GM算法通过两阶段的提交，可以保证 master节点发送到所有 slave节点上的消息要么全部执行成功，要么全部失败；通过环形的消息发送顺序，即 master节点发送消息给一个 slave节点，这个 slave节点依次发送给下一个 slave节点，最终消息回到 master节点，保证了主从节点上的负载差别不大。通过传入“x-ha-policy”参数设置队列为镜像队列(mirrored queue)：定义一个policy：以“ha.”开头的队列都被镜像到集群中的所有节点上：rabbitmqctl set_policy ha-all "^ha\." '{"ha-mode":"all"}'。定义一个policy：以“cinder”开头的队列被镜像到集群中的任意两个节点上，并且自动同步：rabbitmqctl set_policy ha-cinder-two "^cinder"或者设置'{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'；
▶ all：队列将 mirrored 到所有集群中的节点中，当新节点添加进来时也会 mirrored 到新的节点；
▶ exactly(需指定count)：如果节点数小于 count 数，则队列将 mirrored 到所有的节点。如果节点数大于 count，新的节点将不再创建队列的 mirror(即使原来已创建 mirror 的节点挂掉也不会创建)；
▶ nodes：对指定的节点进行 mirror。如果没有一个指定的节点在运行中，那么只有 client 连接的那个节点才会声明 queue(这里有个迁移策略：假如 queue是在[A,B]上且A为 master，若给定的新的策略为nodes[C,D]，那么为了防止数据丢失，在迁移中会同时存在[A,C,D]直到C,D已经同步好以后，A才会关闭)；

Kafka 吞吐量高的原因

【1】顺序读写磁盘，充分利用了操作系统的预读机制。
【2】Linux 中使用 sendfile 命令，减少一次数据拷贝：
   ①、把数据从硬盘读取到内核中的页缓存。
   ②、把数据从内核中读取到用户空间(sendfile 命令跳过此步骤)。
   ③、把用户空间的数据写到 socket 缓存区中。
   ④、操作系统将数据从 socket 缓冲区中复制到网卡缓冲区，以便将数据经网络发出。
【3】生产者缓存消息批量发送，消费者批量从 broker 获取消息，减少 IO 次数，充分利用磁盘顺序读写的性能。
【4】通常情况下 Kafka 的瓶颈不是 CPU或者磁盘，而是网络宽带，所以生产者可以对数据进行压缩。

Kafka 和其他消息队列的区别

【与 RabbitMQ 的区别】：ღ RabbitMQ：用在实时的对可靠性要求比较高的消息传递上。kafka：用于处理活跃的流式数据，大数据量的数据处理上。
【1】在架构模型方面：RabbitMQ 遵循 AMQP 协议，RabbitMQ 的 Broker由 Exchange、Binding、Queue 组成，其中 Exchange 和 Binding 组成了消息的路由键；Producer 通过连接 Channel 和 Server 进行通信，Consumer 从 Queue 获取消息进行消费(长连接，queue 有消息会推送到 consumer端，consumer 循环从输入流读取数据)。rabbitMQ 以 Broker为中心；有消息的确认机制。
♐ kafka 遵从一般的MQ结构，Producer，Broker，Consumer，以 Consumer为中心，消费信息保存的客户端 Consumer上，Consumer根据消费的点，从 Broker上批量 pull数据，无消息确认机制。
【2】在吞吐量方面：RabbitMQ在吞吐量方面稍逊于Kafka，他们的出发点不一样，RabbitMQ支持对消息的可靠的传递，支持事务，不支持批量的操作；基于存储的可靠性的要求存储可以采用内存或者硬盘。
♐ kafka具有高的吞吐量，内部采用消息的批量处理，zero-copy(sendfile 函数) 机制，数据的存储和获取是本地磁盘顺序批量操作，具有O(1)的复杂度，消息处理的效率很高。
【3】在可用性方面：RabbitMQ 支持 mirror 的 queue，主 queue失效，mirror queue接管。
♐ Kafka 的 Broker支持主备模式。
【4】在集群负载均衡方面：RabbitMQ 的负载均衡需要单独的 loadbalancer 进行支持。
♐ Kafka 采用 Zookeeper对集群中的 Broker、Consumer进行管理，可以注册 Topic 到Zookeeper上；通过 Zookeeper的协调机制，Producer 保存对应 Topic的 Broker信息，可以随机或者轮询发送到 Broker上；并且 Producer可以基于语义指定分片，消息发送到 Broker的某分片上。

【与 ActiveMQ 的区别】：ActiveMQ 和 Kafka，前者完全实现了 JMS 的规范，后者并没有纠结于JMS规范，设计了另一套吞吐非常高的分布式发布-订阅消息系统，非常流行。目前归属于 Apache 定级项目。它只用文件系统来管理消息的生命周期。接下来我们结合三个点(消息安全性，服务器的稳定容错性以及吞吐量)来分别谈谈这两个消息中间件。
【1】消息的安全性：Kafka 集群中的 Leader 负责某一 Topic 的某一 Partition 的消息的读写，理论上 Consumer 和 Producer 只与该 Leader 节点打交道，一个集群里的某一 Broker 即是 Leader 的同时也可以担当某一 Partition 的 Follower，即 Replica。Kafka 分配 Replica 的算法如下：
(1)将所有 Broker(假设共n个Broker)和待分配的 Partition排序。
(2)将第i个 Partition分配到第(i mod n)个 Broker上。
(3)将第i个 Partition的第j个 Replica分配到第((i + j) mod n)个 Broker上。
同时，Kafka 与 Replica 既非同步也不是严格意义上的异步。一个典型的 Kafka 发送-消费消息的过程如下：首先 Producer消息发送给某 Topic 的某 Partition 的 Leader，Leader 先是将消息写入本地 Log，同时 follower(如果落后过多将会被踢出 Replica列表)从Leader上 pull 消息，并且在未写入 log 的同时即向 Leader 发送 ACK 的反馈，所以对于某一条已经算作 commit 的消息来讲，在某一时刻，其存在于 Leader的 log中，以及 Replica的内存中。这可以算作一个危险的情况(听起来吓人)，因为如果此时集群挂了这条消息就算丢失了，但结合 producer的属性(request.required.acks=-1，当所有follower都收到消息后返回ack)可以保证在绝大多数情况下消息的安全性。当消息算作 commit的时候才会暴露给 consumer，并保证 at-least-once的投递原则。
【2】服务的稳定容错性：前面提到过，Kafka天然支持HA，整个 leader/follower 机制通过 zookeeper调度，它在所有 Broker中选出一个 controller，所有 Partition的 Leader选举都由 controller决定，同时 controller也负责增删 Topic以及 Replica的重新分配。如果Leader挂了，集群将在ISR(in-sync replicas)中选出新的Leader，选举基本原则是：新的 Leader必须拥有原来的 Leader commit 过的所有消息。假如所有的 follower都挂了，Kafka会选择第一个“活”过来的 Replica(不一定是ISR中的)作为 Leader，因为如果此时等待 ISR中的 Replica是有风险的，假如所有的ISR都无法“活”，那此 Partition将会变成不可用。
【3】吞吐量：Leader 节点负责某一 Topic(可以分成多个 Partition)的某一 Partition的消息的读写，任何发布到此 Partition的消息都会被直接追加到 log文件的尾部，因为每条消息都被 append 到该 Partition中，是顺序写磁盘，因此效率非常高(经验证，顺序写磁盘效率比随机写内存还要高，这是 Kafka高吞吐率的一个很重要的保证)，同时通过合理的 Partition，消息可以均匀的分布在不同的 Partition里面。Kafka基于时间或者 Partition的大小来删除消息，同时 Broker是无状态的，Consumer的消费状态(offset)是由Consumer 自己控制的(每一个 Consumer实例只会消费某一个或多个特定 Partition的数据，而某个 Partition的数据只会被某一个特定的 Consumer实例所消费)，也不需要 Broker通过锁机制去控制消息的消费，所以吞吐量惊人，这也是 Kafka吸引人的地方。最后说下由于 zookeeper 引起的脑裂(Split Brain)问题：脑裂问题就是产生了两个 Leader，导致集群行为不一致了。1个集群如果发生了网络故障，很可能出现1个集群分成了两部分，而这两个部分都不知道对方是否存活，不知道到底是网络问题还是直接机器down了，所以这两部分都要选举1个Leader，而一旦两部分都选出了Leader, 并且网络又恢复了，那么就会出现两个 Brain的情况，整个集群的行为不一致了。解决：只有集群中超过半数节点投票才能选举出 Leader。ZooKeeper默认采用了这种方式。

【Kafka 的设计目标】：kafka在设计之初就需要考虑以下5个方面的问题
【1】以时间复杂度为O(1)的方式提供消息持久化能力，即使对 TB级以上数据也能保证常数时间复杂度的访问性能。
【2】高吞吐率，即使在非常廉价的商用机器上也能做到单机支持每秒100K条以上消息的传输。
【3】支持Kafka Server间的消息分区，及分布式消费，同时保证每个Partition内的消息顺序传输。
【4】同时支持离线数据处理和实时数据处理。
【5】Scale out：支持在线水平扩展。
所以，不像 AMQ，Kafka 从设计开始极为高可用为目的，天然 HA。Broker 支持集群，消息亦支持负载均衡，还有副本机制。同样，Kafka 也是使用 Zookeeper 管理集群节点信息，包括 Consumer 的消费信息也是保存在 zk 中，下面我们分话题来谈：

【和传统的MQ不同】：消费者需要自己保留一个offset，从kafka 获取消息时，只拉取当前offset 以后的消息。将 offset 保存到 zookeeper 上。每个消费者可以选择一个id，同样id 的消费者对于同一条消息只会收到一次。一个Topic 的消费者如果都使用相同的id，就是传统的 Queue；如果每个消费者都使用不同的id, 就是传统的pub-sub。
【kafka 主从同步怎么实现】：Kafka 的主从同步，主要是针对它的 Broker来说。在 Kafka 的 Broker 中，同一个 Topic 可以被分配成多个 Partition，每个 Partition的可以有一个或者多个 replicas(备份)，即会有一个 Leader 以及 0到多个 Follower，在Consumer 读取数据的时候，只会从 Leader上读取数据，Follower只是在 Leader宕机的时候来替代 Leader，主从同步有两种方式：同步复制和异步复制，Kafka采用的是中间策略 ISR(In Sync Replicas)。
【Kafka 的 ISR策略】：有数据写 Leader的时候，Leader会查看 Follower组成的 ISR列表，并且符合以下两点才算是属于 ISR列表：【1】Broker 可以维护和 zookeeper的连接，zookeeper通过心跳机制检查每个节点的连接。【2】如果节点是个 Follow它必须能及时同步 Leader的写操作，不能延时太久。当有写消息的时候，我们可以根据配置做如下配置：request.required.acks 参数的设置来进行调整：

☞ 0 ，相当于异步发送，消息发送完毕即 offset增加，继续生产；相当于At most once；
☞ 1，Leader 收到Leader Replica 对一个消息的接收 ack才增加 offset，然后继续生产；
☞ -1，Leader 收到所有 Replica 对一个消息的接收 ack才增加 offset，然后继续生产；

MQ 的消息延迟了怎么处理

【1】延迟处理：可以通过设置延迟级别，控制消息延迟的时间。
【2】设置过期时间：

<broker>     ...         <plugins>                              <timeStampingBrokerPluginttlCeiling="30000" zeroExpirationOverride="30000" />          plugins>     ... broker>

1)Message 过期则客户端不能接收；
   2)ttlCeiling：表示过期时间上限(程序写的过期时间不能超过此时间)；
   3)zeroExpirationOverride：表示过期时间(给未分配过期时间的消息分配过期时间)；
【3】过期消息处理办法：消息过期后会进入死信队列，如不想抛弃死信队列，默认进入 ACTIVEMQ.DLQ队列，且不会自动清除；对于过期的消息进入死信队列还有一些可选的策略：放入各自的死信通道、保存在一个共享的队列(默认)，且可以设置是否将过期消息放入队列的开关以及死信队列消息过期时间。
   1)直接抛弃死信队列：AcitveMQ提供了一个便捷的插件：DiscardingDLQBrokerPlugin，来抛弃DeadLetter。如果开发者不需要关心DeadLetter，可以使用此策略。

<broker>...    <plugins>        <discardingDLQBrokerPlugindropAll="true"  dropTemporaryTopics="true" dropTemporaryQueues="true" />                    plugins>    ... broker>

2)定时抛弃死信队列：默认情况下，ActiveMQ永远不会过期发送到 DLQ的消息。但是，从 ActiveMQ5.12开始，deadLetterStrategy 支持 expiration属性，其值以毫秒为单位。

<policyEntryqueue=">"…>   ...  <deadLetterStrategy>    <sharedDeadLetterStrategy processExpired="true" expiration="30000"/>  deadLetterStrategy>   ...policyEntry>

3)慢消费者策略设置：Broker将会启动一个后台线程用来检测所有的慢速消费者，并定期关闭它们;中断慢速消费者，慢速消费将会被关闭。abortConnection是否关闭连接;如果慢速消费者最后一个ACK距离现在的时间间隔超过阀 maxTimeSinceLastAck，则中断慢速消费者。

<policyEntryqueue=">"…>    …    <slowConsumerStrategy>        <abortSlowConsumerStrategyabortConnection="false"/>      slowConsumerStrategy>    …policyEntry>

利用 MQ 怎么实现最终一致性

RabbitMQ 遵循了 AMQP 规范，用消息确认机制来保证：只要消息发送，就能确保被消费者消费，来做到了消息最终一致性。Rabbitmq 的整个发送过程如下：【1】生产者发送消息到消息服务。
【2】如果消息落地持久化完成，则返回一个标志给生产者。生产者拿到这个确认后，才能放心的说消息终于成功发到消息服务了。否则进入异常处理流程。
【3】消息服务将消息发送给消费者。
【4】消费者接受并处理消息，如果处理成功则手动确认。当消息服务拿到这个确认后，才放心的说终于消费完成了。否则重发，或者进入异常处理。

使用 kafka 有没有遇到什么问题，怎么解决的

【问题】：两台设备上只有一个上存在 logs；
【基本情况】：一个 Topic 配置了四个 Partition，一个 Consumer Group 消费此Topic，但使用两台服务器，分别创建 Consumer 实例。都运行日志收集程序。
【问题】：Consumer Group 是将消费到的日志写入服务器磁盘文件中。有两台服务器都在运行此日志收集程序，每个服务器上的程序都创建了一个 Group 的 Consumer实例，此 Consumer实例会分配到两个 Partition进行处理，因此每个服务器都只存储了一部分日志文件。但是在测试时发现，所有日志都写入了 ServerA，ServerB上没有日志，即便使用测试工具发送了大量数据，ServerB仍然没有日志。
【原因】：查看 log发现，ServerA 上的 Consumer实例分配的 Partition 为 Partition_0 / Partition_1，serverB 上的 Consumer实例分配的 Partition 为partition_3 / Partition_4，两个 Server上的 Consumer实例都被分配了Partition，Partition分配正常，消费应该没有问题。ServerB 上没有日志数据，说明没有数据供其消费，也就是说，所有数据都被 Producer发送到了 Partition_1 或Partition_2 上，这是生产的问题，应该是与生产者的分区路由有关，因此有必要了解下生产者的分区路由策略。Kafka 中的每个Topic 分配了4个 Partition，生产者(Producer)在将消息记录(ProducerRecord)发送到某个 Topic时是要选择对应的 Partition的，选择 Partition的策略如下：
【1】消息中指定Partition：判断 Partition字段是否有值，有值就直接将该消息发送到指定的 Partition就行；
【2】如果没有指定分区(Partition)，则使用分区器进行分区路由，首先判断消息中是否指定了key；
【3】如果指定了key，则使用该 key进行 hash操作，并转为正数，然后将其对 Topic相应的分区数进行取余操作，得到一个分区；
【4】如果没有指定key，则在一个随机数上以自增的方式产生一个数(第一次时生成随机数，之后在其基础上进行自增)，转为正数之后对分区数量进行取余操作，得到一个分区。
由于在程序中 Producer发送记录的时候指定了固定的 key，根据这个 key进行分区路由总是会选择同一个分区，所有日志都被发送给了同一个分区，因此只有关联这个分区的 Consumer实例才能消费，只有此 Consumer实例所在的 Server上才有日志。

Kafka 中的 ISR、AR又代表什么

【1】ISR：In-Sync Replicas 副本同步队列；
【2】AR：Assigned Replicas 所有副本；
ISR是由 Leader维护，Follower 从 Leader同步数据有一些延迟(包括延迟时间 replica.lag.time.max.ms 和延迟条数replica.lag.max.messages 两个维度, 当前最新的版本0.10.x中只支持 replica.lag.time.max.ms这个维度)，任意一个超过阈值都会把 Follower剔除出 ISR，存入OSR(Outof-Sync Replicas)列表，新加入的 Follower 也会先存放在OSR中。AR=ISR+OSR。

十七、Kafka 为什么不支持读写分离

在 Kafka 中这种功能完全可以支持，同时主写从读可以让从节点去分担主节点的负载压力，预防主节点负载过重而从节点却空闲的情况发生。但是主写从读也有 2 个很明显的缺点：
【1】数据一致性问题。数据从主节点转到从节点必然会有一个延时的时间窗口，这个时间窗口会导致主从节点之间的数据不一致。某一时刻，在主节点和从节点中 A 数据的值都为 X，之后将主节点中 A 的值修改为 Y，那么在这个变更通知到从节点之前，应用读取从节点中的 A 数据的值并不为最新的 Y，由此便产生了数据不一致的问题。
【2】延时问题。类似 Redis 这种组件，数据从写入主节点到同步至从节点中的过程需要经历网络→主节点内存→网络→从节点内存这几个阶段，整个过程会耗费一定的时间。而在 Kafka 中，主从同步会比 Redis 更加耗时，它需要经历网络→主节点内存→主节点磁盘→网络→从节点内存→从节点磁盘这几个阶段。对延时敏感的应用而言，主写从读的功能并不太适用。
【Kafka 架构导致我们没有必要使用主从分离】在 Kafka 中这种负载均衡是在主写主读的架构上实现的。我们来看一下 Kafka 的生产消费模型，如下图所示。
在 Kafka 集群中有 3 个分区，每个分区有 3 个副本，正好均匀地分布在 3个 broker 上，灰色阴影的代表 Leader 副本，非灰色阴影的代表 Follower 副本，虚线表示 Follower 副本从 Leader 副本上拉取消息。当生产者写入消息的时候都写入 Leader 副本，对于图中的情形，每个 Broker 都有消息从生产者流入。当消费者读取消息的时候也是从 Leader 副本中读取的，对于图中的情形，每个 Broker 都有消息流出到消费者。从而将压力分配到每个服务器上，从而实现了负载均衡功能。

ZK 在 kafka 中的作用

【1】Broker 注册：Broker 是分布式部署并且相互之间相互独立，但是需要有一个注册系统能够将整个集群中的 Broker管理起来，此时就使用到了 Zookeeper。在 Zookeeper上会有一个专门用来进行 Broker服务器列表记录的节点：/brokers/ids 每个Broker在启动时，都会到 Zookeeper上进行注册，即到 /brokers/ids下创建属于自己的节点，如/brokers/ids/[0...N]。Kafka 使用了全局唯一的数字来指代每个 Broker服务器，不同的 Broker必须使用不同的 Broker ID进行注册，创建完节点后，每个 Broker就会将自己的 IP地址和端口信息记录到该节点中去。其中，Broker创建的节点类型是临时节点，一旦 Broker宕机，则对应的临时节点也会被自动删除。
【2】Topic 注册：在 Kafka中，同一个Topic的消息会被分成多个分区并将其分布在多个 Broker上，这些分区信息及与Broker的对应关系也都是由 Zookeeper在维护，由专门的节点来记录，如：/borkers/topics Kafka 中每个 Topic都会以 /brokers/topics/[topic] 的形式被记录，如 /brokers/topics/login 和 /brokers/topics/search 等。Broker服务器启动后，会到对应 Topic节点(/brokers/topics)上注册自己的 Broker ID并写入针对该 Topic的分区总数，如 /brokers/topics/login/3->2，这个节点表示Broker ID为3的一个 Broker服务器，对于"login" 这个 Topic的消息，提供了2个分区进行消息存储，同样，这个分区节点也是临时节点。
【3】生产者负载均衡：由于同一个 Topic消息会被分区并将其分布在多个 Broker上，因此，生产者需要将消息合理地发送到这些分布式的Broker上，那么如何实现生产者的负载均衡，Kafka 支持传统的四层负载均衡，也支持 Zookeeper方式实现负载均衡。
■ 四层负载均衡，根据生产者的 IP地址和端口来为其确定一个相关联的 Broker。通常，一个生产者只会对应单个 Broker，然后该生产者产生的消息都发往该Broker。这种方式逻辑简单，每个生产者不需要同其他系统建立额外的 TCP连接，只需要和 Broker维护单个 TCP连接即可。但是，其无法做到真正的负载均衡，因为实际系统中的每个生产者产生的消息量及每个 Broker的消息存储量都是不一样的，如果有些生产者产生的消息远多于其他生产者的话，那么会导致不同的 Broker接收到的消息总数差异巨大，同时，生产者也无法实时感知到 Broker的新增和删除。
■ 使用 Zookeeper进行负载均衡，由于每个Broker启动时，都会完成 Broker注册过程，生产者会通过该节点的变化来动态地感知到 Broker服务器列表的变更，这样就可以实现动态的负载均衡机制。
【4】消费者负载均衡：与生产者类似，Kafka 中的消费者同样需要进行负载均衡来实现多个消费者合理地从对应的 Broker服务器上接收消息，每个消费者分组包含若干消费者，每条消息都只会发送给分组中的一个消费者，不同的消费者分组消费自己特定的Topic下面的消息，互不干扰。
【5】分区与消费者的关系：消费组 (Consumer Group)下有多个 Consumer(消费者)。对于每个消费者组 (Consumer Group)，Kafka 都会为其分配一个全局唯一的Group ID，Group 内部的所有消费者共享该 ID。订阅的 Topic下的每个分区只能分配给某个 group 下的一个 Consumer(当然该分区还可以被分配给其他 group)。同时，Kafka为每个消费者分配一个Consumer ID，通常采用"Hostname:UUID"形式表示。在 Kafka中，规定了每个消息分区只能被同组的一个消费者进行消费，因此，需要在 Zk 上记录消息分区与 Consumer 之间的关系，每个消费者一旦确定了对一个消息分区的消费权力，需要将其Consumer ID 写入到 Zookeeper 对应消息分区的临时节点上，例如：/consumers/[group_id]/owners/[topic]/[broker_id-partition_id] 其中，[broker_id-partition_id]就是一个消息分区的标识，节点内容就是该消息分区上消费者的 Consumer ID。
【6】消息消费进度Offset 记录：在消费者对指定消息分区进行消息消费的过程中，需要定时地将分区消息的消费进度 Offset记录到Zookeeper上，以便在该消费者进行重启或者其他消费者重新接管该消息分区的消息消费后，能够从之前的进度开始继续进行消息消费。Offset 在 Zookeeper中由一个专门节点进行记录，其节点路径为:/consumers/[group_id]/offsets/[topic]/[broker_id-partition_id] 节点内容就是 Offset的值。
【7】消费者注册：消费者服务器在初始化启动时加入消费者分组的步骤如下：注册到消费者分组。每个消费者服务器启动时，都会到 Zookeeper的指定节点下创建一个属于自己的消费者节点，例如/consumers/[group_id]/ids/[consumer_id]，完成节点创建后，消费者就会将自己订阅的 Topic信息写入该临时节点。对消费者分组中的消费者的变化注册监听。每个消费者都需要关注所属消费者分组中其他消费者服务器的变化情况，即对/consumers/[group_id]/ids节点注册子节点变化的 Watcher监听，一旦发现消费者新增或减少，就触发消费者的负载均衡。对Broker服务器变化注册监听。消费者需要对/broker/ids/[0-N]中的节点进行监听，如果发现 Broker服务器列表发生变化，那么就根据具体情况来决定是否需要进行消费者负载均衡。进行消费者负载均衡。为了让同一个Topic下不同分区的消息尽量均衡地被多个消费者消费而进行消费者与消息分区分配的过程，通常，对于一个消费者分组，如果组内的消费者服务器发生变更或 Broker服务器发生变更，会发出消费者负载均衡。
【ZK 的详细存储结构图】:

早期版本的 kafka 用 zk 做 meta 信息存储，consumer 的消费状态，group 的管理以及 offset 的值。考虑到 zk本身的一些因素以及整个架构较大概率存在单点问题，新版本中确实逐渐弱化了 zookeeper的作用。新的 consumer使用了 kafka内部的 group coordination 协议，也减少了对 zookeeper的依赖。

Kafka Follower 如何与 Leader同步数据

Kafka 使用 ISR的方式很好的均衡了确保数据不丢失以及吞吐率。Follower 可以批量的从 Leader复制数据，而且Leader充分利用磁盘顺序读以及send file(zero copy)机制，这样极大的提高复制性能，内部批量写磁盘，大幅减少了 Follower与 Leader的消息量差。所有的 Follower 都复制 Leader 的日志，日志中的消息和顺序都和 Leader 中的一致。Follower 像普通的 Consumer 那样从 Leader 那里拉取消息并保存在自己的日志文件中。ISR 中有f+1个节点，就可以允许在f个节点 Down掉的情况下不会丢失消息并正常提供服。ISR 的成员是动态的，如果一个节点被淘汰了，当它重新达到“同步中”的状态时，他可以重新加入ISR。因此如果 Leader宕了，直接从 ISR中选择一个 Follower就行。只有当消息被所有的副本加入到日志中时，才算是“committed”，只有 committed的消息才会发送给 consumer，这样就不用担心 Leader Down掉了消息会丢失。Kafka 选择一个节点作为“controller”,当发现有Leader 节点 Down掉的时候它负责在LSR 分区的所有节点中选择新的 Leader，这使得 Kafka可以批量的高效的管理所有分区节点的主从关系。如果 controller down掉了，活着的节点中的一个会被切换为新的 controller。

什么情况下 Follower 会从 ISR 中踢除

Leader 维护一个与其基本保持同步的 Replica列表，该列表称为 ISR(in-sync Replica)，每个 Partition都会有一个 ISR，而且是由Leader动态维护，如果 Follower 比 Leader落后太多消息数量【replica.lag.max.messages】，或者超过一定时间未发起数据复制请求【replica.lag.time.max.ms】，则 Leader将其从 ISR中移除。

Kafka 为什么那么快

Kafka 的消息是保存或缓存在磁盘上的，一般认为在磁盘上读写数据是会降低性能的，因为寻址会比较消耗时间，但是实际上，Kafka 的特性之一就是高吞吐率。Kafka 之所以能这么快，是因为顺序写磁盘、大量使用内存页 、零拷贝技术的使用
【数据写入】：Kafka 会把收到的消息都写入到硬盘中，不会丢失数据。为了优化写入速度 Kafka 采用了两个技术， 顺序写入和 MMFile(Memory Mapped File)
【原因一：顺序写入】：磁盘读写的快慢取决于你怎么使用它，也就是顺序读写或者随机读写。在顺序读写的情况下，磁盘的顺序读写速度和内存持平。因为硬盘是机械结构，每次读写都会寻址->写入，其中寻址是一个“机械动作”，它是最耗时的。所以硬盘最讨厌随机 I/O，最喜欢顺序 I/O。为了提高读写硬盘的速度，Kafka 就是使用顺序 I/O。如果在内存做这些操作的时候，一个是 Java 对象的内存开销很大，另一个是随着堆内存数据的增多，Java 的 GC 时间会变得很长。
【使用磁盘操作有以下几个好处】：①、磁盘顺序读写速度超过内存随机读写。②、JVM 的 GC 效率低，内存占用大。使用磁盘可以避免这一问题。③、系统冷启动后，磁盘缓存依然可用。下图就展示了 Kafka 是如何写入数据的，每一个 Partition 其实都是一个文件，收到消息后 Kafka 会把数据插入到文件末尾(虚框部分)：

该方法的缺陷：没有办法删除数据，所以 Kafka 是不会删除数据的，它会把所有的数据都保留下来，每个消费者(Consumer)对每个 Topic 都有一个 Offset 用来表示读取到了第几条数据。
如果不删除硬盘肯定会被撑满，所以 Kakfa 提供了两种策略来删除数据：
【1】基于时间；
【2】基于 Partition 文件大小；
【原因二：Memory Mapped Files】：即便是顺序写入硬盘，硬盘的访问速度还是不可能追上内存。所以 Kafka 的数据并不是实时的写入硬盘，它充分利用了现代操作系统分页存储来利用内存提高 I/O 效率。Memory Mapped Files(后面简称 mmap)也被翻译成内存映射文件，在 64 位操作系统中一般可以表示 20G 的数据文件，它的工作原理是直接利用操作系统的 Page 来实现文件到物理内存的直接映射。完成映射之后你对物理内存的操作会被同步到硬盘上(操作系统在适当的时候)。通过 mmap，进程像读写硬盘一样读写内存(当然是虚拟机内存)，也不必关心内存的大小，有虚拟内存为我们兜底。使用这种方式可以获取很大的 I/O 提升，省去了用户空间到内核空间复制的开销。(调用文件的 Read 会把数据先放到内核空间的内存中，然后再复制到用户空间的内存中)但也有一个很明显的缺陷：不可靠，写到 mmap 中的数据并没有被真正的写到硬盘，操作系统会在程序主动调用 Flush 的时候才把数据真正的写到硬盘。Kafka 提供了一个参数 producer.type 来控制是不是主动 Flush：如果 Kafka 写入到 mmap 之后就立即 Flush，然后再返回 Producer 叫同步 (Sync)。如果 Kafka 写入 mmap 之后立即返回 Producer 不调用 Flush 叫异步 (Async)。
【原因三：Zero Copy】：传统模式下，当需要对一个文件进行传输的时候，其具体流程细节如下：调用 Read 函数，文件数据被 Copy 到内核缓冲区。Read 函数返回，文件数据从内核缓冲区 Copy 到用户缓冲区。Write 函数调用，将文件数据从用户缓冲区 Copy 到内核与 Socket 相关的缓冲区。数据从 Socket 缓冲区 Copy 到相关协议引擎。以上细节是传统 Read/Write 方式进行网络文件传输的方式，我们可以看到，在这个过程当中，文件数据实际上是经过了四次 Copy 操作：硬盘—>内核 buf—>用户 buf—>Socket 相关缓冲区—>协议引擎。而 Sendfile 系统调用则提供了一种减少以上多次 Copy，提升文件传输性能的方法。在内核版本 2.1 中，引入了 Sendfile 系统调用，以简化网络上和两个本地文件之间的数据传输。Sendfile 的引入不仅减少了数据复制，还减少了上下文切换。sendfile(socket, file, len);
【运行流程如下】：【1】Sendfile 系统调用，文件数据被 Copy 至内核缓冲区。【2】再从内核缓冲区 Copy 至内核中 Socket 相关的缓冲区。【3】最后再 Socket 相关的缓冲区 Copy 到协议引擎。
相较传统 Read/Write 方式，2.1 版本内核引进的 Sendfile 已经减少了内核缓冲区到 User 缓冲区，再由 User 缓冲区到 Socket 相关缓冲区的文件 Copy。而在内核版本 2.4 之后，文件描述符结果被改变，Sendfile 实现了更简单的方式，再次减少了一次 Copy 操作。在 Apache、Nginx、Lighttpd 等 Web 服务器当中，都有一项 Sendfile 相关的配置，使用 Sendfile 可以大幅提升文件传输性能。Kafka 把所有的消息都存放在一个一个的文件中，当消费者需要数据的时候 Kafka 直接把文件发送给消费者，配合 mmap 作为文件读写方式，直接把它传给 Sendfile。
【原因四：批量压缩】：在很多情况下，系统的瓶颈不是 CPU 或磁盘，而是网络 IO，对于需要在广域网上的数据中心之间发送消息的数据流水线尤其如此。进行数据压缩会消耗少量的 CPU 资源，不过对于 Kafka 而言，网络 IO 更应该考虑：
■ 因为每个消息都压缩，但是压缩率相对很低，所以 Kafka 使用了批量压缩，即将多个消息一起压缩而不是单个消息压缩。
■ Kafka 允许使用递归的消息集合，批量的消息可以通过压缩的形式传输并且在日志中也可以保持压缩格式，直到被消费者解压缩。
■ Kafka 支持多种压缩协议，包括 Gzip 和 Snappy 压缩协议。
【总结】：Kafka 速度的秘诀在于，它把所有的消息都变成一个批量的文件，并且进行合理的批量压缩，减少网络 IO 损耗，通过 mmap 提高 I/O 速度。写入数据的时候由于单个 Partion 是末尾添加，所以速度最优；读取数据的时候配合 Sendfile 直接暴力输出。

kafka 使用过程中遇到的问题

【基本情况】：两台设备上只有一个上存在 logs；
【详细情况】：一个topic，此topic配置了四个partition。两个consumer group，这两个consumer group用于消费同一个topic，但做不同的处理任务。每个consumer group中都只有一个 consumer实例进行消费。两台服务器，都运行此日志收集程序。
【问题】：　两个consumer group用于消费同一个 topic并做不同的处理，其中一个 consumer group(称作 group2)是将消费到的日志写入服务器磁盘文件中。有两台服务器都在运行此日志收集程序，每个服务器上的程序都创建了一个group2的consumer实例，此consumer实例会分配到两个 partition进行处理，因此每个服务器都只存储了一部分日志文件。但是在测试时发现，所有日志都写入了server1，server2上没有日志，即便使用测试工具发送了大量数据，server2仍然没有日志。
【原因】：查看 log发现，server1上的 consumer实例分配的 partition为 partition_0 partition_1，server2上的 consumer实例分配的partition为partition_3、partition_4，两个server上的consumer实例都被分配了partition，partition分配正常，消费应该没有问题。server2上没有日志数据，说明没有数据供其消费，也就是说，所有数据都被 producer发送到了 partition_1或 partition_2上，这是生产的问题，应该是与生产者的分区路由有关，因此有必要了解下生产者的分区路由策略。Kafka中的每个 Topic分配了4个Partition，生产者(Producer)在将消息记录(ProducerRecord)发送到某个 Topic时是要选择对应的 Partition的，选择 Partition的策略如下：
【1】判断消息中的 partition字段是否有值，有值的话就是指定了partition，直接将该消息发送到指定的 partition就行。
【2】如果没有指定分区(partition)，则使用分区器进行分区路由，首先判断消息中是否指定了key。
【3】如果指定了key，则使用该key进行hash操作，并转为正数，然后将其对topic相应的分区数进行取余操作，得到一个分区。
【4】如果没有指定key，则在一个随机数上以自增的方式产生一个数(第一次时生成随机数，之后在其基础上进行自增)，转为正数之后对分区数量进行取余操作，得到一个分区。
由于在程序中Producer发送记录的时候指定了固定的key，根据这个key进行分区路由总是会选择同一个分区，所有日志都被发送给了同一个分区，因此只有关联这个分区的 consumer实例才能消费，只有此 consumer实例所在的 server上才有日志。

end