消息队列的使用场景

---

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

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” 属性来修改。

                                                          deadLetterStrategy>      policyEntry>      policyEntries>    policyMap>    destinationPolicy>    ...  broker>  

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

                                                          deadLetterStrategy>       policyEntry>     policyEntries>     policyMap>    destinationPolicy>  ...  broker>  

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

request.required.acks=1

queue.buffering.max.ms=5000  #异步发送的时候 发送时间间隔 单位是毫秒

queue.buffering.max.messages=10000

queue.enqueue.timeout.ms = -1

batch.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具有高的吞吐量�