kafka

Author: GitTsewell

Microservices/readme.md

@@ -0,0 +1,10 @@
+## 微服务架构
+### 服务注册和服务发现
+使用 K8S 的 Service 和 DNS:
+
+每个微服务 都在 K8S 中创建一个 Service ,名起名比如: user.xingren.host ,
+然后,其他微服务只需要 配置好这个 K8s 中的 Service name 即可,
+最后,只要这些微服务服务都在一个 k8S 集群中运行,便可省去注册中心与服务发现的这些微服务组件
+
+! [K8S 在微服务架构下做服务注册中心的一种思路](https://blog.csdn.net/itguangit/article/details/109731971)
+

golang/array/array_test.go

@@ -106,3 +106,18 @@ func TestSliceCopy(t *testing.T) {
 	fmt.Println(c, d)
 	fmt.Printf("c addres is %p , d address is %p \n", &c[0], &d[0])
 }
+
+func TestSliceAdd(t *testing.T) {
+	a := []int{0}
+	b := a
+
+	fmt.Printf("a address : %p,b address : %p\n", a, b)
+
+	a = append(a, 1)
+	a = append(a, 2)
+	a = append(a, 3)
+	a = append(a, 4)
+
+	fmt.Printf("a address : %p,b address : %p\n", a, b)
+
+}

golang/concurrent_programming/channel_test.go

@@ -24,3 +24,14 @@ func TestNilChannel(t *testing.T) {
 
 	time.Sleep(1 * time.Second)
 }
+
+func TestChannelBuffer(t *testing.T) {
+	ch := make(chan int, 5)
+
+	for i := 1; i <= 10; i++ {
+		fmt.Println(i)
+		ch <- i
+	}
+
+	fmt.Println(<-ch)
+}

golang/concurrent_programming/sync.md

@@ -479,4 +479,36 @@ func (g *Group) Do(key string, fn func() (interface{}, error)) (v interface{}, e
 ```
 原理很简单,先加锁,然后判断key是否在map中有值,如果有就waitGroup等待执行,如果没有就add(1)阻塞执行,最后解锁,然后调用call()
 
+# sync map
+结构模型
+```
+type Map struct {
+    mu Mutex
+    read atomic.Value // readOnly
+    dirty map[interface{}]*entry
+    misses int
+}
+```
+```
+type readOnly struct {
+	m       map[interface{}]*entry
+	amended bool // true if the dirty map contains some key not in m.
+}
+```
+
+两个map,优先存dirty中,触发一定条件后刷进read中
+
+每次操作dirty都有mutex锁
+
+![sync_map](./sync_map.png)
+
+流程概述:
+1. 读的时候先去readonly里面找,找不到就判断amended,如果为true,说明dirty里面可能有,就去dirty里面找,然后....
+2. 写的时候先判断readonly是否有,有直接更新,没有就去dirty里面新建一个
+3. 相信流程看下面文章链接
+
+! [年度最佳【golang】sync.Map详解](https://segmentfault.com/a/1190000023879083#item-4-2)
+
+# sync pool
+! [年度最佳【golang】sync.Pool详解](https://segmentfault.com/a/1190000023878185)
 

golang/concurrent_programming/sync_map.png

@@ -104,6 +104,36 @@ func TestErrGroup(t *testing.T) {
 	}
 }
 
+type Person struct {
+	Name string
+	Age  int
+}
+
+func initPool() *sync.Pool {
+	return &sync.Pool{
+		New: func() interface{} {
+			fmt.Println("创建一个 person.")
+			return &Person{}
+		},
+	}
+}
+
 func TestSyncPool(t *testing.T) {
-	//pool := sync.Pool{}
+	pool := initPool()
+	person := pool.Get().(*Person)
+	fmt.Println("首次从sync.Pool中获取person：", person)
+	person.Name = "Jack"
+	person.Age = 23
+	pool.Put(person)
+	pool.Put(person)
+	pool.Put(person)
+	pool.Put(person)
+	pool.Put(person)
+	pool.Put(person)
+	pool.Put(person)
+	pool.Put(person)
+	fmt.Println("设置的对象Name: ", person.Name)
+	fmt.Println("设置的对象Age: ", person.Age)
+	fmt.Println("Pool 中有一个对象，调用Get方法获取：", pool.Get().(*Person))
+	fmt.Println("Pool 中没有对象了，再次调用Get方法：", pool.Get().(*Person))
 }

golang/concurrent_programming/sync_test.go

@@ -39,5 +39,23 @@ func TestMapValue(t *testing.T) {
 func TestSyncMap(t *testing.T) {
 	m := sync.Map{}
 	m.Store(1, 1)
+	m.Store(2, 2)
 	m.Load(1)
+	m.Load(2)
+	m.Store(3, 3)
+	m.Store(4, 4)
+	m.Load(3)
+}
+
+func TestMapAdress(t *testing.T) {
+	m1 := map[int]int{1: 1}
+	m2 := m1
+
+	fmt.Printf("m1 address : %p,m2 address : %p\n", m1, m2)
+
+	for i := 10; i <= 10000; i++ {
+		m1[i] = i
+	}
+
+	fmt.Printf("m1 address : %p,m2 address : %p\n", m1, m2)
 }

golang/map/map_test.go

@@ -34,7 +34,7 @@ STW停止整个程序,从根节点开始遍历标记. STW-->遍历-->标记-->
 
 
 ## 垃圾收集的多个阶段
-1. 清理终止阶段；
+1. 清理终止阶段；z
   + 暂停程序，所有的处理器在这时会进入安全点（Safe point）；
   +  如果当前垃圾收集循环是强制触发的，我们还需要处理还未被清理的内存管理单元；
 2. 标记阶段；

golang/memory/gc.md

@@ -0,0 +1,160 @@
+#kafka
+整体分为三个部分分解,producer,broker,consumer
+
+## producer
+### 发送流程
+
+![producer_process](./image/producer_process.png)
+
+### 分区的好处
+1. 不同服务器broker的硬盘容量不同
+2. 提高并行度(发送,消费)
+
+### produce默认的分区策略
+1. 有指定按照指定发送
+2. 按照 key 的hash 取模
+3. sticky partition模式,先随机一个,然后一段时间内一直使用这个
+
+### producer提高吞吐量
+1. 配置batch.size 合适的批次大小
+2. liner.ms 等待时间 (默认是0ms)
+3. compression 压缩
+4. 缓冲区大小
+
+### producer的数据可靠性
+broker的ack应答模式有三种 0(落入缓冲区就应答) 1(leader partition写入磁盘就应答) all(所有partition写入磁盘才应答,一般情况下都用all)
+
+引出一个问题: 如果配置是all,然后同步过程中一个follower挂了,是不是leader会一直等待
+A: leader节点会维护ISR (in-sync replica set) 如果一个follower长时间没有和leader通信,会被踢到,默认30s
+
+#### 确保发送的数据完全可靠条件
+数据完全可靠条件 = AKC->ALL + follower partition >= 2 + ISR应答副本 >= 2
+
+#### producer数据重复发送问题
+![producer_repetition](./image/producer_repetition.png)
+leader节点在同步完所有follower之后正准备给producer返回ack,但是突然挂了,然后另一个follower成了leader,然后producer一直没收到这条消息的ACK,
+于是又发送了一次msg,造成数据重复发送
+
+解决方案:kafka内部支持幂等性,上面的情况再次发送kafka集群是不会存入broker的
+
+原理: 内部有一个实现幂等性的标准 集群的pid+partition+seq_number 不重复,但是这也只能保证不重启之后幂等,因为重启kafka之后,pid会变
+
+解决上述问题的方案: 用事务替代,事务可以自定义transaction_id替代pid实现幂等
+
+### producer发送数据的有序性
+目前kafka只能实现单个partition的有序性
+
+## broker
+### broker原理 (ZK版本), leader的选举
+1. kafka开启之后去zk注册对应的broker_id
+2. zk的controller中,那个kafka先注册那个说了算,controller决定了leader的选举
+3. 选举规则 在iSR存货为前提,按照AR中的顺序,来轮询作为leader节点
+4. controller将选举结果上传到ZK决定leader节点
+
+### broker上下线 
+ZK中 broker_ids管理节点
+#### broker 服役新节点
+问题: 之前有[0,1,2]三个broker,一个topic存储在[0,1,2]上,当新上线一个名为3的broker节点之后,怎么把之前的topic分配在3号broker上面的
+解答: 可以在配置文件中添加新节点
+
+#### broker 退役旧节点
+和服役新节点类似
+
+### broker的副本
+副本统称AR , AR = ISR + OSR (和leader通信超时的副本)
+
+leader的颗粒度,每一个partition都有一个leader
+
+### 节点故障处理
+![broker_follower_failed](./image/broker_follower_failed.png)
+
+LEO : 每个副本最后一个offset + 1
+
+HW : high watermark , 所有副本的最小LEO
+
+follower节点挂掉之后重新连接,需要追上HW之后,才能重新加入ISR
+
+leader节点故障挂掉之后,选出了一个新的leader,当前旧的leader节点需要把高于当前的HW丢掉,从新从leader同步
+
+### leader partition 自动平衡
+问题: 如果某些broker宕机,导致leader partition集中在少数broker上,会导致这几台服务器压力过高,造成集群不平衡
+
+解决:kafka有自动平衡机制,不平衡的leader比率默认10%,默认检查时间300s
+
+不平衡算法: 
+![leader_auto_balance](./image/leader_partition_auto_balance.png)
+
+生成环境一般不开启自动平衡算法,频繁的选举对性能消耗大,或者增大比率
+
+### kafka的文件存储
+
+#### topic和partition概念
+topic只是一个逻辑概念,partition才是一个物理概念
+
+![kafka_file_store](./image/kafka_file_store.png)
+
+#### 文件存储和index索引
+![index](./image/kafka_index.png)
+
+#### 文件保存
+默认保存7天,7天之后默认删除,也可以配置成压缩
+
+### kafka如何搞笑读写数据
+1. 本身是分布式集群,采用了分区技术,数据分散,并行度高
+2. 读数据采用稀疏索引,快速定位消费数据
+3. 采用顺序写磁盘
+4. 页缓存+零拷贝技术
+![page_cache](./image/page_cache.png)
+
+## kafka consumer
+### process
+kafka采用pull的模式,好处是consumer可以自己控制速度,不足是如果kafka没有数据,consumer会陷入循环
+
+![producer_process](./image/consumer_process.png)
+
+### 消费组
+每个组内负责消费不同分区数据,一个分区只能由一个消费组内消费
+
+#### 消费组初始化流程
+![CG_init](./image/GC_init.png)
+
+#### 消费组消费流程
+![CG_process](./image/GC_process.png)
+
+### 消费者分区以及再平衡
+问题: 一个CG(consumer group)里面有多个consumer,到底那个consumer消费那个partition呢
+![producer_balance](./image/producer_balance.png)
+
+1. range方案
+![range](./image/consumer_range.png)
+2. roundRobin : hash code 排序分配
+3. sticky: 和range类型,不过是随机乱序分配
+
+### offset的维护
+以前的存储在zk中,现在存储在一个topic中
+
+#### offset的自动提交
+默认是开启自动提交,默认一个MSG到达kafka的partition之后,每隔5S自动提交offset
+
+#### offset的手动提交
+分为同步提交和异步提交
+
+#### 指定offset消费
+分为指定offset消费和指定时间消费,注意的是客户端需要等到分区初始化完成之后再设置需要指定的offset才生效
+
+### consumer的重读消费问题
+![consumer_repetition](./image/consumer_repetition.png)
+使用消费事务解决,在数据操作落库之后再提交事务
+
+### consumer数据积压问题
+![consumer_data](./image/consumer_data.png)
+
+## kafka的监控
+可以使用kafka-eagle监控kafka状态,又名EFAK
+
+## kafka新模式  KRAFT
+去掉了依赖的ZK,原理就是用一个broker节点成为controller,让这个替代ZK所做的工作,这个成为的controller的broker也可以成为一个partition节点
+
+
+
+