andy的博客

原创 中流-介绍

中流，让用户能够快速把流程引擎接入自己的系统，可配置、定制化的界面，丰富的流程模板，让用户轻松创建、改变、维护业务流程应用

原创 Kafka-服务端-GroupMetadataManager

consumer_offsets的某Partition记录某consumerGroup的GroupMotadata消息记录某ConsumerGroup对Partition的offset消息记录某ConsumerGroup对Partition1的offset消费记录某ConsumerGroup的Partition2的offset消费。记录GroupMetadata信息的消息和记录消费offset位置的消息都是通过partitionForO方法在Offsets Topic中选择合适的分区。

原创 Kafka-服务端-GroupCoordinator

三是记录Consumer Group的相关信息，即使Broker宕机导致Consumer Group由新的GroupCoordinator进行管理，新GroupCoordinator也可以知道Consumer Group中每个消费者负责处理哪个分区等信息；在每一个Broker上都会实例化一个GroupCoordinator对象，Kafka按照Consumer Group的名称将其分配给对应的GroupCoordinator进行管理；请注意与KafkaController以及副本机制中的主从模式的区别。

原创 Kafka-服务端-PartitionLeaderSelector、ReplicaStateMachine

ReassignedPartitionLeaderSelector涉及到副本的重新分配，副本重新分配的相关概念后面详细分析，这里先简单了解ReassignedPartitionLeaderSelector的策略：选取的新Leader副本必须在新指定的AR集合中且同时在当前ISR集合中，当前ISR集合为新ISR集合，接收LeaderAndIsrRequest的副本是新指定的AR集合中的副本。副本状态由ReplicaState接口表示，它有七个子类，分别代表了副本的七种不同的状态，如表所示。

原创 Kafka-服务端-PartitionStateMachine

为分区选择新的Leader副本和ISR集合，并将结果写入ZooKeeper。之后，向需要进行角色切换的副本发送LeaderAndIsrRequest,指导这些副本进行Leader/Follower的角色切换，并向所有可用的Broker发送UpdateMetadataRequest来更新其上的MetadataCache。首先将Leader副本和ISR集合的信息写入到ZooKeeper中，这里会将分区的AR集合中第一个可用的副本选举为Leader副本，并将分区的所有可用副本作为ISR集合。

原创 Kafka-服务端-ControllerChannelManager、ControllerContext、ControllerBrokerRequestBatch

ControllerBrokerRequestBatch.addLeaderAndIsrRequestForBrokers()方法会 向leaderAndIsrRequestMap集合中添加待发送的LeaderAndIsrRequest所需的数据，同时会调用addUpdateMetadataRequestForBrokers()方法准备向集群中所有可用的Broker发送UpdateMetadataRequest。ControllerContext中各个字段的含义和作用如下所述。

原创 Kafka-服务端-KafkaController

为了实现Controller的高可用，一个Broker被选为Leader之后，其他的Broker都会成为Follower(不加特殊说明的情况下，本文的“Leader/Follower”指的都是KafkaController的Leader/Follower,请不要与副本机制中的Leader副本和Follower副本混淆),会从剩下的Follower中选出新的Controller Leader来管理集群。当使用kafka-topics脚本增加某Topic的分区数量时，由Controller管理分区的重新分配；

原创 Kafka-服务端-副本机制

Kafka从0.8版本开始引入副本(Replica)的机制，其目的是为了增加Kafka集群的高可用性。Kafka实现副本机制之后，每个分区可以有多个副本，并且会从其副本集合(Assigned Replica,AR)中选出一个副本作为Leader副本，所有的读写请求都由选举出的Leader副本处理。剩余的其他副本都作为Follower副本，Follower副本会从Leader副本处获取消息并更新到自己的Log中。我们可以认为Follower副本是Leader副本的热备份。

原创 Kafka-服务端-DelayedOperationPurgatory

DelayedOperationPurgatory是一个相对独立的组件，它的主要功能是管理延迟操作。DelayedOperationPurgatory的底层依赖于Kafka提供的时间轮实现。我们可以使用JDK本身提供的java.util.Timer或是DelayQueue轻松实现定时任务的功能，为什么Kafka还要专门开发DelayedOperationPurgatory组件呢?

原创 Kafka-服务端-日志存储

Kafka中存储的一般都是海量消息数据，为了避免日志文件太大，Log并不是直接对应于磁盘上的一个日志文件，而是对应磁盘上的一个目录，这个目录的命名规则是<topic_name>_≤partition_id>,Log与分区之间的关系是一一对应的，对应分区中的全部消息都存储在此目录下的日志文件中。在重建索引文件过程中，如果遇到了压缩消息需要进行解压，主要原因是因为索引项中保存的相对offset是第一条消息的offset,而外层消息的offset是压缩消息集合中的最后一条消息的offset。

原创 Kafka-服务端-API层

Handler线程会取出Processor线程，放人RequestChannel的请求进行处理，并将产生的响应通过RequestChannel传递给Processor线程。Handler线程属于Kafka的API层，Handler线程对请求的处理通过调用KafkaApis中的方法实现。

原创 Kafka-服务端-网络层

Acceptor和Processor都继承了AbstractServerThread,如图所示，AbstractServerThread是实现了Runnable接口的抽象类。在AbstractServerThread中为Acceptor和Processor提供了一些启动关闭相关的控制类方法。

原创 Kafka-服务端-整体架构

原创 Kafka-消费者-KafkaConsumer分析总结

Fetcher负责从Kafka中拉取消息并进行解析，同时参与position的重置操作，提供获取指定Topic的集群元数据的操作。KafkaConsumer依赖SubscriptionState管理订阅的Topic集合和Partition的消费状态，通过ConsumerCoordinator与服务端的GroupCoordinator交互，完成Rebalance操作并请求最近提交的offset。我们可以看出，这并不是一种锁的实现，仅实现了检测多线程并发操作的检测。这里使用CAS操作可以保证线程之间的可见性。

原创 Kafka-消费者-KafkaConsumer分析-offset操作

commitOffsetsSync()方法与commitOffsetsAsync()方法的实现类似，也是调用sendOffsetCommitRequest()方法创建并缓存OffsetCommitRequest,使用OffsetCommitResponseHandler处理OffsetCommitResponse。在有些场景下，例如第一次消费某个Topic的分区，服务端的内部Offsets Topic中并没有记录当前消费者在此分区上的消费位置，所以消费者无法从服务端获取最近提交的offset。

原创 Kafka-消费者-KafkaConsumer分析-Rebalance

GroupCoordinatorResponse消息体包含了错误码(short类型)、coordinator的节点Id(int类型)、GroupCoordinator的host(String类型)、GroupCoordinator的端口号(int类型)。Rebalance操作的第一步就是查找GroupCoordinator,这个阶段消费者会向Kafka集群中的任意一个Broker发送GroupCoordinatorRequest请求，并处理返回的GroupCoordinatorResponse响应。

原创 Kafka-消费者-KafkaConsumer分析-Heartbeat

HeartbeatCompletionHandler.handle()方法中会调用RequestFuture的complete方法或raise方法，这两个方法中没有处理逻辑，但是会触发其上的RequestFutureListener(在HeartbeatTaskrun)方法的步骤6中注册),此监听器会将requestlnFlight设置为false,表示所有HeartbeatRequest都已经完成，并将HeartbeatTask重新放入定时任务队列，等待下一次到期执行。首先了解一下心跳请求和响应的格式。

原创 Kafka-消费者-KafkaConsumer分析-PartitionAssignor

Leader消费者在收到JoinGroupResponse后，会按照其中指定的分区分配策略进行分区分配，每个分区分配策略就是一个PartitionAssignor接口的实现。Assignment中保存了分区的分配结果，partitions表示的是分配给某消费者的TopicPartition集合，userData是用户自定义的数据。为了用户增强对分配结果的控制，就将用户订阅信息和一些影响分配的用户自定义信息封装成Subscription,例如，“用户自定义数据”可以是每个消费者的权重。

原创 Kafka-消费者-KafkaConsumer分析-ConsumerCoordinator

heartbeat:心跳任务的辅助类，其中记录了两次发送心跳消息的间隔(interval字段)、最近发送心跳的时间(lastHeartbeatSend字段)、最后收到心跳响应的时间(lastHeartbeatReceive字段)、过期时间(timeout字段)、心跳任务重置时间(lastSessionReset字段),同时还提供了计算下次发送心跳的时间(timeToNextHeartbeat()方法)、检测是否过期的方法(sessionTimeoutExpired()方法)。

原创 Kafka-消费者-KafkaConsumer分析-SubscriptionState

我们在前面介绍Metadata的时候提到过，可以在其上添加Listener,当Metadata更新时会触发Metadata.Listener.onMetadataUpdate()方法，图中的②处就是在Metadata的Listener中通过subscribedPattern模式过滤Topic,并调用changeSubscription()方法修改subscription集合。SubscriptionType是SubscriptionState的一个内部枚举类型，表示的是订阅Topic的模式，分为四类。

原创 Kafka-消费者-KafkaConsumer分析-ConsumerNetworkClient

前面介绍过NetworkClient的实现，它依赖于KSelector、InFlightRequests、Metadata等组件，负责管理客户端与Kafka集群中各个Node节点之间的连接，通过KSelector法实现了发送请求的功能，并通过一系列handle*方法处理请求响应、超时请求以及断线重连。ConsumerNetworkClient在NetworkClient之上进行了封装，提供了更高级的功能和更易用的API。在图中展示了ConsumerNetworkClient的核心字段以及其依赖的组件。

原创 Kafka-消费者-KafkaConsumer分析

其中一个线程池中每个线程拥有一个KafkaConsumer对象，负责从Kafka集群拉取消息，然后将消息放入队列中缓存，而另一个线程池中的线程负责从队列中获取消息，执行处理消息的业务逻辑。这种设计将实现多线程处理消息的逻辑转移到了调用KafkaConsumer的代码中，可以根据业务逻辑使用不同的实现方式。例如，可以使用“线程封闭”的方式，每个业务线程拥有一个KafkaConsumer对象，这种方式实现简单、快速。为了便于分析，我们认为下面介绍的所有操作都是在同一线程中完成的，所以不需要考虑锁的问题。

原创 Kafka-消费者-Consumer Group Rebalance设计

在同一个Consumer Group中，同一个Topic的不同分区会分配给不同的消费者进行消费，那么为消费者分配分区的操作是在Kafka服务端完成的吗?分区是如何进行分配呢?下面来分析Rebalance操作的原理。

原创 Kafka-消费者-传递保证语义(Delivery guarantee semantic)

Offsets Topic可以用来保存消费者提交的offset,当出现消费者上/下线时会触发Consumer Group进行Rebalance操作，对分区进行重新分配，待Rebalance操作完成后，消费者就可以读取Offsets Topic中记录的offset,并从此offset位置继续消费。KafkaConsumer中还提供了两个手动提交offset的方法，分别是commitSync()方法和commitAsync()方法，它们都可以指定提交的offset值，区别在于前者是同步提交，后者是异步提交。

原创 Kafka-Sender分析

通过前面的分析我们知道，主线程通过KafkaProducer.send方法将消息放入RecordAccumulator中缓存，并没有实际的网络I/O操作。网络I/O操作是由Sender线程统一进行的。我们先来了解一下Sender线程发送消息的整个流程：首先，它根据RecordAccumulator的缓存情况，筛选出可以向哪些Node节点发送消息，即上一节介绍的RecordAccumulatorready方法；然后，根据生产者与各个节点的连接情况由NetworkClient管理,过滤Node节点；

原创 Kafka-RecordAccumulator分析

前面介绍过，KafkaProducer可以有同步和异步两种方式发送消息，其实两者的底层实现相同，都是通过异步方式实现的。主线程调用KafkaProducer.send方法发送消息的时候，先将消息放到RecordAccumulator中暂存，然后主线程就可以从sendO方法中返回了，此时消息并没有真正地发送给Kafka,而是缓存在了RecordAccumulator中。

原创 Kafka-生产者

Kafka在实际应用中，经常被用作高性能、可扩展的消息中间件。Kafka自定义了一套网络协议，只要遵守这套协议的格式，就可以向Kafka发送消息，也可以从Kafka中拉取消息。在实践生产过程中，一套API封装良好、灵活易用的客户端可以避免开发人员重复劳动，提高开发效率，也可以提高程序的健壮性和可靠性。

原创 Kafka-核心概念

介绍完Kafka的核心概念，我们通过图进行总结，并从更高的视角审视整个Kafka集群的架构。

原创 Rust-Panic

在Rust中，有一类错误叫作panic。示例如下：编译，没有错误，执行这段程序，输出为：这种情况就引发了一个panic。在这段代码中，我们调用了Option::unwrap()方法，正是这个方法有可能导致panic。根据提示，我们设置一个环境变量RUST_BACKTRACE=1之后再执行这个程序，可以看到这个程序在发生panic时候的函数调用栈。

原创 Rust-泄漏

在C++中，如果引用计数智能指针出现了循环引用，就会导致内存泄漏。而Rust中也一样存在引用计数智能指针Rc,那么Rust中是否可能制造出内存泄漏呢?

原创 Rust-解引用

解引用操作可以被自定义。方法是，实现标准库中的std::ops::Deref或者std::ops::DerefMut这两个trait。Deref的定义如下所示。DerefMut的唯一区别是返回的是&mut型引用都是类似的，因此不过多介绍了。这个trait有一个关联类型Target,代表解引用之后的目标类型。比如，标准库中实现了String向str的解引用转换：deref()方法返回的类型是&Target,而不是Target。

原创 Rust-内部可变性

Rust的borrow checker的核心思想是“共享不可变，可变不共享”。但是只有这个规则是不够的，在某些情况下，我们的确需要在存在共享的情况下可变。为了让这种情况是可控的、安全的，Rust还设计了一种“内部可变性”(interior mutability)。“内部可变性”的概念，是与“承袭可变性”(inherited mutability)相对应的。大家应该注意到了，Rust中的mut关键字不能在声明类型的时候使用，只能跟变量一起使用。类型本身不能规定自己是否是可变的。

原创 Rust-NLL(Non-Lexical-Lifetime)

Rust防范“内存不安全”代码的原则极其清晰明了。如果你对同一块内存存在多个引用，就不要试图对这块内存做修改；如果你需要对一块内存做修改，就不要同时保留多个引用。只要保证了这个原则，我们就可以保证内存安全。它在实践中发挥了强大的作用，可以帮助我们尽早发现问题。这个原则是Rust的立身之本、生命之基、活力之源。这个原则是没问题的，但是，初始的实现版本有一个主要问题，那就是它让借用指针的生命周期规则与普通对象的生命周期规则一样，是按作用域来确定的。

原创 Rust-借用检查

Rust语言的核心特点是：在没有放弃对内存的直接控制力的情况下，实现了内存安全。所谓对内存的直接控制能力，前文已经有所展示：可以自行决定内存布局，包括在栈上分配内存，还是在堆上分配内存；支持指针类型；可以对一个变量实施取地址操作；有确定性的内存释放；等等。另一方面，从安全性的角度来说，我们可以看到，Rust有所有权概念、借用指针、生命周期分析等这些内容。随便写个小程序都编译不通过，学习曲线非常陡峭。那么,Rust设计者究竟是如何考虑的这个问题，为什么要设计这样复杂的规则?

原创 Rust-借用和生命周期

test返回的那个指针在’t这个生命周期范围内都是合法的，在一个被’t包围的更小范围的生命周期内，它当然也是合法的。在函数被调用的时候，它传人的实际参数是&t,它是指向t的引用。要特别说明的是，若是有where T:'static的约束，意思则是，类型T里面不包含任何指向短生命周期的借用指针，意思是要么完全不包含任何借用，要么可以有指向’static的借用指针。select函数的形式参数要求的是同样的生命周期，而实际参数是两个不同生命周期的引用，这个类型之所以可以匹配成功，就是因为生命周期的协变特性。

原创 Rust-析构函数

所谓“析构函数”(destructor),是与“构造函数”(constructor)相对应的概念。“构造函数”是对象被创建的时候调用的函数，“析构函数”是对象被销毁的时候调用的函数。Rust中没有统一的“构造函数”这个语法，对象的构造是直接对每个成员进行初始化完成的，我们一般将对象的创建封装到普通静态函数中。相对于构造函数，析构函数有更重要的作用。它会在对象消亡之前由编译器自动调用，因此特别适合承担对象销毁时释放所拥有的资源的作用。

原创 Rust-所有权和移动语义

拿C语言的代码来打个比方。然而，这段代码之后，谁能猜得到，指针p指向的对象究竟发生了什么?它是否被修改过了?它还存在吗，是否已经被释放?是否有另外一个指针现在也同时指向这个对象?我们还能继续读取、修改或者释放这个对象吗?实际上，除了去了解use_object的内部实现之外，我们没办法回答以上问题。对此，C++进行了一个改进，即通过“智能指针”来描述“所有权”(Ownership)概念。这在一定程度上减少了内存使用bug,实现了“半自动化”的内存管理。

原创 Rust-内存安全

堆是在内存中动态分配的内存，是无序的。同样，内存耗尽也不是事关安全性的问题，出现内存耗尽的时候，Rust程序的行为依然是确定性的和可控的(目前版本下，如果内存耗尽则发生panic,也有人认为在这种情况发生的时候，应该给个机会由用户自己处理，这种情况后面应该会有改进)。某些应用场景下用这样的代价换取开发效率和安全性是非常划算的，而在某些应用场景下这样的代价是不可接受的。内存泄漏显然是一种bug,但是它不会直接造成非常严重的后果，至少比上面列出的那些错误危险性要低一些，解决的办法也是完全不一样的。

原创 Rust-模式解构

下划线表示省略一个元素，两个点可以表示省略多个元素。

原创 Rust-数组

数组是一个容器，它在一块连续空间内存中，存储了一系列的同样类型的数据。数组中元素的占用空间大小必须是编译期确定的。数组本身所容纳的元素个数也必须是编译期确定的，执行阶段不可变。如果需要使用变长的容器，可以使用标准库中的Vec/LinkedList等。数组类型的表示方式为[T;n]。其中T代表元素类型；n代表元素个数；它必须是编译期常量整数；中间用分号隔开。