ByteCoding

日常管理创建主题1$ kafka-topics --bootstrap-server localhost:9092 --create --topic t1 --partitions 1 --replication-factor 1 从Kafka 2.2版本开始，推荐使用--bootstrap-server代替--zookeeper（标记为已过期） 原因 使用--zookeeper会绕过Kafka的安全体系，不受认证体系的约束 使用--bootstrap-server与集群交互是未来的趋势 查询主题列表12345$ kafka-topics --bootstrap-server localhost:9092 --list__consumer_offsets_schemast1transaction 查询单个主题1234567$ kafka-topics --bootstrap-server localhost:9092 --describe --topic __consumer_offsetsTopic:__consumer_offsets PartitionCount:50 Replica ...

生产者消费者模式实现方式Object的wait/notify/notifyAll 基于Object的wait/notify/notifyAll与对象监视器（Monitor）实现线程间的等待和通知 这种方式实现的生产者消费者模式是基于内核实现的，可能会导致大量的上下文切换，性能不是最理想的 Lock中Condition的await/signal/signalAll 相对于Object的wait/notify/notifyAll，更推荐JUC包提供的Lock && Condition实现的生产者消费者模式 Lock && Condition实现的生产者消费者模式，是基于Java代码层实现的，在性能和扩展性方面更有优势 BlockingQueue 简单明了 限流算法漏桶算法通过限制容量池大小来控制流量，而令牌桶算法则通过限制发放令牌的速率来控制流量 漏桶算法 请求如果要进入业务层，就必须经过漏桶，而漏桶出口的请求速率是均衡的 如果漏桶已经满了，请求将会溢出，不会因为入口的请求量突然增加而导致系统 ...

线程上下文模式 线程上下文指的是贯穿线程整个生命周期的对象中的一些全局信息，如Spring中的ApplicationContext 可以使用ThreadLocal实现上下文 ThreadLocal是线程本地变量，可以实现多线程的数据隔离，每个线程只能访问各自内部的副本变量 Thread-Per-Message模式 一个消息一个线程 在Socket通信中，一个线程监听IO事件，每当监听到一个IO事件，就交给另一个处理线程执行IO操作 如果遇到高并发，就会出现严重的性能问题，因为线程在操作系统中也是昂贵的资源，不能无限制地创建 如果针对每个IO请求都创建一个线程来处理，在有大量请求同时进来时，就会创建大量线程 每次请求都需要创建和销毁线程，性能开销很大 可以使用线程池来代替线程的创建和销毁 ServerHandler1234567891011121314151617181920212223242526272829303132333435363738@AllArgsConstructorpublic class ServerHandler implements Runnable { ...

高水位水位的定义 经典教科书 在时刻T，任意创建时间（Event Time）为T'，且T'<=T的所有事件都已经到达，那么T就被定义为水位 《Streaming System》 水位是一个单调增加且表征最早未完成工作的时间戳 上图中标注为Completed的蓝色部分代表已经完成的工作，标注为In-Flight的红色部分代表正在进行中的工作 两者的边界就是水位线 在Kafka中，水位不是时间戳，而是与位置信息绑定的，即用消息位移来表征水位 Kafka中也有低水位（Low Watermark），是与Kafka删除消息相关联的概念 高水位的作用 两个作用 定义消息可见性，即用来标识分区下的哪些消息可以被消费者消费的 帮助Kafka完成副本同步 上图是某个分区Leader副本的高水位图，在分区高水位以下的消息被认为是已提交消息，反之为未提交消息 消费者只能消费已提交消息，即位移小于8的所有消息 暂不讨论Kafka事务，Kafka的事务机制会影响消费者所能看到的消息的范围，不只是简单依赖高水位来判断 而是依靠LSO（Log Stable Offset）的位移值来判 ...

原型模式 原型模式：通过给出一个原型对象来指明所创建的对象的类型，然后使用自身实现的克隆接口来复制这个原型对象 使用这种方式创新的对象的话，就无需再通过new实例化来创建对象了 Object类的clone方法是一个Native方法，可以直接操作内存中的二进制流，所以相对new实例化来说，性能更佳 实现原型模式12345678910111213141516171819202122232425262728class Prototype implements Cloneable { @Override public Prototype clone() { Prototype prototype = null; try { prototype = (Prototype) super.clone(); } catch (CloneNotSupportedException e) { e.printStackTrace(); } ...

控制器 控制器（Controller）是Kafka的核心组件，主要作用是在ZK的帮助下管理和协调整个Kafka集群 集群中任一Broker都能充当控制器的角色，但在运行过程中，只能有一个Broker成为控制器，行使管理和协调的职责 12345678910111213[zk: localhost:2181(CONNECTED) 1] get /controller{"version":1,"brokerid":0,"timestamp":"1571311742367"}cZxid = 0xd68ctime = Thu Oct 17 19:29:02 CST 2019mZxid = 0xd68mtime = Thu Oct 17 19:29:02 CST 2019pZxid = 0xd68cversion = 0dataVersion = 0aclVersion = 0ephemeralOwner = 0x1000209974b0000dataLength = 54numChildren = 0 Zooke ...

饿汉模式class1234567891011// 饿汉模式public final class Singleton { private static Singleton instance = new Singleton(); private Singleton() { } public static Singleton getInstance() { return instance; }} 使用了static修饰了成员变量instance，所以该变量会在类初始化的过程中被收集进_类构造器<clinit>_ 在多线程场景下，JVM会保证只有一个线程能够执行该类的<clinit>方法，其它线程将会被阻塞等待 等到唯一的一次<clinit>方法执行完成后，其它线程将不会再执行<clinit>方法，转而执行自己的代码 因此，static修饰的成员变量instance，在多线程的情况下能保证只实例化一次 在类初始化阶段就已经在堆内存中开辟了一块内存，用于存放 ...

触发重平衡 组成员数量发生变化 – 最常见 订阅主题数量发生变化 订阅主题的分区数发生变化 通知 重平衡过程是通过消费者的心跳线程通知到其它消费者实例的 Kafka Java消费者需要定期地发送心跳请求到Broker端的协调者，表明它还活着 在Kafka 0.10.1.0之前，发送心跳请求是在消费者主线程完成的，即调用poll方法的那个线程 弊端 消息处理逻辑是也在主线程完成的 一旦消息处理消耗了很长时间，心跳请求将无法及时发送给协调者，导致协调者误以为消费者已死 从Kafka 0.10.1.0开始，社区引入了单独的心跳线程 重平衡的通知机制是通过心跳线程来完成的 当协调者决定开启新一轮重平衡后，会将REBALANCE_IN_PROGRESS封装进心跳请求的响应中 当消费者实例发现心跳响应中包含REBALANCE_IN_PROGRESS，就知道重平衡要开始了，这是重平衡的通知机制 heartbeat.interval.ms的真正作用是控制重平衡通知的频率 消费者组状态机 状态 描述 Empty 组内没有任何成员，但消费者组可能存在已提交的位移数据，而且这些位移尚未过期 Dea ...

请求协议 Kafka自定义了一组请求协议，用于实现各种各样的交互操作 PRODUCE请求用于生产消息，FETCH请求用于消费消息，METADATA请求用于请求Kafka集群元数据信息 Kafka 2.3总共定义了45种请求格式，所有请求都通过TCP网络以Socket的方式进行通讯 处理请求方案顺序处理实现简单，但吞吐量太差，只适用于请求发送非常不频繁的场景 1234while (true) { Request request = accept(connection); handle(request);} 单独线程处理为每个请求都创建一个新的线程异步处理，完全异步，但开销极大，只适用于请求发送频率很低的场景 12345while (true) { Request request = accept(connection); Thread thread = new Thread(() -> { handle(request); }); thread.start();} Reactor模式 Reactor ...

副本机制的优点 提供数据冗余 即使系统部分组件失效，系统依然能够继续运转，增加了整体可用性和数据持久性 提供高伸缩性 支持横向扩展，能够通过增加机器的方式来提升读性能，进而提高读操作吞吐量 改善数据局部性 允许将数据放入与用户地理位置相近的地方，从而降低系统延时 Kafka只能享受副本机制提供数据冗余实现的高可用性和高持久性 副本定义 Kafka主题划分为若干个分区，副本的概念上是在分区层级下定义的，每个分区配置若干个副本 副本：本质上是一个_只能追加写消息的提交日志_ 同一个分区下的所有副本保存有相同的消息序列，这些副本分散保存在不同的Broker上，提高了数据可用性 实际生产环境中，每台Broker都可能保存有各个主题不同分区的不同副本 副本角色 Kafka采用基于领导者（Leader-based）的副本机制 副本分为两类：领导者副本（Leader Replica）和追随者副本（Follower Replica） 每个分区在创建时都要选举一个副本，称为领导者副本，其余的副本自动称为追随者副本 追随者副本是不对外提供服务的，所有的读写请求都必须发往领导者副本所在的Brok ...