# 我的2024年度总结

&ensp;&ensp;&ensp;&ensp;2024好像过的很快，却又好像过的很慢，懵懵懂懂间，一年又结束了。

&ensp;&ensp;&ensp;&ensp;回顾2024年，我的全年关键词是：祛魅。若相比2023年的“变化”，2024年更多的，我见识了形形色色的人与事，感悟到很多东西非我所想，也非我所能，坦然接受，踏步向前，才是人生的基调。


## 为什么是祛魅？


&ensp;&ensp;&ensp;&ensp;&ensp;祛魅，应该算是今年来网络上比较火热的词，对世界祛魅，对职场祛魅。对人生祛魅。意为慢慢消除对一切事物的滤镜，消解美好期待和幻想光环，不去神化、美化一些人或事物，将关注点回到自身。

&ensp;&ensp;&ensp;&ensp;&ensp;回顾我的这一年，经历了很多事情，也见到了很多形形色色的人，发现很多人或事，并没有我想象的的那么美好，也并没有我想象的那么糟糕。

&ensp;&ensp;&ensp;&ensp;&ensp;工作上祛魅，曾经我觉得大厂工作是一件非常自豪的事情，向往大厂的技术氛围，向往大厂的顶级工作环境与先进业务，幻想大厂唯业务技术说话的工作环境。但经历了一年的大厂工作，发现其实大家也都是草台班子，同样会出现的边缘业务，同样会出现的低级故障，同样会出现的人情世故，同样会出现的工作失误，同样会出现的形势误判，同样会出现的吹牛浮夸，浑水摸鱼，终归大家都没有什么不同罢了，终是**优秀的人在大厂施展拳脚，而不是大厂促成了优秀的人**。
> 以自我为中心不断进步的人，才是真正的优秀。

&ensp;&ensp;&ensp;&ensp;&ensp;生活上祛魅，过去一年，曾经那个跟着大人上宴席蹭吃蹭喝的我，也到了要作为一个独立的人，出席这些事情的年纪。身边的老人逐渐离去，新人逐渐成婚，逐渐像当年的父母那样，帮衬朋友、上礼的人。小时候跟着大人，只盼着宴席快快上菜，但当轮到自己出席的时候，才深感面对逝者的悲伤，面对新人婚姻的庆贺，面对人生进程加速的无奈。宴席这个在曾经稚嫩的我看来一个可以跟着大人大吃大喝的好事，却变得令我感情复杂。时常会想，如果当轮到我作为宴席的主角，我又能否克制感情，主持好大局？

&ensp;&ensp;&ensp;&ensp;&ensp;感情上祛魅，从小受父母影响，对一份感情，特别是恋爱的感情，非常看重，幻想着，只得一人心，白首不相离，希望有像父母那样纯粹、一眼便是终生的浪漫感情，对爱情这种事情无限幻想着美好。但是经历了身边其他朋友分分合合，见到了各种各样分崩离析的感情，我顿感无力。直到有天无意间看到了付航的脱口秀，“**你是你自己剧本的主角**”，豁然开朗，再美好的感情，再美好的人，也不能成为某个人生命的主角，我永远作为一个自由的个体，作为自己的主角，任何人的离开，只是结束了这一段剧本，还会有下一段新的剧本等待我的演绎。人生就是一趟永无尽头的旅行，我的列车上会搭载形形色色的人，并与我有一段或深或浅的故事，但终归他们都有到站下车的一天，而我的列车永无止境。所以无论是亲人的离去、挚爱的离去、朋友的离去，无需怀疑自己，也无需遗憾，行好事，问前程，有更加精彩的下一段旅程在等着自己。
> “离开的人不要挽留，到来的人不要轻薄”。

&ensp;&ensp;&ensp;&ensp;&ensp;人际交往上祛魅。以前认为高学历、高地位的人总是各方面都远优秀于自己，而没读过什么书，也没有家庭背景的人好像样样不如自己。经常在和高学历背景的朋友交往中有些自卑，但和老家提早进入社会的朋友相比又觉得自己混的还行，有些自负。但随着愈发了解到真实的他们，发现，大家也就都那样。高学历高背景的同样会有抽象的人，离谱的性格，摆烂挂科的废人；提早步入社会，地位较低的劳苦大众也有工作出色，年纪轻轻便在自己的行业发光发热，月入过万，小有所成，甚至有些朋友的现状超过了绝大多数同龄人。终归大家也只是芸芸众生的普通人，只是学历不同、家庭背景不同罢了。在某些特定方面，我自己做的远比高学历高地位的优秀，又在某些地方远不及没读过几年书的人看的透彻，做的更好。人皆平等，没有人是神话，也没有人是完全烂透的人渣。
> “大家也就都那样”

&ensp;&ensp;&ensp;&ensp;&ensp;自我祛魅，作为标准的小镇做题家出身，不给别人添麻烦，要多考虑别人，多吃苦才能有出息。类似的语言总在耳边提起，仿佛生来是为了这个世界、为了社会、为了国家、为了家庭、为了他人而活。好像所有的老中普通人一辈子都这样，为了父母而学习，为了公司而工作，为了孩子割舍自己所爱。这样的自我我并不认可，所以今年我随着自己任性了，去了沪苏杭、去了很多次漫展、去了洛天依的演唱会、去了初音未来的演唱会、在有限的时间里做了我想做的事情，我认为这些是真正有意义的事情。
> “想做就做，死了就做不成了”



&ensp;&ensp;&ensp;&ensp;&ensp;也在这些事情之后，对很多事情没有了太多的执着，对很多人或事没有了那么深的滤镜，随心而为，很多时候只有自己才是自己的主角，其他人或事，只是自己人生剧本中的一小段插曲。

&ensp;&ensp;&ensp;&ensp;&ensp;因此，我选择了祛魅，作为我这一年的总结。

## 生活的总结
[2024-时间线](https://umb.ink/post/33)

## 工作的总结

今年工作是比较繁杂的，整体压力也相对较大，但总归看着工作上线，也是有所成就感。

因为事情太杂太琐碎，索性直接总结今年在工作中学到的东西与成果好了。

#### 结果

1. 去年做的班车项目，今年加入了基于GPS的自动切换路线迭代了一版本，并且！总乘车人次破一百万啦，看数据的时候感觉非常满足。
2. 今年新做了一个网络可视化的项目，第一次接触到了海量的数据计算，首次学习并使用了Clickhouse进行数据聚合和处理，第一次写程序处理每天千万的数据量，感觉很棒。
3. 相比去年，感觉今年的工作主导性更少，支持性更多，大部分时候都是配合大项目推进，负责其中的一些模块开发，但看着项目如期能够上线，依然有充足感。
#### 感受

累，今年真实感受到了大厂对牛马的压榨与内卷的环境，工作中要处处留心，随时留痕，精力一半在工作上另一半在操心和各种任何事情的处理上，身心俱疲。
## 来年的愿望

1. 来年希望能去一次日本，去一次洱海。
2. 希望能对工作更得心应手。
3. 希望生活更规律一些，能有精力去运动。
4. 遇到一些更美好的人与事情。
5. 健康的活着。

2024年度总结

# 攻略说明
本攻略侧重云南自然风光游玩，轻人文景观（古城等），因此仅供想要侧重游玩自然风光的进行参考



# 涉及地点
云南北线攻略：昆明-大理-丽江，以大理与丽江为主，昆明走马观花

因为六天五夜时间比较紧张，其实完整北线应该是 昆明-大理-丽江-香格里拉

核心地点：大理洱海、大理古城、丽江古城、玉龙雪山风景区、昆明滇池

预算：旺季人均2k（不算机票与自购特产）
# 时间安排：

1. 第一天 落地昆明，然后前往大理，先游玩大理古城（第一天下午+晚上），晚上可以在大理古城附近吃傣味手抓饭体验，逛到晚上10点可以打车前往酒店入住。

2. 第二天 在龙龛码头看日出，然后骑电动车环游洱海西岸，裳苍山雪。

-  码头日出：

![IMG_9592.JPG](/static/img/ae3c4a6fe76af8ccb250c7392a024cdf.IMG_9592.webp)

- 苍山雪：

![KIM20250212-025419.png](/static/img/2464effac91f8204bfeb9d6f0cf5e628.KIM20250212-025419.webp)


- 洱海西岸：

![IMG_2EC6FEC603DB-1.jpeg](/static/img/941f84f8eb193e22a4813c648d709f69.IMG_2EC6FEC603DB-1.webp)



3. 第三天 包车或自驾游洱海东岸，游双廊古镇，看鹿卧山，小普陀。下午在东岸看洱海日落，晚上前往丽江。

- 双廊古镇：



![image.png](/static/img/05ce4c7c14910eed6ed3d21e33337e56.image.webp)



- 鹿卧山：

![image.png](/static/img/0ec02e8cd450e43df6e30cdc7633d25d.image.webp)

4. 第四天 在丽江古城早起看古城的日照金山（需要碰运气），全天游古城，晚上可以去古城酒吧一条街体验丽江古城的民谣酒吧。


![image.png](/static/img/55432eebae820cca7c0ce604fadee980.image.webp)


![image.png](/static/img/d131262902b6522fb19486cc105ddc57.image.webp)

5. 第五天 包车或自驾前往玉龙雪山风景区，凌晨出发，早上可以到东巴谷看日照金山，然后进入山区上大索道（如果抢不到票去中索道牦牛坪也很不错），下山后坐景区大巴前往蓝月谷。

玉龙雪山脚下：

![image.png](/static/img/13ff4a3c19689f060b9a7c2d15dd4479.image.webp)


牦牛坪：

![image.png](/static/img/f4e2a0b3b5be95236772f4e7f6d4e1cc.image.webp)

6. 第六天 早起返回昆明，时间充裕可以去海洪湿地公园看滇池海鸥，然后去翠湖公园看郁金香，如果还有时间推荐斗南花卉市场。

# 路线安排：

注意：建议不要携带行李箱，因为路途遥远且各个景点之间相距甚远，行李箱会非常不方便，最好做背包客，一个背包解决所有问题。

- 第一天：
   
   早上飞机落地昆明，走马上任前往大理。昆明-大理 动车约一个半小时，约下午三点到达大理。
    
   大理站到大理古城可以坐大理旅游观光线路大巴，途中会经过兴盛大桥，到达兴盛大桥可以下车打卡，下午刚好是打卡的好时候。
   
![6d7808104393cc84210fddd90be7c1ae.png](/static/img/11afda3d06a3ffac389d9fc0aeb95dce.6d7808104393cc84210fddd90be7c1ae.webp)
    
   下午和晚上在大理古城游玩，玩够了可以骑共享电动车前往酒店入住（注意千万不要租古城附近的电动车，否则你还需要回来还车）。
  
- 第二天：

早起去看龙龛码头的日出（注意可能人会很多，建议提早去占位置），龙龛古渡的观景台上有“上关花、下关风、苍山雪、洱海月”四个石碑，可以打卡。

日出回来可以吃个早餐，洱海附近很多卖烤饵块早餐的。然后可以租个车在洱海西岸游玩。

- 第三天
 
洱海东岸环线

旺季建议在东岸包车（注意，不要包本地人车行的车，非常贵而且坑），建议打一个滴滴，然后咨询师傅能不能包他的车，直接找网约车师傅包车会更便宜（没有中间商赚差价）而且很多都是本地人，比导游更了解哪里好玩。

淡季建议租车（不堵车的洱海东岸值得自驾）

- 第四天

从大理到丽江，可以在大理选择直达丽江古城的大巴车（70每人）经济实惠，也可以坐高铁直达。

到达丽江后可以打车到丽江古城（如果是大巴会直接送到古城门口）


- 第五天

玉龙雪山（记住提前一周抢票！！！）玉龙雪山分为大索道-上雪山、中索道-上牦牛坪、小索道-去云杉坪

去之前建议在古城购买一瓶氧气（十块钱）千万不要到了景区再买，景区都是几百一瓶。

尽量不要在古城购买一日游套餐！体验非常差。导游会忽悠你去买高价氧气，租用很烂的那种保温服。

在雪山如果准备去蓝月谷的话，不要坐观光车，直接乘坐景区通勤大巴到蓝月谷的下一站，可以直接徒步，体验更好。

- 第六天 


从丽江坐高铁返回昆明返程，路上会经过洱海，可以留意下火车窗外的风光。

大理-丽江六天五夜旅游攻略

Android-常用ADB命令与工具类

整理一下之前工作中用到的GRPC相关内容




# grpc定义

grpc是一种接口描述协议，我们可以通过定义proto文件来自动实现相关代码逻辑，举个例子：

```bash
syntax = "proto3";

// 标记所属包名
package grpc_test;
// 标记生成的go代码目录：即生成代码后会把代码文件丢到当前目录下的grpc_test文件夹中（如果是java-grpc，则是java_package)
option go_package = "./grpc_test";

// 可以把这个messsage理解为结构体定义，这个message映射为代码类或实体结构体，这里的=1和=2是在标记这个字段，从1开始，第几个字段标记为几
message GrpcHelloWorldReq {
    string message = 1;
    // 可以定义string list
    repeated string messageList = 2;
}

message GrpcHelloWorldResp {
    string message = 1;
    // 可以定义结构体list
    repeated GrpcHelloWorldReq req = 2;
}

service MeetingDeviceService {
    rpc HelloWorld(GrpcHelloWorldReq) returns (GrpcHelloWorldResp){}
    // 定义一个rpc接口，输入GrpcHelloWorldReq结构体，返回GrpcHelloWorldResp
}

```

# 代码生成
在proto文件目录下运行下面的指令
```bash
protoc --go_out=. *.proto
protoc --go-grpc_out=. *.proto    
```

成功运行后你就会在目录/grpc_test/下看到.pb.go的文件，这个里面包含了根据定义的proto文件生成的代码

# 接口实现

服务端：
```go
package main

import (
	"context"
	"fmt"
	"net"

	"kafka-test/main/grpc_test"

	"google.golang.org/grpc"
)

type server struct {
	grpc_test.MeetingDeviceServiceServer
}

func (s *server) HelloWorld(ctx context.Context, req *grpc_test.GrpcHelloWorldReq) (*grpc_test.GrpcHelloWorldResp, error) {
	fmt.Println("request:", req.Message)
	return &grpc_test.GrpcHelloWorldResp{Message: "hello " + req.Message}, nil
}

func main() {
	listen, err := net.Listen("tcp", ":8001")
	if err != nil {
		fmt.Printf("failed to listen: %v", err)
		return
	}
	s := grpc.NewServer()
	//reflection.Register(s)

	defer func() {
		s.Stop()
		listen.Close()
	}()

	fmt.Println("Serving 8001...")
	err = s.Serve(listen)
	if err != nil {
		fmt.Printf("failed to serve: %v", err)
		return
	}
}
```
客户端：
```go
package main

import (
	"context"
	"fmt"
	"sync"

	"kafka-test/main/grpc_test"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
)

var (
	meetingRoomRpcClient grpc_test.MeetingDeviceServiceClient
	once                 = sync.Once{}
)

func init() {
	var serviceHost = "127.0.0.1:8001"

	conn, err := grpc.Dial(serviceHost, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		fmt.Println(err)
	}
	defer conn.Close()
	// 初始化client（这里client作为调用方，所以只需要初始化一次）
	once.Do(func() {
		meetingRoomRpcClient = grpc_test.NewMeetingDeviceServiceClient(conn)
		//meetingRoomRpcClient = grpc_test.NewMeetingDeviceServiceClient("meeting-device-service")
	})

}

func RpcHellWorld() {
	worldResp, err := meetingRoomRpcClient.HelloWorld(context.Background(), &grpc_test.GrpcHelloWorldReq{
		Message:     "hello world",
		MessageList: []string{"hello world", "hello world"},
	})
	if err != nil {
		return
	}
	fmt.Println(worldResp.Message)
}
```

启动服务后，rpc服务会运行在本地的8001端口，因此作为下游客户端的rpc服务只需要监听这个8001端口即可实现rpc调用。当然，下游的rpc客户端服务也可以定义自己的proto rpc文件，然后被其他服务调用自己的rpc方法。

在rpc服务调用过程中，谁提供方法接口，谁是server，谁调用方法接口，谁是client，只是一个相对概念。


# 进阶——服务发现ETCD

当RPC服务多了以后，我们需要配置每个RPC服务的IP和端口，非常麻烦，因此这就需要用到注册中心来简化RPC服务管理。用注册中心的话，我们只需要记住每个服务的服务名即可，就不用再挨个记IP了。

放一个官方文档：https://etcd.io/docs/v3.5/quickstart/
安装请自行参考官方文档

ETCD注册服务端：

```golang

import (
	"context"
	"flag"
	"fmt"
	"net"
	"time"

	grpc_test2 "kafka-test/rpc/grpc_test"

	// etcd
	eclient "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/client/v3/naming/endpoints"
	"google.golang.org/grpc"
)

const (
	// grpc 服务名
	GrpcTestServiceName = "umb/grpc-test"
	// etcd 端口
	ETCD_ADDR = "http://localhost:2379"
)

type server struct {
	grpc_test2.UnimplementedGrpcHelloWorldServiceServer
}

func (s *server) HelloWorld(ctx context.Context, req *grpc_test2.GrpcHelloWorldReq) (*grpc_test2.GrpcHelloWorldResp, error) {
	fmt.Println("request:", req.Message)
	return &grpc_test2.GrpcHelloWorldResp{Message: "hello " + req.Message}, nil
}

func registerEndPointToEtcd(ctx context.Context, addr string) {
	// 创建 etcd 客户端
	etcdClient, _ := eclient.NewFromURL(ETCD_ADDR)
	etcdManager, _ := endpoints.NewManager(etcdClient, GrpcTestServiceName)

	// 创建一个租约，每隔 10s 需要向 etcd 汇报一次心跳，证明当前节点仍然存活
	var ttl int64 = 10
	lease, _ := etcdClient.Grant(ctx, ttl)

	// 添加注册节点到 etcd 中，并且携带上租约 id
	_ = etcdManager.AddEndpoint(ctx, fmt.Sprintf("%s/%s", GrpcTestServiceName, addr), endpoints.Endpoint{Addr: addr}, eclient.WithLease(lease.ID))

	// 每隔 5 s进行一次延续租约的动作
	for {
		select {
		case <-time.After(5 * time.Second):
			// 续约操作
			resp, _ := etcdClient.KeepAliveOnce(ctx, lease.ID)
			fmt.Printf("keep alive resp: %+v", resp)
		case <-ctx.Done():
			return
		}
	}
}

func InitRpcServer() {
	var port int
	flag.IntVar(&port, "port", 8001, "port")
	flag.Parse()
	addr := fmt.Sprintf("http://localhost:%d", port)
	listen, err := net.Listen("tcp", addr)
	if err != nil {
		fmt.Printf("failed to listen: %v", err)
		return
	}
	s := grpc.NewServer()
	//reflection.Register(s)
	grpc_test2.RegisterGrpcHelloWorldServiceServer(s, &server{})
	defer func() {
		s.Stop()
		listen.Close()
	}()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	// 注册 grpc 服务节点到 etcd 中
	go registerEndPointToEtcd(ctx, addr)
	fmt.Println("Serving 8001...")
	err = s.Serve(listen)
	if err != nil {
		fmt.Printf("failed to serve: %v", err)
		return
	}
}
```


ETCD注册客户端：

```golang
import (
	"context"
	"fmt"
	"sync"

	grpc_test2 "kafka-test/rpc/grpc_test"

	// etcd
	eclient "go.etcd.io/etcd/client/v3"
	eresolver "go.etcd.io/etcd/client/v3/naming/resolver"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
)

var (
	meetingRoomRpcClient grpc_test2.GrpcHelloWorldServiceClient
	once                 = sync.Once{}
)

func init() {
	// 创建 etcd 客户端
	etcdClient, _ := eclient.NewFromURL(ETCD_ADDR)

	// 创建 etcd 实现的 grpc 服务注册发现模块 resolver
	etcdResolverBuilder, _ := eresolver.NewBuilder(etcdClient)

	// 拼接服务名称，需要固定义 etcd:/// 作为前缀
	etcdTarget := fmt.Sprintf("etcd:///%s", GrpcTestServiceName)

	// 创建 grpc 连接代理
	conn, _ := grpc.Dial(
		// 服务名称
		etcdTarget,
		// 注入 etcd resolver
		grpc.WithResolvers(etcdResolverBuilder),
		// 声明使用的负载均衡策略为 roundrobin     grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, roundrobin.Name)),
		grpc.WithTransportCredentials(insecure.NewCredentials()),
	)
	defer conn.Close()
	defer conn.Close()
	// 初始化client（这里client作为调用方，所以只需要初始化一次）
	once.Do(func() {
		meetingRoomRpcClient = grpc_test2.NewGrpcHelloWorldServiceClient(conn)
		//meetingRoomRpcClient = grpc_test.NewMeetingDeviceServiceClient("meeting-device-service")
	})

}

func RpcHellWorld() {
	worldResp, err := meetingRoomRpcClient.HelloWorld(context.Background(), &grpc_test2.GrpcHelloWorldReq{
		Message:     "hello world",
		MessageList: []string{"hello world", "hello world"},
	})
	if err != nil {
		return
	}
	fmt.Println(worldResp.Message)
}
```

这样基于ETCD改造之后，server和客户端只需要知道对方的名称就能用了

Go-快速使用Grpc

# 前置准备
安装kafka-sarama库

直接附带我的go mod：

```go
require (
	github.com/IBM/sarama v1.45.0
	github.com/xdg/scram v1.0.5
	go.uber.org/zap v1.27.0
)

require (
	github.com/davecgh/go-spew v1.1.1 // indirect
	github.com/eapache/go-resiliency v1.7.0 // indirect
	github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 // indirect
	github.com/eapache/queue v1.1.0 // indirect
	github.com/golang/snappy v0.0.4 // indirect
	github.com/hashicorp/errwrap v1.1.0 // indirect
	github.com/hashicorp/go-multierror v1.1.1 // indirect
	github.com/hashicorp/go-uuid v1.0.3 // indirect
	github.com/jcmturner/aescts/v2 v2.0.0 // indirect
	github.com/jcmturner/dnsutils/v2 v2.0.0 // indirect
	github.com/jcmturner/gofork v1.7.6 // indirect
	github.com/jcmturner/gokrb5/v8 v8.4.4 // indirect
	github.com/jcmturner/rpc/v2 v2.0.3 // indirect
	github.com/klauspost/compress v1.17.11 // indirect
	github.com/pierrec/lz4/v4 v4.1.22 // indirect
	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
	github.com/xdg/stringprep v1.0.3 // indirect
	go.uber.org/multierr v1.10.0 // indirect
	golang.org/x/crypto v0.32.0 // indirect
	golang.org/x/net v0.34.0 // indirect
	golang.org/x/text v0.21.0 // indirect
)

```

# XDGSCRAM 解密

```go

import (
	"crypto/sha256"
	"crypto/sha512"

	"github.com/xdg/scram"
)

var (
	// SHA256 ...
	SHA256 scram.HashGeneratorFcn = sha256.New
	// SHA512 ...
	SHA512 scram.HashGeneratorFcn = sha512.New
)

// XDGSCRAMClient ...
type XDGSCRAMClient struct {
	*scram.Client
	*scram.ClientConversation
	scram.HashGeneratorFcn
}

// Begin ...
func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error) {
	x.Client, err = x.HashGeneratorFcn.NewClient(userName, password, authzID)
	if err != nil {
		return err
	}
	x.ClientConversation = x.Client.NewConversation()
	return nil
}

// Step ...
func (x *XDGSCRAMClient) Step(challenge string) (response string, err error) {
	response, err = x.ClientConversation.Step(challenge)
	return
}

// Done ...
func (x *XDGSCRAMClient) Done() bool {
	return x.ClientConversation.Done()
}

```

# 消费者接入：
```go
package main

import (
	"context"
	"log"
	"os"
	"os/signal"
	"sync"
	"syscall"

	"github.com/IBM/sarama"
)

type Kafka struct {
	brokers           []string
	topics            []string
	startOffset       int64
	version           string
	ready             chan bool
	group             string
	channelBufferSize int
	assignor          string
}

var assignor = "range"
var topics = []string{"your_topic"} // topic
var group = "consumer_group" // 订阅消费组
var user = "user" // sarma用户
var password = "pwd"  // sarama 密码
var brokers = []string{"127.0.0.1:6279"} // 你的kafka host

func NewKafka() *Kafka {
	return &Kafka{
		brokers:           brokers,
		topics:            topics,
		group:             group,
		channelBufferSize: 1000,
		ready:             make(chan bool),
		assignor:          assignor,
	}
}

func (k *Kafka) Connect() func() {
	log.Println("kafka init...")
	config := sarama.NewConfig()
	config.Version = sarama.V2_4_1_0 // 版本号必须注意要同步，否则会报错
	// 分区分配策略
	switch assignor {
	case "sticky":
		config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategySticky
	case "roundrobin":
		config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin
	case "range":
		config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRange
	default:
		log.Panicf("Unrecognized consumer group partition assignor: %s", assignor)
	}
	config.Consumer.Offsets.Initial = sarama.OffsetNewest
	config.ChannelBufferSize = k.channelBufferSize // channel长度
	config.Consumer.Offsets.Initial = sarama.OffsetOldest
	config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin
	config.Net.SASL.Enable = true
	config.Net.SASL.User = user
	config.Net.SASL.Password = password
	config.Net.SASL.Handshake = true
	config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient {
		return &XDGSCRAMClient{HashGeneratorFcn: SHA256}
	}
	config.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA256
	// 创建client
	log.Println("brokers: ", brokers)
	newClient, err := sarama.NewClient(brokers, config)
	if err != nil {
		log.Fatalf("Error creating client: %v", err)
	}
	// 获取所有的topic
	topics, err := newClient.Topics()
	if err != nil {
		log.Fatal(err)
	}
	log.Println("topics: ", topics)

	// 根据client创建consumerGroup
	client, err := sarama.NewConsumerGroupFromClient(k.group, newClient)
	if err != nil {
		log.Fatalf("Error creating consumer group client: %v", err)
	}

	ctx, cancel := context.WithCancel(context.Background())
	wg := &sync.WaitGroup{}
	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			if err := client.Consume(ctx, k.topics, k); err != nil {
				// 当setup失败的时候，error会返回到这里
				log.Println("Error from consumer: %v", err)
				return
			}
			// check if context was cancelled, signaling that the consumer should stop
			if ctx.Err() != nil {
				log.Println(ctx.Err())
				return
			}
			k.ready = make(chan bool)
		}
	}()
	<-k.ready
	log.Println("Sarama consumer up and running!...")
	// 保证在系统退出时，通道里面的消息被消费
	return func() {
		log.Println("kafka close")
		cancel()
		wg.Wait()
		if err = client.Close(); err != nil {
			log.Println("Error closing client: %v", err)
		}
	}
}

// Setup is run at the beginning of a new session, before ConsumeClaim
func (k *Kafka) Setup(session sarama.ConsumerGroupSession) error {
	log.Println("setup")
	session.ResetOffset("lvmi", 0, 13, "")
	log.Println(session.Claims())
	// Mark the consumer as ready
	close(k.ready)
	return nil
}

// Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited
func (k *Kafka) Cleanup(sarama.ConsumerGroupSession) error {
	log.Println("cleanup")
	return nil
}

// ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().
func (k *Kafka) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {

	// NOTE:
	// Do not move the code below to a goroutine.
	// The `ConsumeClaim` itself is called within a goroutine, see:
	// https://github.com/Shopify/sarama/blob/master/consumer_group.go#L27-L29
	// 具体消费消息
	for message := range claim.Messages() {
		log.Printf("[topic:%s] [partiton:%d] [offset:%d] [value:%s] [time:%v]",
			message.Topic, message.Partition, message.Offset, string(message.Value), message.Timestamp)
		// 更新位移
		session.MarkMessage(message, "")
	}
	return nil
}

func main() {
	k := NewKafka()
	c := k.Connect()

	sigterm := make(chan os.Signal, 1)
	signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)
	select {
	case <-sigterm:
		log.Println("terminating: via signal")
	}
	c()
}
```