
[toc]

通过数据结构、实现原理、读写操作来了解go hashmap

# 数据结构

![](/images/blog/golang/assert/go-map.jpg)

hash有2个关键数据结构: `hmap` `bmap`

**hmap:  `runtime/map.go`**

```go
type hmap struct {
	count     	int 
	flags     	uint8
	B         	uint8  
	noverflow 	uint16 
	hash0     	uint32 
	buckets    	unsafe.Pointer 
	oldbuckets 	unsafe.Pointer
	nevacuate  	uintptr
	extra 		*mapextra 
}
```

-   `count`  元素数量
-   `B`  2^B个buckets桶
-   `noverflow` buckets溢出桶的数量，近似值
-   `buckets` 桶
-   `oldbuckets` 扩容时指向原buckets桶

**bmap: `runtime/map.go`  `cmd/compile/internal/gc/reflect.go`**

```go
type bmap struct {
    topbits 	[8]uint8
    keys 		[8]keytype
    elems 		[8]elemtype
    pad 		uintptr
    overflow	uintptr
}
```

哈希表中桶的真正结构其实是在编译期间运行的函数 `bmap` 中被『动态』创建的， 代码在`cmd/compile/internal/gc/reflect.go`

-   `topbits` 存储hash值的高8位，通过比对高8位减少键值对访问次数以提高性能
-   `keys` / `elems` 数组

-   `pad` 内存对齐
-   `overflow` 溢出桶，map存储数据过多时使用

# 实现原理

时间复杂度: O(1)

hash函数和hash冲突解决方法

**hash函数**

实现哈希表的关键点在于如何选择哈希函数，哈希函数的选择在很大程度上能够决定哈希表的读写性能，在理想情况下，哈希函数应该能够将不同键映射到不同的索引上，这要求**哈希函数输出范围大于输入范围**，但是由于键的数量会远远大于映射的范围，所以在实际使用时，这个理想的结果是不可能实现的。

**hash冲突**

开放寻址法：对数组中的元素依次比较键值对是否存在于数组

拉链法： 使用数组加上链表

![](/images/blog/golang/assert/hashmap-拉链法.png)

# 读写操作

**读**

![](/images/blog/golang/assert/go-map-get.jpg)

-   计算出key的hash
-   用最后的“B”位来确定在哪个桶（“B”就是前面说的那个，B为4，就有16个桶，0101用十进制表示为5，所以在5号桶）
-   根据key的前8位快速确定是在哪个格子（额外说明一下，在bmap中存放了每个key对应的tophash，是key的前8位）
-   最终还是需要比对key完整的hash是否匹配，如果匹配则获取对应value
-   如果都没有找到，就去下一个overflow找

**写**

-   通过key的后“B”位确定是哪一个桶
-   通过key的前8位快速确定是否已经存在
-   最终确定存放位置，如果8个格子已经满了，没地方放了，那么就重新创建一个bmap作为溢出桶连接在overflow

**扩容**

条件：

-   装载因子大于6.5
-   溢出桶 大于15个

```go
func mapassign(t *maptype, h *hmap, key unsafe.Pointer) unsafe.Pointer {
	...
	if !h.growing() && (overLoadFactor(h.count+1, h.B) || tooManyOverflowBuckets(h.noverflow, h.B)) {
		hashGrow(t, h)
		goto again
	}
	...
}
```

方式：

-   等量扩容
-   翻倍扩容



知道map数据结构长什么样才能知道是怎么访问的, 冲突时如何解决的

go基础数据结构 hashmap数据结构与实现原理


[toc]

# redis延迟高问题解决思路

延迟是 客户端从发出命令与客户端接收命令的回复之间的最大延迟，通常redis的处理时间是非常低的，但是存在高延迟的某些条件

解决思路：
1. 确保没有慢执行的命令，通过redis慢查询日志查看
2. 对于EC2用户，EC2是基于HVM，否则fork太慢了
3. 禁用transparent_hugepage，`echo never > /sys/kernel/mm/transparent_hugepage/enabled`, 然后重启redis
4. 可能存在与redis无关的内在延迟，`redis-cli --intrinsic-latency 100`, 需要redis-server服务器上运行
5. aof/rdb 持久化 性能/延迟的权衡
   1. AOF+fsync 总是很慢，不推荐使用
   2. AOF+fsync+everysec 每秒执行一次，推荐(也可能会造成阻塞的问题,[当执行bgrewrite对aof进程归并优化时，redis出现被阻塞的问题](./持久化.md#1))
   3. AOF+fsync+everysec+no-appendfsync-no-write 主进程不调用fsync()操作，最长不超过30秒，推荐
   4. AOF+fsync never
   5. RDB 配置触发save条件

## 网络和通信引起的延迟
1Gbit/s的网络延迟大约为200us，uninx域套接字的延迟低至30us，如果在同一个服务器可以用unix_socket加速，redis启用unix_socket

`redis.conf`
```conf
unixsocket /var/run/redis/redis.sock
unixsocketperm 777
```
php 连接redis
```php
$redis->connect('/var/run/redis/redis.sock');
```
解决方案：
- 优先选择物理机托管服务器
- 不要系统的连接/断开服务器
- 如果客户端与服务器位于同一主机上，使用域套接字
- 尽量使用MSET/MGET批量操作
- redis不适合绑定在单个cpu上,bgsave/aof重写会抢占主进程的时间

## redis单线程特性
redis主要使用单线程设计，多路复用为客户端的请求提供服务，所有请求都是安装顺序提供

## 慢查询导致的延迟
单线程的结果是，当有慢查询时所有其他客户端将等待该请求完成

如果有慢查询时必须的可以放在salve上执行所有慢查询

查看慢查询日志
```shell
slowlog get 2
```
sort\lrem\sunion\keys 这些都慢查询命令，用scan代替keys

## fork生成的延迟
开启RDB和AOF重写，redis必须fork后台进程，fork操作可能会阻塞主进程，fork涉及到复制大量对象，尤其对于使用虚拟内存机制的页表
linux/64，内存为4kb每页，每个进程存储一个页表，对于24G内存的redis实例需要(24GB/4kb)*8=48M,每次fork都会分配和复制48M内存

## transparent_hugepage 引起的延迟
transparent_hugepage linux内核尽可能的尝试分配2M的页大小给程序使用，内核空间在内存中自身是2M对齐的，目的是减少内核TLB的压力(cpu使用的一小块关联内存)，如果内存没有2M的连续大小的空间可分配，内核会回退到分配4kB

禁用transparent_hugepage[](/linux/页缓存.md#pdflush)
```shell
echo never > /sys/kernel/mm/transparent_hugepage/enabled
```
strace只显示慢速系统调用:
```shell
sudo strace -f -p $(pidof redis-server) -T -e trace=fdatasync,write 2>&1 | grep -v '0.0' | grep -v unfinished
```

## swap内存交换引起的延迟
- redis使用的内存比可用内存多
- 某些进程产生了大量的读取和写入I/O，因为文件是会被缓存的，会增加内核压力增大文件内存缓存，产生内存交换，包括RDB和AOF

[vmstat](/article/f46716e9f100c184924ca4680e032d84)
```shell
[root@00h ~]# vmstat 5
procs -----------memory---------- ---swap-- -----io---- -system-- ------cpu-----
 r  b   swpd   free   buff  cache   si   so    bi    bo   in   cs us sy id wa st
 2  0      0  93016  16576 212056    0    0    81   151    1   11  2  0 98  0  0
 0  0      0  92892  16588 212116    0    0    44     0  288  591  1  0 98  1  0
 0  0      0  92892  16616 212068    0    0    28     0  322  663  0  1 99  0  0
 0  0      0  92892  16628 212084    0    0    12     0  326  650  1  1 98  0  0
^C
```
swap si/so 为0,说明swap没有内存交换活动

[iostat](/article/11a1575fd553f8e69b846aea8248d146)
```shell
[www@00h www.00h.tv]$ iostat -xk 1
Linux 3.10.0-957.1.3.el7.x86_64 (00h) 	06/15/2019 	_x86_64_	(1 CPU)

avg-cpu:  %user   %nice %system %iowait  %steal   %idle
           1.54    0.00    0.43    0.20    0.00   97.83

Device:         rrqm/s   wrqm/s     r/s     w/s    rkB/s    wkB/s avgrq-sz avgqu-sz   await r_await w_await  svctm  %util
vda               0.16     1.14    1.58    1.71    81.23   150.43   140.57     0.04   10.98   14.13    8.08   0.97   0.32
```
如果延迟问题是由于redis内存在磁盘上交换造成，需要降低系统内存压力，增加更多的内存

## AOF和磁盘I/O造成的延迟
AOF有两个系统调用：

write：将数据写入aof文件，产生阻塞的情况：1. 当系统缓冲区已满 2. 系统正在将缓冲区的内容写入到文件

fdatasync: 刷新内核文件缓冲区的文件到磁盘上，相比fsync要[高效](/article/6db15910cac7885f6a97958db5ac631c#性能)

redis可以配置3种fsync：
- appendfysnc=no redis不执行fsync，再次配置中，唯一的延迟源是write
- appendfsync=everysec redis每秒执行一次，使用不同的线程，如果fsync仍在进行中，redis使用缓冲区将write调用延迟两秒（如果fsync正在针对同一文件进行，write会在Linux上阻塞），如果fsync占用时间太长，redis最终也会执行write
- appendfsync=always 每次写入都执行fsync，这种情况下性能一般非常低

可以使用[strace](/article/7109e41fccb9190faa5d897be3e3df76)查看进程对fdatasync/write的调用
```
sudo strace -p $(pidof redis-server) -T -e trace=fdatasync，write
```

## redis过期产生的延迟
redis使key过期的方式：
- 惰性淘汰：在key请求时，判断key是否失效，失效就淘汰
- 定时策略：每隔100毫秒检测一次

redis没100毫秒会启动一个serverCron来执行定期淘汰策略
- `ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP` 移除所有已过期的键
- 如果发现超过25%的键过期，会重复执行

正常情况下,`ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP`默认情况下设置为20，每秒执行10次，通常每秒只有200个密钥主动过期，每秒200个key过期对redis实例是没有影响的，
然而算法是自适应的，如果发现该组采样key中发现超过25%的key已经过期，该算法会循环，如果数据库有超过25%的key可以在同一秒过期，并且这些key占当前key的25%

# 延迟监控 & 慢查询日志
设置延迟阀值(单位：毫秒)，只有超过阀值的事件才会被记录成为延迟峰值，根据需要设置
```conf
config set latency-monitor-threshold 100
```

redis慢查询配置:

-   slowlog-log-slower-than 单位毫秒，默认10毫秒，超过的会被记录，对于高OPS的场景可以设置为1毫秒
-   Slowly-max-len 最多纪录多少条, 线上可设置为1000条以上

获取慢查询日志：

```redis
slowlog get [n]
```

# redis测量延迟工具

```shell
[work@cpc-dev01 ~]$ redis-cli -p 6379 --intrinsic-latency 100
Max latency so far: 1 microseconds.
Max latency so far: 3 microseconds.
Max latency so far: 4 microseconds.
Max latency so far: 5 microseconds.
Max latency so far: 9 microseconds.
Max latency so far: 13 microseconds.
Max latency so far: 17 microseconds.
Max latency so far: 28 microseconds.
Max latency so far: 110 microseconds.
Max latency so far: 159 microseconds.
Max latency so far: 548 microseconds.
^C
685295282 total runs (avg latency: 0.1459 microseconds / 145.92 nanoseconds per run).
Worst run took 3755x longer than the average latency.
```
这个命令要在服务器上运行

# latency 查看延迟信息
- `latency latest` 最新延迟事件
- `latency history command` 从事件时间序列中获取原始数据非常有用
- `latency reset` 不带参数，重置所有事件
- `latency graph command` 生成ASCII图形
```shell
#
|
|
|_

11
77
mm
```
这个图要竖着看，17m 代表17分钟前发生

# 延迟诊断 
`latency history`

# 推荐阅读
- [Redis latency problems troubleshooting](https://redis.io/topics/latency)


# 总结
1. 有时候也需要注意一下redis服务器的流量负载情况，由于redis本身的高性能服务，可能会出现直接把服务器带宽打满的情况
2. 注意请求的复杂度，建议将O(n)的甚至是更复杂的操作，排序，多键操作等指令转化放在redis slave上
   - 案例1: 项目推送模块redis优化，将ZREVRANGE(O(log(N)+M)， N 为有序集的基数，而 M 为结果集的基数)操作转换为list(O(1))的操作，QPS从5000上升到6500
   - 在线用户统计模块，包含了大量的keys的遍历操作，取消keys操作以后，吞吐量回升
3. 不必要的操作,`EXISTS`、`DEL`这样的操作，很多场景下是不需要的，QPS不必要的情况，取消这些可以提升有效用户请求的数量
   - XX项目点播项目redis实例，平均存在37%的EXISTS请求，每次GET、ZRANG、HMGET前都会执行，用于判断是否存在，redis对这些不存在的key都会提示，可以直接取消EXISTS请求
   - XX项目推送业务redis实例中存在25.5%的DEL操作，占据40%的时间，后将key直接设置过期时间
   - 具体案例具体分析，存在必有其道理
4. IO优化，主要是redis快照和AOF产生的，根据数据的主要成读来设置快照频率，或者快照值在slave上发生
5. CPU的使用频率，redis是单线程作业，除了分离业务、读写分离。`keepalive + twemproxy` 也能有效提升redis的CPU利用率

redis_延迟高问题解决思路_'latency_monitor_threshold'


kafka的应用场景

- 实时日志聚合，支持高容量事件流
- 处理大量的数据积压，支持来自离线系统的周期性数据加载
- 处理低延迟分发，分区，分布式，以及实时处理



### 如何处理大量的数据积压？

熟知的mysql是通过B+树来持久化到磁盘的，查找数据需要从磁盘读取并构建B+树内存数据结构。而Kafka则是：使用linux文件系统和pagecache，放弃内存数据结构

- 即使进程维护了内存数据结构，该数据也可能会被复制到操作系统的 pagecache 中，事实上所有内容都被存储了两份。

- 通过自动访问所有空闲内存将可用缓存的容量至少翻倍，并且通过存储紧凑的字节结构而不是独立的对象，有望将缓存容量再翻一番。 这样使得32GB的机器缓存容量可以达到28-30GB,并且不会有额外的 GC 负担
- 保持 内存cache 和 文件系统之间一致性逻辑，不需要做过多的转化。

- 简单的读写文件，读写操作不会相互影响。数据的大小不在会影响性能。可以无限量的使用物理硬盘，价格便宜，容量充足。可以充分解决大量的数据积压难题



### 如何在1秒内接收并持久化这1亿条数据呢？

存储问题解决了，面临的问题：大量的I/O和字节拷贝

I/O操作主要是 客户端与服务端的消息传递，服务端的数据持久化。

将多个消息打包成一组消息块，一次请求发送一组消息，服务端接收，consumer每次获取多组消息块，大量的I/O操作将不在是主要的性能问题，其实主要矛盾是在服务带宽问题上。producer/broker/consumer 三者使用二进制数据传输，producer 和 consumer 之间使用相同的压缩算法也可以极大的节省带宽问题

sendfile 将数据从 pagecache 直接发送到网络，这样避免重新复制数据。



每秒1亿条数据，保存7天，kafka是如何处理的？


[toc]

# 安装

**zookeeper** 下载地址: [http://zookeeper.apache.org/releases.html#download](http://zookeeper.apache.org/releases.html#download)

解压：`tar -zxvf kafka_2.12-2.6.0.tgz`

运行：`./bin/zkServer.sh --config conf start`

> --config 指的是配置目录



**kafka** 下载地址: [https://kafka.apache.org/downloads](https://kafka.apache.org/downloads)

解压：`tar -zxvf kafka_2.12-2.6.0.tgz`

运行：`./bin/kafka-server-start.sh config/server.properties`

# 配置

**kafka配置：**

`kakfa-configs.sh` 可以对kafka配置查看或修改

别名： 每次敲这么多命令很烦

`alias kconfigs="~/Documents/kafka/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092"`

查看:  --describe

`kconfigs --describe --entity-type [topics/clients/users/brokers/broker-loggers] --all`

- brokers
  - log.dirs：指定文件存储目录, 多个逗号分隔，建议目录指定不同的磁盘
    - 提升读写性能，多个磁盘同时读写的性能更高
    - 故障转移：Failover 1.1 版本添加
  - listeners：通过什么协议访问kafka服务，内网服务指定
  - advertised.listeners：外网访问kafka服务指定
  - log.retension.hours=168：记录保留时长 默认7天（hours/minutes/ms）
  - log.retension.bytes=1073741824：可用磁盘大小 默认 128G， -1不限制
  - zookeeper.connect=localhost:2181：zookeeper连接地址
  - auto.create.topics.enable: false topic不存在时是否自动创建（建议显示配置false）
  - unclean.leader.election.enblae: false 允许落后的副本参与leader选举（建议显示配置false）
  - auto.leader.rebalance.enable: false 定期leader选举，正常服务的leader到期也会被替换掉（建议显示配置false）
- topics
  - retension.hours 保留时长
  - retension.bytes 为该Topic预留多大的磁盘空间，-1无限使用磁盘空间
- users
- clients
- broker-loggers

修改：--alter

`kconfigs --alter --entity-type topics --entity-name test --add-config max.message.bytes=104858800`

- --alter：代表修改
- --entity-type: 代表上面这5中实例
- --entity-name: 指定那个broker（broker.id）, topic (topic.name)
- --add-config: --help 可以查看支持实时修改的配置

**JVM配置**：

- 堆大小(heap size)配置：建议6G  环境变量
  - `export KAFKA_HEAP_OPTS=--Xms6g  --Xmx6g`
- 垃圾回收：G1收集器 
  - `export  KAFKA_JVM_PERFORMANCE_OPTS= -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=true`

**系统配置**:

- 文件描述符限制：`ulimit -n`
  - `/etc/security/limits.conf` 
    - `soft nofile 102400`
    - `hard nofile 104800`
- 文件系统类型： xfs 性能较好
- swap内存：建议配置成1，方便报警
- 脏页刷新时间：默认5秒

- 磁盘大小
  - 新增消息数
  - 消息留存时间
  - 平均消息大小
  - 备份数
  - 是否启用压缩

# 本地集群

本地运行多个实例，需要指定不同的端口，和broker.id, 创建多个server.propertie，指定端口

```
config/server-1.properties:
  broker.id=1
  listeners=PLAINTEXT://:9093
  log.dir="～/Documents/kafka/data1"
```

```
config/server-2.properties:
  broker.id=2
  listeners=PLAINTEXT://:9094
  log.dir="~/Documents/kafka/data2"
```

启动：

```shell
./bin/zookeeper-server-start.sh -daemon config/zookeeper.properties 
./bin/kafka-server-start.sh -daemon config/server.properties
./bin/kafka-server-start.sh -daemon config/server1.properties 
./bin/kafka-server-start.sh -daemon config/server2.properties 
```

创建：3副本/1分区 topic

> alias ktopics="~/Documents/kafka/bin/kafka-topics --bootstrap-server localhost:9092,localhost:9093,localhost:9094"

```shell
ktopics --create --replication-factor 3 --partitions 1 --topic test
```

查看：topic状态

```shell
ktopics --descibe --topic test

Topic: test	PartitionCount: 1	ReplicationFactor: 3	Configs: segment.bytes=1073741824
	Topic: test	Partition: 0	Leader: 3	Replicas: 2,3,1	Isr: 3,1,2
```

# kafka 分区策略 & 压缩算法

分区策略是决定生产者将消息发送到哪个分区的算法

常见的分区策略:

1. 轮询策略，能保证消息尽可能分配到不同的分区上 
2. 随机策略，随机返回小于partition的数，不能保证均匀性，适用及其性能不一的情形
3. hash策略，具有相同key的消息会保证在同一个分区中，这样可以保证消息的前后顺序

go语言包：[github.com/Shopify/sarama](https://pkg.go.dev/github.com/Shopify/sarama)  提供了这3种分区策略的实现

**轮询:** `func NewRoundRobinPartitioner(topic string) Partitioner`

**随机:** `func NewRandomPartitioner(topic string) Partitioner`

**hash:** `func NewHashPartitioner(topic string) Partitioner`

**指定分区** `func NewManualPartitioner(topic string) Partitioner`



压缩：

保证Producer 和 Consumer 压缩算法一致

```go
const (
	//CompressionNone no compression
	CompressionNone CompressionCodec = iota
	//CompressionGZIP compression using GZIP
	CompressionGZIP
	//CompressionSnappy compression using snappy
	CompressionSnappy
	//CompressionLZ4 compression using LZ4
	CompressionLZ4
	//CompressionZSTD compression using ZSTD
	CompressionZSTD
)
```

吞吐量： LZ4 > Snappy > zstd > GZIP

压缩比：zstd > LZ4 > GZIP > Snappy

CPU:  各算法差不多压缩时 Snappy 算法使用的 CPU 较多, 解压缩时 GZIP 算法可能使用更多的 CPU



# FAQs

**如何保证消息不丢失?**

producer: 

1. 消息发送异常处理, [sarama](https://github.com/Shopify/sarama)提供了2种发送消息的方式，sync/async，同步如果发送失败会返回error，异步提供了 `Errors() <-chan *ProducerError` 接口来处理错误
2. 消息发送异常重试, `config.Producer.Retry`, 默认是 3次
3. `config.Producer.RequiredAcks=WaitForAll` 所有副本都接受到消息，该消息才算做提交
   1. `NoResponse RequiredAcks = 0`
   2. `WaitForLocal RequiredAcks = 1`
   3. `WaitForAll RequiredAcks = -1`

broker:

1. `unclean.leader.election.enable=false` 控制落后台太多的broker不参加leader选举
2. `replication.factor >= 3` 副本数量不要少于3份
3. `min.insync.replicas >= 1` 消息最少写入多少个副本才算做已提交，正式环境中建议>=3
4. `replication.factor > min.insync.replicas`  如果2着相等，只要有1个副本挂掉，整个分区就无法工作了

consumer:

1. 先消费，在更新位移, `cfg.Consumer.Offsets.AutoCommit.Enable = false` 默认是自动提交的


kafka生产环境配置


[TOC]

# 	时间格式常量

## 标准

```go
const (
    ANSIC       = "Mon Jan _2 15:04:05 2006"
    UnixDate    = "Mon Jan _2 15:04:05 MST 2006"
    RubyDate    = "Mon Jan 02 15:04:05 -0700 2006"
    RFC822      = "02 Jan 06 15:04 MST"
    RFC822Z     = "02 Jan 06 15:04 -0700" // RFC822 with numeric zone
    RFC850      = "Monday, 02-Jan-06 15:04:05 MST"
    RFC1123     = "Mon, 02 Jan 2006 15:04:05 MST"
    RFC1123Z    = "Mon, 02 Jan 2006 15:04:05 -0700" // RFC1123 with numeric zone
    RFC3339     = "2006-01-02T15:04:05Z07:00"
    RFC3339Nano = "2006-01-02T15:04:05.999999999Z07:00"
    Kitchen     = "3:04PM"
    // Handy time stamps.
    Stamp      = "Jan _2 15:04:05"
    StampMilli = "Jan _2 15:04:05.000"
    StampMicro = "Jan _2 15:04:05.000000"
    StampNano  = "Jan _2 15:04:05.000000000"
)
```

## 时/分/秒/毫秒/微妙/纳秒

```go
const (
    Nanosecond  Duration = 1
    Microsecond          = 1000 * Nanosecond
    Millisecond          = 1000 * Microsecond
    Second               = 1000 * Millisecond
    Minute               = 60 * Second
    Hour                 = 60 * Minute
)
```

## 月份

```go
const (
	January Month = 1 + iota
	February
	March
	April
	May
	June
	July
	August
	September
	October
	November
	December
)
```

## 星期几

```go
const (
	Sunday Weekday = iota
	Monday
	Tuesday
	Wednesday
	Thursday
	Friday
	Saturday
)
```

# 超时控制 After

```go
var c chan int

func handle(int) {}

func main() {
	select {
	case m := <-c:
		handle(m)
	case <-time.After(10 * time.Second):
		fmt.Println("timed out")gg
	}
}
```

# 时间间隔 Duration

time.Add(t Duration)传参需要使用Durtaion 

## time.Sub

计算两个时间差，返回duration

```go
func expensiveCall() {}

func main() {
	t0 := time.Now()
	expensiveCall()
	t1 := time.Now()
	fmt.Printf("The call took %v to run.\n", t1.Sub(t0))
}
```

## ParseDuration

有效符号："ns", "us" (or "µs"), "ms", "s", "m", "h".

```go
func main() {
	hours, _ := time.ParseDuration("10h")
	complex, _ := time.ParseDuration("1h10m10s")
	micro, _ := time.ParseDuration("1µs")
	// The package also accepts the incorrect but common prefix u for micro.
	micro2, _ := time.ParseDuration("1us")

	fmt.Println(hours)
	fmt.Println(complex)
	fmt.Printf("There are %.0f seconds in %v.\n", complex.Seconds(), complex)
	fmt.Printf("There are %d nanoseconds in %v.\n", micro.Nanoseconds(), micro)
	fmt.Printf("There are %6.2e seconds in %v.\n", micro2.Seconds(), micro)
}
```

# 时间 Time

基础结构体:

```go
type Time struct {
	wall uint64
	ext  int64
	loc *Location
}
```

## Date

返回指定日期的time

```go
func main() {
	t := time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC)
	fmt.Printf("Go launched at %s\n", t.Local())
}
// Go launched at 2009-11-11 07:00:00 +0800 CST
```

## 当前时间 Now

返回当前时间,这个用的最多

## 时间解析 Parse & ParseInLocation

```go
func Parse(layout, value string) (Time, error)
```

Layout 时间格式，value时间值，解析时间返回time

```go
func main() {
	t, _ := time.Parse("2006-01-02", "2020-12-05")
	fmt.Println(t)
  m, _ := time.Parse("2006-01-02 15:04:05", "2020-12-05 11:23:08")
	fmt.Println(m)
}
// 2020-12-05 00:00:00 +0000 UTC
// 2020-12-05 11:23:00 +0000 UTC
```

上面的用法其实是错误的，没有时区，正确的做法是：

```go
func main() {
	t, _ := time.ParseInLocation("2006-01-02", "2020-12-05")
	fmt.Println(t)
  m, _ := time.ParseInLocation("2006-01-02 15:04:05", "2020-12-05 11:23:08")
	fmt.Println(m)
}
// 2020-12-05 00:00:00 +0800 CST
// 2020-12-05 11:23:00 +0800 CST
```

## 时间戳 Unix

解析时间戳为time

```go
func main() {
	t := time.Unix(1576053402, 100000000)
	fmt.Print(t)
}
// 2019-12-11 16:36:42.01 +0800 CST
```

## 格式化 Format

时间格式化

默认输出格式：`2006-01-02 15:04:05.999999999 -0700 MST`

```go
func main() {
	fmt.Printf("当前时间: %s\n", time.Now())
	fmt.Printf("RFC1123Z: %s\n", time.Now().Format(time.RFC1123Z))
	fmt.Printf("RFC3339: %s\n", time.Now().Format(time.RFC3339))
}

// 当前时间: 2019-12-11 16:25:53.987141 +0800 CST m=+0.000513629
// RFC1123Z: Wed, 11 Dec 2019 16:25:53 +0800
// RFC3339: 2019-12-11T16:25:53+08:00
```

## 时间加减 Add

当前时间加上 ？

```go
func main() {
	start := time.Date(2009, 1, 1, 12, 0, 0, 0, time.UTC)
	afterTenSeconds := start.Add(time.Second * 10)
	afterTenMinutes := start.Add(time.Minute * 10)
	afterTenHours := start.Add(time.Hour * 10)
	afterTenDays := start.Add(time.Hour * 24 * 10)

	fmt.Printf("start = %v\n", start)
	fmt.Printf("start.Add(time.Second * 10) = %v\n", afterTenSeconds)
	fmt.Printf("start.Add(time.Minute * 10) = %v\n", afterTenMinutes)
	fmt.Printf("start.Add(time.Hour * 10) = %v\n", afterTenHours)
	fmt.Printf("start.Add(time.Hour * 24 * 10) = %v\n", afterTenDays)
}
```

## 判断时间是否相等 Equal

判断两个时间是否一致

```go
func main() {
	secondsEastOfUTC := int((8 * time.Hour).Seconds())
	beijing := time.FixedZone("Beijing Time", secondsEastOfUTC)

	// Unlike the equal operator, Equal is aware that d1 and d2 are the
	// same instant but in different time zones.
	d1 := time.Date(2000, 2, 1, 12, 30, 0, 0, time.UTC)
	d2 := time.Date(2000, 2, 1, 20, 30, 0, 0, beijing)

	datesEqualUsingEqualOperator := d1 == d2
	datesEqualUsingFunction := d1.Equal(d2)

	fmt.Printf("datesEqualUsingEqualOperator = %v\n", datesEqualUsingEqualOperator)
	fmt.Printf("datesEqualUsingFunction = %v\n", datesEqualUsingFunction)
}
```

## Before/After

判断时间A是否在时间B之前还是之后，返回bool

```go
func main() {
	a := time.Now()
	b := time.Now()

	fmt.Printf("a 在 b 之前: %v\n", a.Before(b))
	fmt.Printf("a 在 b 之后: %v\n", a.After(b))
}

// a 在 b 之前: true
// a 在 b 之后: false
```

## 时间四舍五入 Round

时间4舍5入

```go
func main() {
	t := time.Date(0, 0, 0, 12, 15, 30, 918273645, time.UTC)
	round := []time.Duration{
		time.Nanosecond,
		time.Microsecond,
		time.Millisecond,
		time.Second,
		2 * time.Second,
		time.Minute,
		10 * time.Minute,
		time.Hour,
	}

	for _, d := range round {
		fmt.Printf("t.Round(%6s) = %s\n", d, t.Round(d).Format("15:04:05.999999999"))
	}
}
```

## 舍去 Truncate

```go
func main() {
	t, _ := time.Parse("2006 Jan 02 15:04:05", "2012 Dec 07 12:15:30.918273645")
	trunc := []time.Duration{
		time.Nanosecond,
		time.Microsecond,
		time.Millisecond,
		time.Second,
		2 * time.Second,
		time.Minute,
		10 * time.Minute,
	}

	for _, d := range trunc {
		fmt.Printf("t.Truncate(%5s) = %s\n", d, t.Truncate(d).Format("15:04:05.999999999"))
	}
	// To round to the last midnight in the local timezone, create a new Date.
	midnight := time.Date(t.Year(), t.Month(), t.Day(), 0, 0, 0, 0, time.Local)
	_ = midnight
}
```

# 定时器 Tick 

```go
func statusUpdate() string { return "" }

func main() {
	c := time.Tick(5 * time.Second)
	for now := range c {
		fmt.Printf("%v %s\n", now, statusUpdate())
	}
}
```



go time常用方法


Etcd几个关键特征：

**raft算法**：Leader选举/操作同步，无论访问任意节点，都将获得最终一致性数据，高度可靠。

**mvcc**:  etcd有2级存储结构 [B树内存索引](https://github.com/google/btree)  / [bbolt B+树](https://github.com/etcd-io/bbolt) ，内存中使用btree维护key/value索引，节点中存的是bbolt中键值K，通过这个k在bbolt查找，得到的才是用户传进去的value值

- 每个事务有唯一事务id，就是下面的main id int64
- 一个事务可以包含多个修改操作（put/delete），每一个操作就是一个revision, 共享同一个main id
- 一个事务内多个连续修改操作会从0开始递增，编号 sub id int64

```
type revision struct {
	// main is the main revision of a set of changes that happen atomically.
	main int64

	// sub is the sub revision of a change in a set of changes that happen
	// atomically. Each change has different increasing sub revision in that
	// set.
	sub int64
}

type keyIndex struct {
	key         []byte
	modified    revision // the main rev of the last modification
	generations []generation
}
```

内存索引中，每个原始key会关联一个key_index，里面维护了多版本信息

**事务**: 

- 只读事务
- 读写事务



# 概要

- etcd集群

- 配置UI管理界面

- 监控与维护

- 容灾

#### 应用场景

- 服务发现
- 配置中心
- 分布式锁
- 监控机器活动

#### 基础功能

- 监控 watch
- 租约 lease
- 全局锁 lock
- 选举 elect
- 集群状态 
  - `etcdctl --write-out=table endpoint status`
  - `etcdctl endpoint health`
- 快照 snapshot
  - `etcdctl --endpoints=127.0.0.1:2379 snapshot save my.db`
  - `etcdctl --endpoints=127.0.0.1:2379 --write-out=table snapshot status my.db`
- 迁移 migrate
- 成员管理 member
  - `etcdctl --endpoints=127.0.0.1:2379 member list`

- 授权 auth



# 集群配置

`docker-compose.yml`

```yaml
services:
  etcd1:
    image: 'bitnami/etcd:latest'
    environment:
      - ALLOW_NONE_AUTHENTICATION=yes
      - ETCD_NAME=etcd1
      - ETCD_INITIAL_ADVERTISE_PEER_URLS=http://etcd1:2380
      - ETCD_LISTEN_PEER_URLS=http://0.0.0.0:2380
      - ETCD_LISTEN_CLIENT_URLS=http://0.0.0.0:2379
      - ETCD_ADVERTISE_CLIENT_URLS=http://etcd1:2379
      - ETCD_INITIAL_CLUSTER_TOKEN=etcd-cluster
      - ETCD_INITIAL_CLUSTER=etcd1=http://etcd1:2380,etcd2=http://etcd2:2380,etcd3=http://etcd3:2380
      - ETCD_INITIAL_CLUSTER_STATE=new
    networks:
      - app-tier
  etcd2:
    image: 'bitnami/etcd:latest'
    environment:
      - ALLOW_NONE_AUTHENTICATION=yes
      - ETCD_NAME=etcd2
      - ETCD_INITIAL_ADVERTISE_PEER_URLS=http://etcd2:2380
      - ETCD_LISTEN_PEER_URLS=http://0.0.0.0:2380
      - ETCD_LISTEN_CLIENT_URLS=http://0.0.0.0:2379
      - ETCD_ADVERTISE_CLIENT_URLS=http://etcd1:2379
      - ETCD_INITIAL_CLUSTER_TOKEN=etcd-cluster
      - ETCD_INITIAL_CLUSTER=etcd1=http://etcd1:2380,etcd2=http://etcd2:2380,etcd3=http://etcd3:2380
      - ETCD_INITIAL_CLUSTER_STATE=new
    networks:
      - app-tier
  etcd3:
    image: 'bitnami/etcd:latest'
    environment:
      - ALLOW_NONE_AUTHENTICATION=yes
      - ETCD_NAME=etcd3
      - ETCD_INITIAL_ADVERTISE_PEER_URLS=http://etcd3:2380
      - ETCD_LISTEN_PEER_URLS=http://0.0.0.0:2380
      - ETCD_LISTEN_CLIENT_URLS=http://0.0.0.0:2379
      - ETCD_ADVERTISE_CLIENT_URLS=http://etcd1:2379
      - ETCD_INITIAL_CLUSTER_TOKEN=etcd-cluster
      - ETCD_INITIAL_CLUSTER=etcd1=http://etcd1:2380,etcd2=http://etcd2:2380,etcd3=http://etcd3:2380
      - ETCD_INITIAL_CLUSTER_STATE=new
    networks:
      - app-tier
  etcdkeeper:
    image: deltaprojects/etcdkeeper:latest
    ports:
      - 8080:8080
    networks:
      - app-tier
```

etcdkeeper 连接etcd: `etcd1:2379` ，它们处在同一个network中直接使用服务名就可以

# 服务注册与服务发现

**服务注册：**

1. 申请lease租约，设置服务生存周期ttl，让key自动过期
2. 服务正常状态下，通过keep-alive定期去租约，避免过期
3. key有统一前缀，比如：`etcdctl --endpoints=127.0.0.1:2379 put /sdmaster-service/192.168.31.110   192.168.31.110`

**服务发现：**

1. 服务`watch /sdmaster-service --prefix`
2. `sdmaster-service` 下发生变化都会发生通知

相关代码已经发到github [discovery](https://github.com/douyacun/discovery)

**版本兼容:**

1. 正常安装`github.com/coreos/etcd/clientv3`会存在 `module declares its path as: go.etcd.io/bbolt`

```
go: discovery/discovery-service-protocol/client imports
        github.com/coreos/etcd/clientv3 tested by
        github.com/coreos/etcd/clientv3.test imports
        github.com/coreos/etcd/auth imports
        github.com/coreos/etcd/mvcc/backend imports
        github.com/coreos/bbolt: github.com/coreos/bbolt@v1.3.5: parsing go.mod:
        module declares its path as: go.etcd.io/bbolt
                but was required as: github.com/coreos/bbolt
```

解决方法：

`replace github.com/coreos/bbolt latest => go.etcd.io/bbolt latest`

2. 如果使用grpc的话，会存在grpc版本和etcd不兼容的问题

使用最先版本grpc, 会报`google.golang.org/grpc/naming`缺失

`module google.golang.org/grpc@latest found (v1.33.1), but does not contain package google.golang.org/grpc/naming`

如果降低到1.29.0可以解决这个问题，如果运行代码会报

`undefined: resolver.BuildOption` 

`undefined: resolver.ResolveNowOption`

这是因为grpc版本和etcd不兼容的问题，需要降低版本到 `1.26.0` 以下

解决方法:

```
replace google.golang.org/grpc latest => google.golang.org/grpc v1.26.0
```

3. 目前 protobuf grpc 生成的go代码, 会报错 `undefined grpc.ClientConnInterface` `undefined: grpc.SupportPackageIsVersion6`

```
go get -d -u github.com/golang/protobuf/protoc-gen-go
git -C "$(go env GOPATH)"/src/github.com/golang/protobuf checkout $GIT_TAG
go install github.com/golang/protobuf/protoc-gen-go
```

# go-grpc 负载均衡

正常http负载均衡策略，client -> proxy gateway load balance -> service

grpc提供负载均衡策略，client  load balance -> service 

grpc关键点是负载均衡在客户端

1. 服务端启动时，首先将服务地址注册到服务注册表，定期keepalive健康检测
2. 客户端访问某个服务，watch订阅服务注册表，然后以某种负载均衡策略选择一个目标地址

![](/images/blog/1/architecture/assert/grpc-load-balancing.png)

相关代码已发到github: [go-grpc 负载均衡](https://github.com/douyacun/discovery)

# etcd 配置中心

- 动态加载配置（watch）
- 历史版本（revision）
- 权限控制
- 命名空间

思路:

1. 加载配置文件 put etcd
2. watch prefix 配置变更可以及时变更配置

相关代码已经发到 github : [etcd配置中心](https://github.com/douyacun/etcd/)

# etcd 分布式锁

原理:

1. 租约创建一个key
2. key不存在，创建key，成功获取到锁
3. key存在，无法创建key，获取锁失败

etcd分布式锁的实现在`go.etcd.io/etcd/clientv3/concurrency`包中

1. 初始化sessions

```go
func NewSession(client *v3.Client, opts ...SessionOption) (*Session, error) {
	ops := &sessionOptions{ttl: defaultSessionTTL, ctx: client.Ctx()}
	for _, opt := range opts {
		opt(ops)
	}
	// 初始化租约
	id := ops.leaseID
	if id == v3.NoLease {
		resp, err := client.Grant(ops.ctx, int64(ops.ttl))
		if err != nil {
			return nil, err
		}
		id = v3.LeaseID(resp.ID)
	}
	// 由上层调用控制是否中断
	ctx, cancel := context.WithCancel(ops.ctx)
	keepAlive, err := client.KeepAlive(ctx, id)
	if err != nil || keepAlive == nil {
		cancel()
		return nil, err
	}
	donec := make(chan struct{})
	s := &Session{client: client, opts: ops, id: id, cancel: cancel, donec: donec}
	go func() {
		defer close(donec)
		for range keepAlive {
			// eat messages until keep alive channel closes
		}
	}()
	return s, nil
}
```

2. `func NewMutex(s *Session, pfx string) *Mutex `

初始化`Mutex`

3. `func (m *Mutex) Lock(ctx context.Context) error`

这里类似redis的分布式锁: `set key value NX EX 60`  如果key存在就返回，否则就设置

```go
func (m *Mutex) Lock(ctx context.Context) error {
	s := m.s
	client := m.s.Client()
  // 伪代码
  // if !m.myKey {
  //	client.Put(m.myKey, value)
  // } else {
  // 	client.Get(m.myKey)
	// }
	m.myKey = fmt.Sprintf("%s%x", m.pfx, s.Lease())
	cmp := v3.Compare(v3.CreateRevision(m.myKey), "=", 0)
	put := v3.OpPut(m.myKey, "", v3.WithLease(s.Lease()))
	get := v3.OpGet(m.myKey)
	getOwner := v3.OpGet(m.pfx, v3.WithFirstCreate()...)
	resp, err := client.Txn(ctx).If(cmp).Then(put, getOwner).Else(get, getOwner).Commit()
	if err != nil {
		return err
	}
 	// 此锁是否为自己获得
	m.myRev = resp.Header.Revision
	if !resp.Succeeded {
		m.myRev = resp.Responses[0].GetResponseRange().Kvs[0].CreateRevision
	}
	ownerKey := resp.Responses[1].GetResponseRange().Kvs
	if len(ownerKey) == 0 || ownerKey[0].CreateRevision == m.myRev {
		m.hdr = resp.Header
		return nil
	}
	// 阻塞直到获取到该锁
	hdr, werr := waitDeletes(ctx, client, m.pfx, m.myRev-1)
	if werr != nil {
		m.Unlock(client.Ctx())
	} else {
		m.hdr = hdr
	}
	return werr
}
```

4. `func (m *Mutex) Unlock(ctx context.Context) error` 删除 key

# 推荐阅读

- [高可用分布式存储 etcd 的实现原理](https://draveness.me/etcd-introduction/)
- [etcd使用经验总结](https://alexstocks.github.io/html/etcd.html)





已实现 grpc负载均衡、服务发现、分布式锁、配置中心、

etcd应用场景


Etcd几个关键特征：

**raft算法**：Leader选举/操作同步，无论访问任意节点，都将获得最终一致性数据，高度可靠。

**mvcc**:  etcd有2级存储结构 [B树内存索引](https://github.com/google/btree)  / [bbolt B+树](https://github.com/etcd-io/bbolt) ，内存中使用btree维护key/value索引，节点中存的是bbolt中键值K，通过这个k在bbolt查找，得到的才是用户传进去的value值

- 每个事务有唯一事务id，就是下面的main id int64
- 一个事务可以包含多个修改操作（put/delete），每一个操作就是一个revision, 共享同一个main id
- 一个事务内多个连续修改操作会从0开始递增，编号 sub id int64

```
type revision struct {
	// main is the main revision of a set of changes that happen atomically.
	main int64

	// sub is the sub revision of a change in a set of changes that happen
	// atomically. Each change has different increasing sub revision in that
	// set.
	sub int64
}

type keyIndex struct {
	key         []byte
	modified    revision // the main rev of the last modification
	generations []generation
}
```

内存索引中，每个原始key会关联一个key_index，里面维护了多版本信息

**事务**: 

- 只读事务
- 读写事务



# 概要

- etcd集群

- 配置UI管理界面

- 监控与维护

- 容灾

#### 应用场景

- 服务发现
- 配置中心
- 分布式锁
- 监控机器活动

#### 基础功能

- 监控 watch
- 租约 lease
- 全局锁 lock
- 选举 elect
- 集群状态 
  - `etcdctl --write-out=table endpoint status`
  - `etcdctl endpoint health`
- 快照 snapshot
  - `etcdctl --endpoints=127.0.0.1:2379 snapshot save my.db`
  - `etcdctl --endpoints=127.0.0.1:2379 --write-out=table snapshot status my.db`
- 迁移 migrate
- 成员管理 member
  - `etcdctl --endpoints=127.0.0.1:2379 member list`

- 授权 auth



# 集群配置

`docker-compose.yml`

```yaml
services:
  etcd1:
    image: 'bitnami/etcd:latest'
    environment:
      - ALLOW_NONE_AUTHENTICATION=yes
      - ETCD_NAME=etcd1
      - ETCD_INITIAL_ADVERTISE_PEER_URLS=http://etcd1:2380
      - ETCD_LISTEN_PEER_URLS=http://0.0.0.0:2380
      - ETCD_LISTEN_CLIENT_URLS=http://0.0.0.0:2379
      - ETCD_ADVERTISE_CLIENT_URLS=http://etcd1:2379
      - ETCD_INITIAL_CLUSTER_TOKEN=etcd-cluster
      - ETCD_INITIAL_CLUSTER=etcd1=http://etcd1:2380,etcd2=http://etcd2:2380,etcd3=http://etcd3:2380
      - ETCD_INITIAL_CLUSTER_STATE=new
    networks:
      - app-tier
  etcd2:
    image: 'bitnami/etcd:latest'
    environment:
      - ALLOW_NONE_AUTHENTICATION=yes
      - ETCD_NAME=etcd2
      - ETCD_INITIAL_ADVERTISE_PEER_URLS=http://etcd2:2380
      - ETCD_LISTEN_PEER_URLS=http://0.0.0.0:2380
      - ETCD_LISTEN_CLIENT_URLS=http://0.0.0.0:2379
      - ETCD_ADVERTISE_CLIENT_URLS=http://etcd1:2379
      - ETCD_INITIAL_CLUSTER_TOKEN=etcd-cluster
      - ETCD_INITIAL_CLUSTER=etcd1=http://etcd1:2380,etcd2=http://etcd2:2380,etcd3=http://etcd3:2380
      - ETCD_INITIAL_CLUSTER_STATE=new
    networks:
      - app-tier
  etcd3:
    image: 'bitnami/etcd:latest'
    environment:
      - ALLOW_NONE_AUTHENTICATION=yes
      - ETCD_NAME=etcd3
      - ETCD_INITIAL_ADVERTISE_PEER_URLS=http://etcd3:2380
      - ETCD_LISTEN_PEER_URLS=http://0.0.0.0:2380
      - ETCD_LISTEN_CLIENT_URLS=http://0.0.0.0:2379
      - ETCD_ADVERTISE_CLIENT_URLS=http://etcd1:2379
      - ETCD_INITIAL_CLUSTER_TOKEN=etcd-cluster
      - ETCD_INITIAL_CLUSTER=etcd1=http://etcd1:2380,etcd2=http://etcd2:2380,etcd3=http://etcd3:2380
      - ETCD_INITIAL_CLUSTER_STATE=new
    networks:
      - app-tier
  etcdkeeper:
    image: deltaprojects/etcdkeeper:latest
    ports:
      - 8080:8080
    networks:
      - app-tier
```

etcdkeeper 连接etcd: `etcd1:2379` ，它们处在同一个network中直接使用服务名就可以

# 服务注册与服务发现

**服务注册：**

1. 申请lease租约，设置服务生存周期ttl，让key自动过期
2. 服务正常状态下，通过keep-alive定期去租约，避免过期
3. key有统一前缀，比如：`etcdctl --endpoints=127.0.0.1:2379 put /sdmaster-service/192.168.31.110   192.168.31.110`

**服务发现：**

1. 服务`watch /sdmaster-service --prefix`
2. `sdmaster-service` 下发生变化都会发生通知

相关代码已经发到github [discovery](https://github.com/douyacun/discovery)

**版本兼容:**

1. 正常安装`github.com/coreos/etcd/clientv3`会存在 `module declares its path as: go.etcd.io/bbolt`

```
go: discovery/discovery-service-protocol/client imports
        github.com/coreos/etcd/clientv3 tested by
        github.com/coreos/etcd/clientv3.test imports
        github.com/coreos/etcd/auth imports
        github.com/coreos/etcd/mvcc/backend imports
        github.com/coreos/bbolt: github.com/coreos/bbolt@v1.3.5: parsing go.mod:
        module declares its path as: go.etcd.io/bbolt
                but was required as: github.com/coreos/bbolt
```

解决方法：

`replace github.com/coreos/bbolt latest => go.etcd.io/bbolt latest`

2. 如果使用grpc的话，会存在grpc版本和etcd不兼容的问题

使用最先版本grpc, 会报`google.golang.org/grpc/naming`缺失

`module google.golang.org/grpc@latest found (v1.33.1), but does not contain package google.golang.org/grpc/naming`

如果降低到1.29.0可以解决这个问题，如果运行代码会报

`undefined: resolver.BuildOption` 

`undefined: resolver.ResolveNowOption`

这是因为grpc版本和etcd不兼容的问题，需要降低版本到 `1.26.0` 以下

解决方法:

```
replace google.golang.org/grpc latest => google.golang.org/grpc v1.26.0
```

3. 目前 protobuf grpc 生成的go代码, 会报错 `undefined grpc.ClientConnInterface` `undefined: grpc.SupportPackageIsVersion6`

```
go get -d -u github.com/golang/protobuf/protoc-gen-go
git -C "$(go env GOPATH)"/src/github.com/golang/protobuf checkout $GIT_TAG
go install github.com/golang/protobuf/protoc-gen-go
```

# go-grpc 负载均衡

正常http负载均衡策略，client -> proxy gateway load balance -> service

grpc提供负载均衡策略，client  load balance -> service 

grpc关键点是负载均衡在客户端

1. 服务端启动时，首先将服务地址注册到服务注册表，定期keepalive健康检测
2. 客户端访问某个服务，watch订阅服务注册表，然后以某种负载均衡策略选择一个目标地址

![](/images/blog/architecture/assert/grpc-load-balancing.png)

相关代码已发到github: [go-grpc 负载均衡](https://github.com/douyacun/discovery)

# etcd 配置中心

- 动态加载配置（watch）
- 历史版本（revision）
- 权限控制
- 命名空间

思路:

1. 加载配置文件 put etcd
2. watch prefix 配置变更可以及时变更配置

相关代码已经发到 github : [etcd配置中心](https://github.com/douyacun/etcd/)

# etcd 分布式锁

原理:

1. 租约创建一个key
2. key不存在，创建key，成功获取到锁
3. key存在，无法创建key，获取锁失败

etcd分布式锁的实现在`go.etcd.io/etcd/clientv3/concurrency`包中

1. 初始化sessions

```go
func NewSession(client *v3.Client, opts ...SessionOption) (*Session, error) {
	ops := &sessionOptions{ttl: defaultSessionTTL, ctx: client.Ctx()}
	for _, opt := range opts {
		opt(ops)
	}
	// 初始化租约
	id := ops.leaseID
	if id == v3.NoLease {
		resp, err := client.Grant(ops.ctx, int64(ops.ttl))
		if err != nil {
			return nil, err
		}
		id = v3.LeaseID(resp.ID)
	}
	// 由上层调用控制是否中断
	ctx, cancel := context.WithCancel(ops.ctx)
	keepAlive, err := client.KeepAlive(ctx, id)
	if err != nil || keepAlive == nil {
		cancel()
		return nil, err
	}
	donec := make(chan struct{})
	s := &Session{client: client, opts: ops, id: id, cancel: cancel, donec: donec}
	go func() {
		defer close(donec)
		for range keepAlive {
			// eat messages until keep alive channel closes
		}
	}()
	return s, nil
}
```

2. `func NewMutex(s *Session, pfx string) *Mutex `

初始化`Mutex`

3. `func (m *Mutex) Lock(ctx context.Context) error`

这里类似redis的分布式锁: `set key value NX EX 60`  如果key存在就返回，否则就设置

```go
func (m *Mutex) Lock(ctx context.Context) error {
	s := m.s
	client := m.s.Client()
  // 伪代码
  // if !m.myKey {
  //	client.Put(m.myKey, value)
  // } else {
  // 	client.Get(m.myKey)
	// }
	m.myKey = fmt.Sprintf("%s%x", m.pfx, s.Lease())
	cmp := v3.Compare(v3.CreateRevision(m.myKey), "=", 0)
	put := v3.OpPut(m.myKey, "", v3.WithLease(s.Lease()))
	get := v3.OpGet(m.myKey)
	getOwner := v3.OpGet(m.pfx, v3.WithFirstCreate()...)
	resp, err := client.Txn(ctx).If(cmp).Then(put, getOwner).Else(get, getOwner).Commit()
	if err != nil {
		return err
	}
 	// 此锁是否为自己获得
	m.myRev = resp.Header.Revision
	if !resp.Succeeded {
		m.myRev = resp.Responses[0].GetResponseRange().Kvs[0].CreateRevision
	}
	ownerKey := resp.Responses[1].GetResponseRange().Kvs
	if len(ownerKey) == 0 || ownerKey[0].CreateRevision == m.myRev {
		m.hdr = resp.Header
		return nil
	}
	// 阻塞直到获取到该锁
	hdr, werr := waitDeletes(ctx, client, m.pfx, m.myRev-1)
	if werr != nil {
		m.Unlock(client.Ctx())
	} else {
		m.hdr = hdr
	}
	return werr
}
```

4. `func (m *Mutex) Unlock(ctx context.Context) error` 删除 key

# 推荐阅读

- [高可用分布式存储 etcd 的实现原理](https://draveness.me/etcd-introduction/)
- [etcd使用经验总结](https://alexstocks.github.io/html/etcd.html)



