分布式消息队列

输入

作者：vincentchma，腾讯IEG后台开发工程师一、消息队列的演进分布式消息队列中间件是是大型分布式系统中常见的中间件。消息队列主要解决应用耦合、异步消息、流量削锋等问题，具有高性能、高可用、可伸缩和最终一致性等特点。消息队列已经逐渐成为企业应用系统内部通信的核心手段，使用较多的消息队列有RabbitMQ、RocketMQ、ActiveMQ、Kafka、ZeroMQ、Pulsar等，此外，利用数据库（如Redis、MySQL等）也可实现消息队列的部分基本功能。1.基于OS的MQ单机消息队列可以通过操作系统原生的进程间通信机制来实现，如消息队列、共享内存等。比如我们可以在共享内存中维护一个双端队列：消息产出进程不停地往队列里添加消息，同时消息消费进程不断地从队尾有序地取出这些消息。添加消息的任务我们称为producer，而取出并使用消息的任务，我们称之为consumer。这种模式在早期单机多进程模式中比较常见，比如IO进程把收到的网络请求存入本机MQ，任务处理进程从本机MQ中读取任务并进行处理。单机MQ易于实现，但是缺点也很明显：因为依赖于单机OS的IPC机制，所以无法实现分布式的消息传递，并且消息队列的容量也受限于单机资源。2.基于DB的MQ即使用存储组件（如Mysql、Redis等）存储消息，然后在消息的生产侧和消费侧实现消息的生产消费逻辑，从而实现MQ功能。以Redis为例，可以使用Redis自带的list实现。Redislist使用lpush命令，从队列左边插入数据；使用rpop命令，从队列右边取出数据。与单机MQ相比，该方案至少满足了分布式，但是仍然带有很多无法接受的缺陷。热key性能问题：不论是用codis还是twemproxy这种集群方案，对某个队列的读写请求最终都会落到同一台redis实例上，并且无法通过扩容来解决问题。如果对某个list的并发读写非常高，就产生了无法解决的热key，严重可能导致系统崩溃没有消费确认机制：每当执行rpop消费一条数据，那条消息就被从list中永久删除了。如果消费者消费失败，这条消息也没法找回了。不支持多订阅者：一条消息只能被一个消费者消费，rpop之后就没了。如果队列中存储的是应用的日志，对于同一条消息，监控系统需要消费它来进行可能的报警，BI系统需要消费它来绘制报表，链路追踪需要消费它来绘制调用关系……这种场景redislist就没办法支持了不支持二次消费：一条消息rpop之后就没了。如果消费者程序运行到一半发现代码有bug，修复之后想从头再消费一次就不行了。针对上述缺点，redis5.0开始引入stream数据类型，它是专门设计成为消息队列的数据结构，借鉴了很多kafka的设计，但是随着很多分布式MQ组件的出现，仍然显得不够友好，毕竟Redis天生就不是用来做消息转发的。3.专用分布式MQ中间件随着时代的发展，一个真正的消息队列，已经不仅仅是一个队列那么简单了，业务对MQ的吞吐量、扩展性、稳定性、可靠性等都提出了严苛的要求。因此，专用的分布式消息中间件开始大量出现。常见的有RabbitMQ、RocketMQ、ActiveMQ、Kafka、ZeroMQ、Pulsar等等。二、消息队列设计要点消息队列本质上是一个消息的转发系统，把一次RPC就可以直接完成的消息投递，转换成多次RPC间接完成，这其中包含两个关键环节：1.消息转储；2.消息投递：时机和对象；基于此，消息队列的整体设计思路是：确定整体的数据流向：如producer发送给MQ，MQ转发给consumer，consumer回复消费确认，消息删除、消息备份等。利用RPC将数据流串起来，最好基于现有的RPC框架，尽量做到无状态，方便水平扩展。存储选型，综合考虑性能、可靠性和开发维护成本等诸多因素。消息投递，消费模式push、pull。消费关系维护，单播、多播等，可以利用zk、configserver等保存消费关系。高级特性，如可靠投递，重复消息，顺序消息等，很多高级特性之间是相互制约的关系，这里要充分结合应用场景做出取舍。1.MQ基本特性RPC通信MQ组件要实现和生产者以及消费者进行通信功能，这里涉及到RPC通信问题。消息队列的RPC，和普通的RPC没有本质区别。对于负载均衡、服务发现、序列化协议等等问题都可以借助现有RPC框架来实现，避免重复造轮子。存储系统存储可以做成很多方式。比如存储在内存里，存储在分布式KV里，存储在磁盘里，存储在数据库里等等。但归结起来，主要有持久化和非持久化两种。持久化的形式能更大程度地保证消息的可靠性（如断电等不可抗外力），并且理论上能承载更大限度的消息堆积（外存的空间远大于内存）。但并不是每种消息都需要持久化存储。很多消息对于投递性能的要求大于可靠性的要求，且数量极大（如日志）。这时候，消息不落地直接暂存内存，尝试几次failover，最终投递出去也未尝不可。常见的消息队列普遍两种形式都支持。从速度来看，理论上，文件系统>分布式KV（持久化）>分布式文件系统>数据库，而可靠性却相反。还是要从支持的业务场景出发作出最合理的选择。高可用MQ的高可用，依赖于RPC和存储的高可用。通常RPC服务自身都具有服务自动发现，负载均衡等功能，保证了其高可用。存储的高可用，例如Kafka，使用分区加主备模式，保证每一个分区内的高可用性，也就是每一个分区至少要有一个备份且需要做数据的同步。推拉模型push和pull模型各有利弊，两种模式也都有被市面上成熟的消息中间件选用。1.慢消费慢消费是push模型最大的致命伤，如果消费者的速度比发送者的速度慢很多，会出现两种恶劣的情况：1.消息在broker的堆积。假设这些消息都是有用的无法丢弃的，消息就要一直在broker端保存。2.broker推送给consumer的消息consumer无法处理，此时consumer只能拒绝或者返回错误。而pull模式下，consumer可以按需消费，不用担心自己处理不了的消息来骚扰自己，而broker堆积消息也会相对简单，无需记录每一个要发送消息的状态，只需要维护所有消息的队列和偏移量就可以了。所以对于慢消费，消息量有限且到来的速度不均匀的情况，pull模式比较合适。2.消息延迟与忙等这是pull模式最大的短板。由于主动权在消费方，消费方无法准确地决定何时去拉取最新的消息。如果一次pull取到消息了还可以继续去pull，如果没有pull取到则需要等待一段时间重新pull。消息投放时机即消费者应该在什么时机消费消息。一般有以下三种方式：攒够了一定数量才投放。到达了一定时间就投放。有新的数据到来就投放。至于如何选择，也要结合具体的业务场景来决定。比如，对及时性要求高的数据，可用采用方式3来完成。消息投放对象不管是JMS规范中的Topic/Queue，Kafka里面的Topic/Partition/ConsumerGroup，还是AMQP（如RabbitMQ）的Exchange等等，都是为了维护消息的消费关系而抽象出来的概念。本质上，消息的消费无外乎点到点的一对一单播，或一对多广播。另外比较特殊的情况是组间广播、组内单播。比较通用的设计是，不同的组注册不同的订阅，支持组间广播。组内不同的机器，如果注册一个相同的ID，则单播；如果注册不同的ID(如IP地址+端口)，则广播。例如pulsar支持的订阅模型有：Exclusive：独占型，一个订阅只能有一个消息者消费消息。Failover：灾备型，一个订阅同时只有一个消费者，可以有多个备份消费者。一旦主消费者故障则备份消费者接管。不会出现同时有两个活跃的消费者。Shared：共享型，一个订阅中同时可以有多个消费者，多个消费者共享Topic中的消息。Key_Shared：键共享型，多个消费者各取一部分消息。通常会在公共存储上维护广播关系，如configserver、zookeeper等。2.队列高级特性常见的高级特性有可靠投递、消息丢失、消息重复、事务等等，他们并非是MQ必备的特性。由于这些特性可能是相互制约的，所以不可能完全兼顾。所以要依照业务的需求，来仔细衡量各种特性实现的成本、利弊，最终做出最为合理的设计。可靠投递如何保证消息完全不丢失？直观的方案是，在任何不可靠操作之前，先将消息落地，然后操作。当失败或者不知道结果（比如超时）时，消息状态是待发送，定时任务不停轮询所有待发送消息，最终一定可以送达。但是，这样必然导致消息可能会重复，并且在异常情况下，消息延迟较大。例如：producer往broker发送消息之前，需要做一次落地。请求到server后，server确保数据落地后再告诉客户端发送成功。支持广播的消息队列需要对每个接收者，持久化一个发送状态，直到所有接收者都确认收到，才可删除消息。即对于任何不能确认消息已送达的情况，都要重推消息。但是，随着而来的问题就是消息重复。在消息重复和消息丢失之间，无法兼顾，要结合应用场景做出取舍。消费确认当broker把消息投递给消费者后，消费者可以立即确认收到了消息。但是，有些情况消费者可能需要再次接收该消息（比如收到消息、但是处理失败），即消费者主动要求重发消息。所以，要允许消费者主动进行消费确认。顺序消息对于push模式，要求支持分区且单分区只支持一个消费者消费，并且消费者只有确认一个消息消费后才能push另外一个消息，还要发送者保证发送顺序唯一。对于pull模式，比如kafka的做法：producer对应partition，并且单线程。consumer对应partition，消费确认（或批量确认），单线程消费。但是这样也只是实现了消息的分区有序性，并不一定全局有序。总体而言，要求消息有序的MQ场景还是比较少的。三、KafkaKafka是一个分布式发布订阅消息系统。它以高吞吐、可持久化、可水平扩展、支持流数据处理等多种特性而被广泛使用（如Storm、Spark、Flink）。在大数据系统中，数据需要在各个子系统中高性能、低延迟的不停流转。传统的企业消息系统并不是非常适合大规模的数据处理，但Kafka出现了，它可以高效的处理实时消息和离线消息，降低编程复杂度，使得各个子系统可以快速高效的进行数据流转，Kafka承担高速数据总线的作用。kafka基础概念BrokerKafka集群包含一个或多个服务器，这种服务器被称为broker。TopicTopic在逻辑上可以被认为是一个queue，每条消费都必须指定它的Topic，可以简单理解为必须指明把这条消息放进哪个queue里。为了使得Kafka的吞吐率可以线性提高，物理上把Topic分成一个或多个Partition，每个Partition在物理上对应一个文件夹，该文件夹下存储这个Partition的所有消息和索引文件。PartitionParition是物理上的概念，每个Topic包含一个或多个Partition。Producer负责发布消息到Kafkabroker。Consumer消息消费者，向Kafkabroker读取消息的客户端。ConsumerGroup每个Consumer属于一个特定的ConsumerGroup（可为每个Consumer指定groupname，若不指定groupname则属于默认的group）。kafka实现原理6一个典型的kafka集群包含若干Producer，若干个Broker（kafka支持水平扩展）、若干个ConsumerGroup，以及一个zookeeper集群。Producer使用push模式将消息发布到broker。consumer使用pull模式从broker订阅并消费消息。多个broker协同工作，producer和consumer部署在各个业务逻辑中。kafka通过zookeeper管理集群配置及服务协同。这样就组成了一个高性能的分布式消息发布和订阅系统。Kafka有一个细节是和其他mq中间件不同的点，producer发送消息到broker的过程是push，而consumer从broker消费消息的过程是pull，主动去拉数据。而不是broker把数据主动发送给consumer。Producer发送消息到broker时，会根据Paritition机制选择将其存储到哪一个Partition。如果Partition机制设置合理，所有消息可以均匀分布到不同的Partition里，这样就实现了负载均衡。如果一个Topic对应一个文件，那这个文件所在的机器I/O将会成为这个Topic的性能瓶颈，而有了Partition后，不同的消息可以并行写入不同broker的不同Partition里，极大的提高了吞吐率。Kafka特点优点：高性能：单机测试能达到100wtps低延时：生产和消费的延时都很低，e2e的延时在正常的cluster中也很低可用性高：replicate+isr+选举机制保证工具链成熟：监控运维管理方案齐全生态成熟：大数据场景必不可少kafkastream不足：无法弹性扩容：对partition的读写都在partitionleader所在的broker，如果该broker压力过大，也无法通过新增broker来解决问题扩容成本高：集群中新增的broker只会处理新topic，如果要分担老topic-partition的压力，需要手动迁移partition，这时会占用大量集群带宽消费者新加入和退出会造成整个消费组rebalance：导致数据重复消费，影响消费速度，增加延迟partition过多会使得性能显著下降：ZK压力大，broker上partition过多让磁盘顺序写几乎退化成随机写高吞吐机制顺序存取如果把消息以随机的方式写入到磁盘，那么磁盘首先要做的就是寻址，也就是定位到数据所在的物理地址，在磁盘上就要找到对应柱面、磁头以及对应的扇区；这个过程相对内存来说会消耗大量时间，为了规避随机读写带来的时间消耗，kafka采用顺序写的方式存储数据。页缓存即使是顺序存取，但是频繁的I/O操作仍然会造成磁盘的性能瓶颈，所以kafka使用了页缓存和零拷贝技术。当进程准备读取磁盘上的文件内容时，操作系统会先查看待读取的数据是否在页缓存中，如果存在则直接返回数据，从而避免了对物理磁盘的I/O操作；如果没有命中，则操作系统会向磁盘发起读取请求并将读取的数据页存入页缓存，之后再将数据返回给进程。一个进程需要将数据写入磁盘，那么操作系统也会检测数据对应的页是否在页缓存中，如果不存在，则会先在页缓存中添加相应的页，最后将数据写入对应的页。被修改过后的页也就变成了脏页，操作系统会在合适的时间把脏页中的数据写入磁盘，以保持数据的一致性。Kafka中大量使用了页缓存，这是Kafka实现高吞吐的重要因素之一。虽然消息都是先被写入页缓存，然后由操作系统负责具体的刷盘任务的，但在Kafka中同样提供了同步刷盘及间断性强制刷盘(fsync),可以通过参数来控制。同步刷盘能够保证消息的可靠性，避免因为宕机导致页缓存数据还未完成同步时造成的数据丢失。但是实际使用上，我们没必要去考虑这样的因素以及这种问题带来的损失，消息可靠性可以由多副本来解决，同步刷盘会带来性能的影响。页缓存的好处：I/OScheduler会将连续的小块写组装成大块的物理写从而提高性能；I/OScheduler会尝试将一些写操作重新按顺序排好，从而减少磁头移动时间；充分利用所有空闲内存（非JVM内存）；读操作可以直接在PageCache内进行，如果消费和生产速度相当，甚至不需要通过物理磁盘交换数据；如果进程重启，JVM内的Cache会失效，但PageCache仍然可用。零拷贝零拷贝技术可以减少CPU的上下文切换和数据拷贝次数。常规方式应用程序一次常规的数据请求过程，发生了4次拷贝，2次DMA和2次CPU，而CPU发生了4次的切换。（DMA简单理解就是，在进行I/O设备和内存的数据传输的时候，数据搬运的工作全部交给DMA控制器，而CPU不再参与任何与数据搬运相关的事情）零拷贝的方式通过零拷贝优化，CPU只发生了2次的上下文切换和3次数据拷贝。批量发送Kafka允许进行批量发送消息，先将消息缓存在内存中，然后一次请求批量发送出去，这种策略将大大减少服务端的I/O次数。数据压缩Kafka还支持对消息集合进行压缩，Producer可以通过GZIP或Snappy格式对消息集合进行压缩，Producer压缩之后，在Consumer需进行解压，虽然增加了CPU的工作，但在对大数据处理上，瓶颈在网络上而不是CPU，所以这个成本很值得。高可用机制副本Producer在发布消息到某个Partition时，先通过ZooKeeper找到该Partition的Leader，然后无论该Topic的ReplicationFactor为多少，Producer只将该消息发送到该Partition的Leader。Leader会将该消息写入其本地Log。每个Follower都从Leaderpull数据。这种方式上，Follower存储的数据顺序与Leader保持一致。Follower在收到该消息后，向Leader发送ACK，并把消息写入其Log。一旦Leader收到了ISR中的所有Replica的ACK，该消息就被认为已经commit了，Leader将增加HW并且向Producer发送ACK。为了提高性能，每个Follower在接收到数据后就立马向Leader发送ACK，而非等到数据写入Log中。因此，对于已经commit的消息，Kafka只能保证它被存于多个Replica的内存中，而不能保证它们被持久化到磁盘中，也就不能完全保证异常发生后该条消息一定能被Consumer消费。Consumer读消息也是从Leader读取，只有被commit过的消息才会暴露给Consumer。KafkaReplication的数据流如下图所示：对于Kafka而言，定义一个Broker是否“活着”包含两个条件：一是它必须维护与ZooKeeper的session（这个通过ZooKeeper的Heartbeat机制来实现）。二是Follower必须能够及时将Leader的消息复制过来，不能“落后太多”。Leader会跟踪与其保持同步的Replica列表，该列表称为ISR（即in-syncReplica）。如果一个Follower宕机，或者落后太多，Leader将把它从ISR中移除。这里所描述的“落后太多”指Follower复制的消息落后于Leader后的条数超过预定值或者Follower超过一定时间未向Leader发送fetch请求。Kafka的复制机制既不是完全的同步复制，也不是单纯的异步复制。完全同步复制要求所有能工作的Follower都复制完，这条消息才会被认为commit，这种复制方式极大的影响了吞吐率（高吞吐率是Kafka非常重要的一个特性）。异步复制方式下，Follower异步的从Leader复制数据，数据只要被Leader写入log就被认为已经commit，这种情况下如果Follower都复制完都落后于Leader，而如果Leader突然宕机，则会丢失数据。而Kafka的这种使用ISR的方式则很好的均衡了确保数据不丢失以及吞吐率。Follower可以批量的从Leader复制数据，这样极大的提高复制性能（批量写磁盘），极大减少了Follower与Leader的差距。一条消息只有被ISR里的所有Follower都从Leader复制过去才会被认为已提交。这样就避免了部分数据被写进了Leader，还没来得及被任何Follower复制就宕机了，而造成数据丢失（Consumer无法消费这些数据）。而对于Producer而言，它可以选择是否等待消息commit。这种机制确保了只要ISR有一个或以上的Follower，一条被commit的消息就不会丢失。故障恢复Leader故障leader发生故障后，会从ISR中选出一个新的leader，之后，为保证多个副本之间的数据一致性，其余的follower会先将各自的log文件高于HW的部分截掉，然后从新的leader同步数据。注意：这只能保证副本之间的数据一致性，并不能保证数据不丢失或者不重复。Kafka在ZooKeeper中动态维护了一个ISR（in-syncreplicas），这个ISR里的所有Replica都跟上了leader，只有ISR里的成员才有被选为Leader的可能。在这种模式下，对于f+1个Replica，一个Partition能在保证不丢失已经commit的消息的前提下容忍f个Replica的失败。LEO：每个副本最大的offset。HW：消费者能见到的最大的offset，ISR队列中最小的LEO。Follower故障follower发生故障后会被临时踢出ISR集合，待该follower恢复后，follower会读取本地磁盘记录的上次的HW，并将log文件高于HW的部分截取掉，从HW开始向leader进行同步数据操作。等该follower的LEO大于等于该partition的HW，即follower追上leader后，就可以重新加入ISR了。扩展性由于Broker存储着特定分区的数据，因此，不管是Broker还是分区的扩缩容，都是比较复杂的，属于典型的“有状态服务”扩缩容问题。接下来，我们看一下Pulsar是怎么针对kafka的不足进行优化的。四、PulsarApachePulsar是Apache软件基金会顶级项目，是下一代云原生分布式消息流平台，集消息、存储、轻量化函数式计算为一体。采用计算与存储分离架构设计，支持多租户、持久化存储、多机房跨区域数据复制，具有强一致性、高吞吐、低延时及高可扩展性等流数据存储特性。在消息领域，Pulsar是第一个将存储计算分离云原生架构落地的开源项目。服务和存储分离在kafka的基础上，把数据存储功能从Broker中分离出来，Broker仅面向生产者、消费者提供数据读写能力，但其自身并不存储数据。而在Broker层下面使用Bookie作为存储层，承担具体的数据存储职责。在Pulsar中，broker的含义和kafka中的broker是一致的，就是一个运行的Pulsar实例，提供多个分区的读写服务。由于broker层不在承担数据存储职责，使得Broker层成为无状态服务。这样一来，Broker的扩缩容就变得非常简单。相比之下，服务存储集于一体的Kafka就非常难以扩容。Broker和Bookie相互独立，方便实现独立的扩展以及独立的容错Broker无状态，便于快速上、下线，更加适合于云原生场景分区存储不受限于单个节点存储容量Bookie数据分布均匀分片存储1.在Kafka分区（Partition）概念的基础上，按照时间或大小，把分区切分成分片（Segment）。2.同一个分区的分片，分散存储在集群中所有的Bookie节点上。3.同一个分片，拥有多个副本，副本数量可以指定，存储于不同的Bookie节点。Pulsar性能和Kafka一样，Pulsar也使用了顺序读写和零拷贝等技术来提高系统的性能。此外，Pulsar还设计了分层缓存机制，在服务层和存储层都做了分层缓存，来提高性能。生产者发送消息时，调用Bookie层写入消息时，同时将消息写入broker缓存中。实时消费时（追尾读），首先从broker缓存中读取数据，避免从持久层bookie中读取，从而降低投递延迟。读取历史消息（追赶读）场景中，bookie会将磁盘消息读入bookie读缓存中，从而避免每次都读取磁盘数据，降低读取延时。Pulsar扩展性分片存储解决了分区容量受单节点存储空间限制的问题，当容量不够时，可以通过扩容Bookie节点的方式支撑更多的分区数据，也解决了分区数据倾斜问题，数据可以均匀的分配在Bookie节点上。Broker和Bookie灵活的容错以及无缝的扩容能力让ApachePulsar具备非常高的可用性，实现了无限制的分区存储。Broker扩展在Pulsar中Broker是无状态的，可以通过增加节点的方式实现快速扩容。当需要支持更多的消费者或生产者时，可以简单地添加更多的Broker节点来满足业务需求。Pulsar支持自动的分区负载均衡，在Broker节点的资源使用率达到阈值时，会将负载迁移到负载较低的Broker节点。新增Broker节点时，分区也将在Brokers中做平衡迁移，一些分区的所有权会转移到新的Broker节点。Bookie扩展存储层的扩容，通过增加Bookie节点来实现。通过资源感知和数据放置策略，流量将自动切换到新的ApacheBookie中，整个过程不会涉及到不必要的数据搬迁。即扩容时，不会将旧数据从现有存储节点重新复制到新存储节点。如图所示，起始状态有四个存储节点，Bookie1,Bookie2,Bookie3,Bookie4，以Topic1-Part2为例，当这个分区的最新的存储分片是SegmentX时，对存储层扩容，添加了新的Bookie节点，BookieX,BookieY，那么当存储分片滚动之后，新生成的存储分片，SegmentX+1,SegmentX+2，会优先选择新的Bookie节点（BookieX,BookieY）来保存数据。Pulsar可用性Broker容错如下图，假设当存储分片滚动到SegmentX时，Broker2节点失败。此时生产者和消费者向其他的Broker发起请求，这个过程会触发分区的所有权转移，即将Broker2拥有的分区Topic1-Part2的所有权转移到其他的Broker(Broker3)。由于数据存储和数据服务分离，所以新Broker接管分区的所有权时，它不需要复制Partiton的数据。新的分区Owner（Broker3）会产生一个新的分片SegmentX+1,如果有新数据到来，会存储在新的分片Segmentx+1上，不会影响分区的可用性。即当某个Broker实例故障时，整个集群的消息存储能力仍然完好。此时，集群只是丧失了特定分区的消息服务，只需要把这些分区的服务权限分配给其他Broker即可。注意，和Kafka一样，Pulsar的一个分区仍然只能由一个Broker提供服务，否则无法保证消息的分区有序性。Bookie容错如下图，假设Bookie2上的Segment4损坏。BookieAuditor会检测到这个错误并进行复制修复。Bookie中的副本修复是Segment级别的多对多快速修复，BookKeeper可以从Bookie3和Bookie4读取Segment4中的消息，并在Bookie1处修复Segment4。如果是Bookie节点故障，这个Bookie节点上所有的Segment会按照上述方式复制到其他的Bookie节点。所有的副本修复都在后台进行，对Broker和应用透明，Broker会产生新的Segment来处理写入请求，不会影响分区的可用性。Pulsar其他特性基于上述的设计特点，Pulsar提供了很多特性。读写分离Pulsar另外一个有吸引力的特性是提供了读写分离的能力，读写分离保证了在有大量滞后消费（磁盘IO会增加）时，不会影响服务的正常运行，尤其是不会影响到数据的写入。读写分离的能力由Bookie提供，简单说一下Bookie存储涉及到的概念：Journals：Journal文件包含了Bookie事务日志，在Ledger(可以认为是分片的一部分)更新之前，Journal保证描述更新的事务写入到Non-volatile的存储介质上；Entrylogger：Entry日志文件管理写入的Entry，来自不同ledger的entry会被聚合然后顺序写入；Indexfiles：每个Ledger都有一个对应的索引文件，记录数据在Entry日志文件中的Offset信息。Entry的读写入过程下图所示，数据的写入流程：数据首先会写入Journal，写入Journal的数据会实时落到磁盘；然后，数据写入到Memtable，Memtable是读写缓存；写入Memtable之后，对写入请求进行响应；Memtable写满之后，会Flush到EntryLogger和Indexcache，EntryLogger中保存了数据，Indexcache保存了数据的索引信息，然后由后台线程将EntryLogger和Indexcache数据落到磁盘。数据的读取流程：如果是Tailingread请求，直接从Memtable中读取Entry；如果是Catch-upread（滞后消费）请求，先读取Index信息，然后索引从EntryLogger文件读取Entry。一般在进行Bookie的配置时，会将Journal和Ledger存储磁盘进行隔离，减少Ledger对于Journal写入的影响，并且推荐Journal使用性能较好的SSD磁盘，读写分离主要体现在：写入Entry时，Journal中的数据需要实时写到磁盘，Ledger的数据不需要实时落盘，通过后台线程批量落盘，因此写入的性能主要受到Journal磁盘的影响；读取Entry时，首先从Memtable读取，命中则返回；如果不命中，再从Ledger磁盘中读取，所以对于Catch-upread的场景，读取数据会影响Ledger磁盘的IO，对Journal磁盘没有影响，也就不会影响到数据的写入。所以，数据写入是主要是受Journal磁盘的负载影响，不会受Ledger磁盘的影响。另外，Segment存储的多个副本都可以提供读取服务，相比于主从副本的设计，ApachePulsar可以提供更好的数据读取能力。通过以上分析，ApachePulsar使用ApacheBookKeeper作为数据存储，可以带来下列的收益：支持将多个Ledger的数据写入到同一个Entrylogger文件，可以避免分区膨胀带来的性能下降问题支持读写分离，可以在滞后消费场景导致磁盘IO上升时，保证数据写入的不受影响支持全副本读取，可以充分利用存储副本的数据读取能力多种消费模型Pulsar提供了多种订阅方式来消费消息，分为三种类型：独占（Exclusive），故障切换（Failover）或共享（Share）。Exclusive独占订阅：在任何时间，一个消费者组（订阅）中有且只有一个消费者来消费Topic中的消息。Failover故障切换：多个消费者（Consumer）可以附加到同一订阅。但是，一个订阅中的所有消费者，只会有一个消费者被选为该订阅的主消费者。其他消费者将被指定为故障转移消费者。当主消费者断开连接时，分区将被重新分配给其中一个故障转移消费者，而新分配的消费者将成为新的主消费者。发生这种情况时，所有未确认（ack）的消息都将传递给新的主消费者。Share共享订阅：使用共享订阅，在同一个订阅背后，用户按照应用的需求挂载任意多的消费者。订阅中的所有消息以循环分发形式发送给订阅背后的多个消费者，并且一个消息仅传递给一个消费者。当消费者断开连接时，所有传递给它但是未被确认（ack）的消息将被重新分配和组织，以便发送给该订阅上剩余的剩余消费者。多种ACK模型消息确认（ACK）的目的就是保证当发生故障后，消费者能够从上一次停止的地方恢复消费，保证既不会丢失消息，也不会重复处理已经确认（ACK）的消息。在Pulsar中，每个订阅中都使用一个专门的数据结构–游标（Cursor）来跟踪订阅中的每条消息的确认（ACK）状态。每当消费者在分区上确认消息时，游标都会更新。Pulsar提供两种消息确认方法：单条确认（IndividualAck），单独确认一条消息。被确认后的消息将不会被重新传递累积确认（CumulativeAck），通过累积确认，消费者只需要确认它收到的最后一条消息上图说明了单条确认和累积确认的差异（灰色框中的消息被确认并且不会被重新传递）。对于累计确认，M12之前的消息被标记为Acked。对于单独进行ACK，仅确认消息M7和M12，在消费者失败的情况下，除了M7和M12之外，其他所有消息将被重新传送。腾讯程序员视频号最新视频欢迎点赞