第14章：消息队列实战与日志监控系统从零搭建
凌晨三点，你被电话炸醒。线上服务挂了，但你毫不知情——因为日志散落在十几台机器上，没有告警，没有大盘，出问题了全靠用户反馈。你SSH到每台机器grep日志，手忙脚乱地排查，等找到根因，已经损失了六位数的订单。
更常见的情况是：你用了消息队列，但消息偶尔丢失，偶尔重复，偶尔积压到几十万条。你查了三天三夜，发现是某个配置参数没调对。这些坑，每一个都是真金白银的教训。
我是怕浪猫，一个在消息队列和可观测性领域踩过无数坑的Go后端开发者。今天这一章，我们把消息队列和日志监控这两个运维命脉系统，从设计到实现完整撸一遍。不是Hello World级别的玩具，是能扛住生产流量的实战方案。篇幅会比较长，因为每一个知识点我都会给出完整的代码和踩坑分析，建议先收藏再慢慢看。
没有监控的系统就像闭眼开车，不出事是运气，出事是必然。而用了消息队列却不懂调优，就像在高速公路上开一辆随时可能熄火的车。
一、Kafka客户端实践：Sarama库使用详解
1.1 为什么选Sarama
Go生态里操作Kafka主流有两个库：confluent-kafka-go（基于librdkafka的CGO绑定）和IBM/sarama（纯Go实现）。怕浪猫在生产环境选Sarama，原因很直接：
- 纯Go实现，不需要CGO，交叉编译无痛。你的CI流水线用一个Docker镜像就能编译所有平台，不用折腾C编译器
- 社区活跃，文档完善，遇到问题能搜到解决方案
- 配置粒度细，生产级调优选项齐全，从批量大小到压缩算法到幂等生产者都能精确控制
- 原来是Shopify维护的，后来捐给了IBM，代码质量有保障
安装很简单：
bash
go get github.com/IBM/sarama@latest1.2 基础连接与配置
先看一个最小可用的连接配置。别小看这段代码，怕浪猫见过不少团队连版本号都不设，用默认的低版本配置跑在高版本Kafka上，莫名其妙地踩坑：
go
package main
import (
"fmt"
"log"
"github.com/IBM/sarama"
)
func main() {
// 创建配置实例
config := sarama.NewConfig()
// 设置Kafka版本（必须与集群版本匹配，否则可能出现协议不兼容）
// 怕浪猫的踩坑经验：不设版本，Sarama会用最低版本协议
// 高版本特性如幂等生产者、增量Rebalance都用不了
config.Version = sarama.V3_5_0_0
// 生产者配置
config.Producer.RequiredAcks = sarama.WaitForAll
config.Producer.Return.Successes = true
config.Producer.Return.Errors = true
// 连接Kafka集群
brokers := []string{"127.0.0.1:9092"}
client, err := sarama.NewClient(brokers, config)
if err != nil {
log.Fatalf("Failed to create Kafka client: %v", err)
}
defer client.Close()
// 检查Topic是否存在
topics, err := client.Topics()
if err != nil {
log.Fatalf("Failed to get topics: %v", err)
}
fmt.Println("Available topics:", topics)
// 获取指定Topic的分区信息
partitions, err := client.Partitions("test-topic")
if err != nil {
log.Fatalf("Failed to get partitions: %v", err)
}
fmt.Println("Partitions:", partitions)
}版本配置是第一个坑。Sarama默认版本很老，不设置Version会导致高版本Kafka的新特性不可用，设太高又会导致低版本集群报协议错误。你的配置文件里应该明确写死Kafka集群的版本号。
1.3 生产者配置与调优
生产者调优是Kafka性能优化的核心战场。怕浪猫逐个参数讲清楚，每个参数都会说明它影响什么，怎么设，为什么这么设：
go
func NewOptimizedProducer(brokers []string) (sarama.SyncProducer, error) {
config := sarama.NewConfig()
config.Version = sarama.V3_5_0_0
// ===== 可靠性配置 =====
// WaitForAll 等待所有ISR副本确认，最高可靠性
// 如果你的消息丢了无所谓，用WaitForLocal就行，延迟更低
config.Producer.RequiredAcks = sarama.WaitForAll
// 成功和失败都返回，方便业务处理
config.Producer.Return.Successes = true
config.Producer.Return.Errors = true
// 消息发送超时，超时后Sarama内部会重试
config.Producer.Timeout = 10 * time.Second
// ===== 批量配置（吞吐量核心） =====
// 批量发送的间隔，攒够这个时间就发
// 怕浪猫实测：日志场景设500ms，交易场景设50ms
config.Producer.Flush.Frequency = 500 * time.Millisecond
// 批量发送的消息数量，攒够这么多条就发
config.Producer.Flush.Messages = 200
// 批量发送的字节数，攒够这么多字节就发
// 三个条件满足任意一个就触发发送
config.Producer.Flush.Bytes = 1 * 1024 * 1024 // 1MB
// ===== 压缩配置 =====
// 开启压缩，大幅减少网络IO，但增加CPU开销
// 怕浪猫强烈推荐LZ4，CPU开销低，压缩率够用
config.Producer.Compression = sarama.CompressionLZ4
// 压缩级别，1最快，9最高压缩率，生产推荐3-4
config.Producer.CompressionLevel = 3
// ===== 重试配置 =====
// 发送失败重试次数
config.Producer.Retry.Max = 3
// 重试间隔，线性退避
config.Producer.Retry.Backoff = 100 * time.Millisecond
// ===== 分区选择 =====
// 使用Hash分区器，相同key的消息进同一分区，保证分区有序
config.Producer.Partitioner = sarama.NewHashPartitioner
// ===== 幂等配置（3.0+版本默认开启） =====
// 开启幂等生产者，防止重试导致的消息重复
// 原理：Kafka为每个生产者分配PID，用序列号去重
config.Producer.Idempotent = true
// 幂等模式下必须满足以下条件
// MaxOpenRequests=1表示同一连接只允许一个未确认的请求
config.Net.MaxOpenRequests = 1
producer, err := sarama.NewSyncProducer(brokers, config)
if err != nil {
return nil, fmt.Errorf("failed to create producer: %w", err)
}
return producer, nil
}实际发送消息的代码：
go
func SendMessage(producer sarama.SyncProducer, topic, key, value string) error {
msg := &sarama.ProducerMessage{
Topic: topic,
Key: sarama.StringEncoder(key),
Value: sarama.StringEncoder(value),
// 可以设置消息头，方便传递traceId等元信息
// 消息头不会参与分区计算，只是附加元数据
Headers: []sarama.RecordHeader{
{Key: []byte("traceId"), Value: []byte("abc-123")},
{Key: []byte("appId"), Value: []byte("order-service")},
},
Timestamp: time.Now(),
}
partition, offset, err := producer.SendMessage(msg)
if err != nil {
return fmt.Errorf("send message failed: %w", err)
}
log.Printf("Message sent to partition %d at offset %d\n", partition, offset)
return nil
}批量配置是吞吐量和延迟的跷跷板。Flush.Frequency太大，消息延迟高；太小，吞吐量上不去。怕浪猫的经验值：高频日志场景500ms/200条，交易场景50ms/10条。你必须在自己的硬件上压测，别人的参数不一定适合你。
关于压缩算法的选择，怕浪猫实测过各算法在真实日志数据上的表现：
| 压缩算法 | CPU开销 | 压缩率 | 压缩速度 | 适用场景 |
|---|---|---|---|---|
| None | 最低 | 0% | - | CPU敏感，网络充裕 |
| GZIP | 中 | 60-70% | 慢 | 兼容性最好，带宽极度紧张 |
| Snappy | 低 | 40-50% | 快 | 延迟敏感场景 |
| LZ4 | 低 | 40-50% | 最快 | 综合最优，生产推荐 |
| ZSTD | 中 | 65-75% | 较快 | 带宽紧张但CPU有富余 |
实测数据：在日志场景下（文本类消息，平均2KB/条），LZ4相比无压缩减少网络传输约45%，CPU开销增加约8%，是性价比最高的选择。ZSTD压缩率更高但CPU开销大约是LZ4的2倍，在带宽不是瓶颈的情况下不值得。
1.4 异步生产者
SyncProducer底层也是异步的，但每次SendMessage会阻塞等待确认。高吞吐场景应该用AsyncProducer，发送速度不受确认速度制约：
go
func NewAsyncProducer(brokers []string) (sarama.AsyncProducer, error) {
config := sarama.NewConfig()
config.Version = sarama.V3_5_0_0
config.Producer.RequiredAcks = sarama.WaitForAll
config.Producer.Return.Successes = true
config.Producer.Return.Errors = true
config.Producer.Flush.Frequency = 200 * time.Millisecond
config.Producer.Flush.Messages = 500
config.Producer.Compression = sarama.CompressionLZ4
producer, err := sarama.NewAsyncProducer(brokers, config)
if err != nil {
return nil, err
}
// 必须处理success和error通道，否则会阻塞
// 这是AsyncProducer使用中最常见的坑
go func() {
for success := range producer.Successes() {
// 生产环境通常不需要逐条记录成功日志
// 可以只更新指标计数器
log.Printf("OK: topic=%s partition=%d offset=%d",
success.Topic, success.Partition, success.Offset)
}
}()
go func() {
for err := range producer.Errors() {
log.Printf("ERR: %v", err)
// 这里应该做重试或告警
// 怕浪猫的实践：把失败消息写入本地落盘文件
// 后台协程定期重发
}
}()
return producer, nil
}AsyncProducer的Successes和Errors通道如果不消费，缓冲区满了之后生产者会永久阻塞。这个坑我踩过，排查了一整天——程序不报错也不退出，就是不发消息了，CPU和内存都正常，最后发现是Successes channel没人读。
异步发送的使用方式：
go
func SendAsync(producer sarama.AsyncProducer, topic, key, value string) {
msg := &sarama.ProducerMessage{
Topic: topic,
Key: sarama.StringEncoder(key),
Value: sarama.StringEncoder(value),
}
// 非阻塞发送，直接写入Input channel
producer.Input() <- msg
}1.5 消费者组管理
消费者组是Kafka实现消息分发和负载均衡的核心机制。同一个消费者组内的消费者共同消费Topic的所有分区，每个分区只被组内一个消费者消费。Sarama提供了ConsumerGroup接口：
go
// LogConsumer 实现sarama.ConsumerGroupHandler接口
type LogConsumer struct {
handler func(*sarama.ConsumerMessage) error
}
func (c *LogConsumer) Setup(sarama.ConsumerGroupSession) error {
// 消费者组启动时调用，可以做初始化
// 比如建立数据库连接、加载配置等
return nil
}
func (c *LogConsumer) Cleanup(sarama.ConsumerGroupSession) error {
// 消费者组关闭时调用，可以做清理
// 比如关闭数据库连接、刷写缓冲区等
return nil
}
func (c *LogConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
// 核心消费逻辑
// 注意：这个方法在每次Rebalance后会被重新调用
for message := range claim.Messages() {
// 处理消息
if err := c.handler(message); err != nil {
log.Printf("process message failed: %v, topic=%s partition=%d offset=%d",
err, message.Topic, message.Partition, message.Offset)
// 处理失败不提交offset，下次会重新消费
// 但要注意：如果是不可恢复的错误（比如消息格式错误）
// 不提交offset会导致后续消息全部被阻塞
// 这种情况应该跳过这条消息并记录到死信队列
continue
}
// 手动提交offset
session.MarkMessage(message, "")
}
return nil
}
func processMessage(msg *sarama.ConsumerMessage) error {
log.Printf("Received: topic=%s partition=%d offset=%d key=%s value=%s",
msg.Topic, msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
// 实际业务逻辑...
return nil
}启动消费者组的完整代码：
go
func StartConsumerGroup(brokers []string, topic, groupID string) error {
config := sarama.NewConfig()
config.Version = sarama.V3_5_0_0
// 消费者组配置
// 分区分配策略：Range是默认策略，按分区范围分配
// CooperativeSticky更平滑，Rebalance时不会暂停所有消费
config.Consumer.Group.Rebalance.GroupStrategies = []sarama.BalanceStrategy{
sarama.NewBalanceStrategyCooperativeSticky(),
}
// Session超时：消费者在这个时间内没发心跳就被认为挂了
// 设太短会导致误判，设太长会导致真正挂掉后很久才Rebalance
config.Consumer.Group.Session.Timeout = 15 * time.Second
// Rebalance超时：等待所有消费者加入Rebalance的时间
config.Consumer.Group.Rebalance.Timeout = 60 * time.Second
// 心跳间隔，一般为Session超时的1/3
config.Consumer.Group.Heartbeat.Interval = 5 * time.Second
// Offset提交方式：手动提交
// 自动提交的风险：消息还没处理完就提交了，crash后消息丢失
config.Consumer.Offsets.AutoCommit.Enable = false
// 从最早的消息开始消费（首次加入消费者组时）
// 已提交offset的消费者不受影响，从已提交位置继续
config.Consumer.Offsets.Initial = sarama.OffsetOldest
// 消费超时：从Kafka读取消息的超时时间
config.Net.ReadTimeout = 30 * time.Second
client, err := sarama.NewConsumerGroup(brokers, groupID, config)
if err != nil {
return err
}
defer client.Close()
ctx := context.Background()
handler := &LogConsumer{
handler: processMessage,
}
for {
// Consume会在rebalance时返回，需要循环调用
// 这是Sarama的设计：Rebalance后需要重新调用Consume
err := client.Consume(ctx, []string{topic}, handler)
if err != nil {
return err
}
// 检查context是否被取消
if ctx.Err() != nil {
return ctx.Err()
}
}
}消费者组的Rebalance是Kafka运维中最容易被忽视的问题。一次Rebalance会导致所有消费者暂停消费，短则几秒，长则几十秒。如果你的处理耗时超过session.timeout，消费者会被踢出组触发Rebalance，形成恶性循环。解法是调大session超时，或者用Cooperative策略。
1.6 手动提交Offset详解
自动提交Offset方便但有风险：消息还没处理完就提交了，如果此时crash，消息就丢了。生产环境必须手动提交。
但手动提交也有讲究，怕浪猫见过三种错误做法：
- 每条消息处理完就提交——性能差，频繁的OffsetCommit请求拖慢消费
- 批量处理完一起提交——但没处理完就crash，这批全部重消费
- 定时提交——在准确性和性能之间取平衡，生产推荐
go
func (c *LogConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
for {
select {
case message, ok := <-claim.Messages():
if !ok {
return nil
}
// 1. 反序列化
var logEntry LogEntry
if err := json.Unmarshal(message.Value, &logEntry); err != nil {
// 反序列化失败，记录错误并跳过，避免毒丸消息阻塞消费
// 这是毒丸消息的处理：格式错误的消息永远无法被正确处理
// 不跳过它会导致消费者一直卡在这条消息上
log.Printf("unmarshal failed: %v, offset=%d", err, message.Offset)
session.MarkMessage(message, "")
continue
}
// 2. 业务处理
if err := c.handleLogEntry(&logEntry); err != nil {
// 处理失败，根据策略决定是否重试
if c.isRetryable(err) {
// 可重试的错误，不提交offset，让下次重新消费
// 但要注意退避，避免快速重试导致CPU打满
time.Sleep(time.Second)
continue
}
// 不可重试的错误，跳过并记录到死信队列
c.sendToDeadLetterQueue(message, err)
}
// 3. 处理成功，手动提交offset
// MarkMessage只是标记，真正的提交在session.Commit()时发生
// 但Sarama会在ConsumeClaim返回时自动提交被标记的offset
session.MarkMessage(message, "")
case <-session.Context().Done():
return nil
}
}
}毒丸消息（Poison Pill）是消费端最隐蔽的坑。一条格式错误的消息导致处理永远失败，如果不跳过它，会阻塞整个分区。解法是设最大重试次数，超限后投递到死信队列。怕浪猫在生产环境遇到过JSON里带了BOM头导致的毒丸消息，排查了一整天。
二、消息可靠性保障
2.1 生产端可靠性
消息从生产到消费，要经过三个环节：生产者到Kafka集群、Kafka集群内部同步、Kafka集群到消费者。每个环节都可能丢消息。
生产端可靠性三板斧：ACK机制、重试策略、本地缓冲。
ACK机制有三个级别，理解它们对于配置可靠性至关重要：
go
// ACK级别详解
// 0 (NoResponse): 生产者不等待任何确认，发了就算成功
// 最快但最容易丢消息：网络丢包、Broker宕机都会丢
config.Producer.RequiredAcks = sarama.NoResponse
// 1 (WaitForLocal): 等待Leader写入确认
// Leader确认了就认为成功，但Leader挂了且副本还没同步时会丢
// 这是性能和可靠性的折中
config.Producer.RequiredAcks = sarama.WaitForLocal
// -1 (WaitForAll): 等待所有ISR（In-Sync Replicas）副本确认
// 最高可靠性，只要还有一个ISR副本存活，消息就不丢
// 配合min.insync.replicas=2使用，保证至少两个副本有这条消息
config.Producer.RequiredAcks = sarama.WaitForAll生产环境用WaitForAll配合幂等生产者，这是标准做法：
go
func NewReliableProducer(brokers []string) (sarama.SyncProducer, error) {
config := sarama.NewConfig()
config.Version = sarama.V3_5_0_0
// 可靠性配置
config.Producer.RequiredAcks = sarama.WaitForAll
// 幂等生产者：防止重试导致的重复消息
// 原理是Kafka为每个生产者分配一个PID（Producer ID）
// 每条消息带一个递增的SequenceNumber
// Broker端用<PID, partition, sequenceNumber>去重
config.Producer.Idempotent = true
// 幂等模式下必须MaxOpenRequests=1
// 否则并发的请求可能导致Broker端序列号乱序
config.Net.MaxOpenRequests = 1
// 重试配置
config.Producer.Retry.Max = 5
// 指数退避，避免重试风暴
config.Producer.Retry.Backoff = func(retry int) time.Duration {
// 100ms, 200ms, 400ms, 800ms, 1600ms
return time.Duration(1<<uint(retry)) * 100 * time.Millisecond
}
// 超时配置
config.Producer.Timeout = 10 * time.Second
config.Producer.Return.Successes = true
config.Producer.Return.Errors = true
return sarama.NewSyncProducer(brokers, config)
}但SyncProducer重试有限，极端情况（网络长时间断开）还是会丢消息。生产级方案需要本地缓冲，这就是怕浪猫在生产环境用的方案：
go
type ReliableProducer struct {
producer sarama.SyncProducer
topic string
queue chan *sarama.ProducerMessage
stopCh chan struct{}
maxRetry int
wg sync.WaitGroup
}
func NewReliableProducerWithBuffer(brokers []string, topic string, bufSize int) (*ReliableProducer, error) {
producer, err := NewReliableProducer(brokers)
if err != nil {
return nil, err
}
p := &ReliableProducer{
producer: producer,
topic: topic,
queue: make(chan *sarama.ProducerMessage, bufSize),
stopCh: make(chan struct{}),
maxRetry: 5,
}
p.wg.Add(1)
go p.dispatch()
return p, nil
}
func (p *ReliableProducer) dispatch() {
defer p.wg.Done()
for {
select {
case msg := <-p.queue:
p.sendWithRetry(msg)
case <-p.stopCh:
// 退出前把队列里的消息发完
for len(p.queue) > 0 {
msg := <-p.queue
p.sendWithRetry(msg)
}
return
}
}
}
func (p *ReliableProducer) sendWithRetry(msg *sarama.ProducerMessage) {
for i := 0; i < p.maxRetry; i++ {
_, _, err := p.producer.SendMessage(msg)
if err == nil {
return
}
log.Printf("send failed (attempt %d/%d): %v", i+1, p.maxRetry, err)
time.Sleep(time.Duration(1<<uint(i)) * time.Second)
}
// 重试全部失败，写入本地落盘文件，等待人工处理或后台重发
p.fallbackToDisk(msg)
}
func (p *ReliableProducer) fallbackToDisk(msg *sarama.ProducerMessage) {
// 简化实现：将失败消息序列化写入本地文件
// 生产环境可以用BadgerDB或BoltDB做持久化队列
file, err := os.OpenFile("failed_messages.jsonl",
os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
if err != nil {
log.Printf("fallback to disk failed: %v", err)
return
}
defer file.Close()
value, _ := msg.Value.Encode()
key, _ := msg.Key.Encode()
record := map[string]interface{}{
"topic": msg.Topic,
"key": string(key),
"value": string(value),
"timestamp": msg.Timestamp,
}
line, _ := json.Marshal(record)
file.Write(append(line, '\n'))
}
func (p *ReliableProducer) Send(key, value string) error {
msg := &sarama.ProducerMessage{
Topic: p.topic,
Key: sarama.StringEncoder(key),
Value: sarama.StringEncoder(value),
}
select {
case p.queue <- msg:
return nil
default:
// 队列满了，直接拒绝，避免OOM
// 这种情况应该告警，说明生产速度远超消费速度
return fmt.Errorf("producer buffer full")
}
}
func (p *ReliableProducer) Close() {
close(p.stopCh)
p.wg.Wait()
p.producer.Close()
}可靠性没有银弹，只有取舍。你愿意为0.01%的可靠性提升付出多少延迟和复杂度的代价，这个问题的答案因业务而异。核心交易用WaitForAll+本地缓冲+幂等，边缘业务用WaitForLocal就够了。
2.2 消费端可靠性：幂等消费
网络重试、生产者重试、消费者Rebalance都可能导致消息重复。比如生产者发送消息后超时了，但消息实际已经到达Broker，重试就会产生重复。消费端必须实现幂等性。
幂等消费的核心思路：用唯一标识（消息ID或业务ID）做去重。去重存储可以用Redis（高性能、有TTL自动过期）或数据库唯一索引（强一致但性能差）：
go
type IdempotentConsumer struct {
redis *redis.Client
handler func(*sarama.ConsumerMessage) error
}
func NewIdempotentConsumer(rdb *redis.Client, handler func(*sarama.ConsumerMessage) error) *IdempotentConsumer {
return &IdempotentConsumer{
redis: rdb,
handler: handler,
}
}
func (c *IdempotentConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
for message := range claim.Messages() {
// 用消息的key或自定义唯一ID做幂等key
// 优先用业务ID（如订单号），没有再用消息元信息
msgID := string(message.Key)
if msgID == "" {
msgID = fmt.Sprintf("%s-%d-%d",
message.Topic, message.Partition, message.Offset)
}
// Redis SETNX 原子操作，过期时间根据业务设置
// 过期时间要大于消息可能重复的时间窗口
// 比如消费者Rebalance可能导致5分钟内的消息重复
// 那TTL至少设10分钟
key := fmt.Sprintf("msg:dedup:%s", msgID)
set, err := c.redis.SetNX(context.Background(), key, "1", 24*time.Hour).Result()
if err != nil {
log.Printf("redis setnx failed: %v", err)
// Redis不可用时的策略选择：
// 1. 保守策略：不跳过，可能重复处理（默认）
// 2. 严格策略：不处理，等Redis恢复（适用于严格幂等场景）
continue
}
if !set {
// 已经处理过这条消息，跳过
// 这是幂等消费的核心：重复消息直接跳过
session.MarkMessage(message, "")
continue
}
// 执行业务逻辑
if err := c.handler(message); err != nil {
// 处理失败，删除幂等key，允许下次重试
// 否则这条消息永远无法被重新处理
c.redis.Del(context.Background(), key)
return err
}
session.MarkMessage(message, "")
}
return nil
}2.3 死信队列实现
处理失败的消息不能丢，也不能无限重试阻塞消费。死信队列（Dead Letter Queue）是标准解法。思路是：消息处理失败达到最大重试次数后，把消息转发到一个专门的死信Topic，由专门的消费者或人工处理：
go
type DeadLetterHandler struct {
producer sarama.SyncProducer
mainTopic string
dlqTopic string
maxRetry int
retryTracker *redis.Client
}
func (h *DeadLetterHandler) Handle(msg *sarama.ConsumerMessage, processErr error) error {
// 用Redis记录每条消息的重试次数
retryKey := fmt.Sprintf("msg:retry:%s-%d-%d",
msg.Topic, msg.Partition, msg.Offset)
// 原子递增重试次数
retryCount, _ := h.retryTracker.Incr(context.Background(), retryKey).Result()
if retryCount == 1 {
// 第一次失败，设置过期时间
// 过期后重试计数清零，相当于给消息一个"重新开始"的机会
h.retryTracker.Expire(context.Background(), retryKey, 24*time.Hour)
}
if retryCount > int64(h.maxRetry) {
// 超过最大重试次数，投递到死信队列
// 死信消息需要携带完整的上下文信息，方便排查
dlqMsg := &sarama.ProducerMessage{
Topic: h.dlqTopic,
Key: msg.Key,
Value: msg.Value,
Headers: []sarama.RecordHeader{
{Key: []byte("originalTopic"), Value: []byte(msg.Topic)},
{Key: []byte("originalPartition"), Value: []byte(fmt.Sprintf("%d", msg.Partition))},
{Key: []byte("originalOffset"), Value: []byte(fmt.Sprintf("%d", msg.Offset))},
{Key: []byte("error"), Value: []byte(processErr.Error())},
{Key: []byte("retryCount"), Value: []byte(fmt.Sprintf("%d", retryCount))},
{Key: []byte("deadAt"), Value: []byte(time.Now().Format(time.RFC3339))},
},
}
_, _, err := h.producer.SendMessage(dlqMsg)
if err != nil {
return fmt.Errorf("send to DLQ failed: %w", err)
}
// 清理重试计数
h.retryTracker.Del(context.Background(), retryKey)
log.Printf("Message sent to DLQ: topic=%s partition=%d offset=%d",
msg.Topic, msg.Partition, msg.Offset)
return nil
}
// 未超过重试次数，返回错误让消费者不提交offset
// 下次会重新消费这条消息
return processErr
}死信队列不是垃圾桶，是安全网。DLQ里的消息要有专人定期处理和分析，否则它就变成了你假装没看见的问题堆。怕浪猫的实践：每天早上看一次DLQ的消息数量趋势，突增必有妖。
2.4 消息积压处理
消息积压是生产环境最常见的运维事件。怕浪猫在之前的公司经历过一次严重的积压事故：大促期间订单消息暴增十倍，消费者处理速度跟不上，Lag从正常的几百条飙升到五十万条。运营说用户下单后迟迟收不到确认短信，客服电话被打爆。我们从发现到处理完花了四个小时，期间损失了大量订单。事后复盘发现，消费者处理逻辑里有一个同步调外部支付接口的操作，平时延迟正常，大促时支付接口也扛不住，导致我们的消费速度跟着暴跌。
这次事故让我总结了一套处理积压的标准流程。这个清单建议打印贴在工位上，关键时刻照着做：
消息积压应急处理清单：
- 确认积压程度：通过Kafka监控查看各分区的Lag（消费者落后生产者的消息数），区分是整体积压还是单分区积压。单分区积压通常是数据倾斜导致，比如某个key的消息量远超其他key
- 判断原因：消费者处理太慢？消费者实例不够？某分区数据倾斜？消费者crash了？下游依赖（数据库、外部接口）变慢了？
- 紧急扩容：增加消费者实例数（注意不能超过分区数，多出的消费者会闲置）。如果分区数也不够，需要紧急扩分区，但扩分区会影响顺序性
- 降级策略：非核心逻辑临时关闭。比如关闭二级缓存刷新、关闭非必要DB写入、跳过部分校验逻辑、关闭报表生成。核心目标是把消费速度提上去
- 批量消费：从逐条处理改为批量处理，减少IO次数。批量写数据库、批量调外部接口，吞吐量可以提升5到10倍
- 跳过策略：如果消息有时效性（如实时推荐、实时风控），考虑跳过过期消息。过期消息处理了也没意义，反而拖慢整体消费
- 事后复盘：积压原因分析、容量评估、预防措施、告警阈值调整。复盘要有产出物——要么改了配置，要么加了告警，要么改了代码
下面这个批量消费示例展示了如何把逐条处理改成批量处理。核心思路是攒一批再一起处理，用时间和数量两个条件控制flush时机：
go
type BatchConsumer struct {
batchSize int
batchTimeout time.Duration
handler func([]*sarama.ConsumerMessage) error
}
func NewBatchConsumer(batchSize int, batchTimeout time.Duration,
handler func([]*sarama.ConsumerMessage) error) *BatchConsumer {
return &BatchConsumer{
batchSize: batchSize,
batchTimeout: batchTimeout,
handler: handler,
}
}
func (c *BatchConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
batch := make([]*sarama.ConsumerMessage, 0, c.batchSize)
timer := time.NewTimer(c.batchTimeout)
defer timer.Stop()
// flush函数：批量处理并提交offset
flush := func() error {
if len(batch) == 0 {
return nil
}
// 调用业务handler批量处理
if err := c.handler(batch); err != nil {
return err
}
// 批量提交offset
for _, msg := range batch {
session.MarkMessage(msg, "")
}
batch = batch[:0]
return nil
}
for {
select {
case msg, ok := <-claim.Messages():
if !ok {
// channel关闭，刷入剩余消息
return flush()
}
batch = append(batch, msg)
// 攒够batchSize就flush
if len(batch) >= c.batchSize {
if err := flush(); err != nil {
return err
}
timer.Reset(c.batchTimeout)
}
case <-timer.C:\n\t\t\t// 超时也flush，避免消息积压在batch里太久\n\t\t\tif err := flush(); err != nil {
return err
}
timer.Reset(c.batchTimeout)
case <-session.Context().Done():
// 退出前flush
return flush()
}
}
}积压处理的核心原则：先止血，再治病。别在告警还在响的时候做根因分析，先把消费速度提上去。等积压清零了，再坐下来复盘为什么会积压，怎么预防。
三、消息顺序性实现
3.1 全局有序 vs 分区有序
Kafka只能保证单个分区内消息有序。全局有序需要把所有消息发到同一个分区，但这意味着完全丧失并行能力——所有消息排队处理，吞吐量直接降为单线程级别。在实际项目中，怕浪猫从未遇到过真正需要全局有序的场景。即使是最严格的金融交易场景，也是按账户ID分区，同一账户的交易有序就够了。
要理解这个问题，需要搞清楚Kafka的分区机制。Topic被分成多个分区，每个分区是一个有序的、不可变的日志。消息被追加到分区末尾，按写入顺序排列。不同分区之间没有顺序保证。这就是为什么按Key分区能保证同Key有序——相同Key的消息总是被Hash到同一个分区。
大部分场景下，按业务ID分区就够了——同一个订单的事件保证有序，不同订单之间不需要有序。订单A的创建、支付、发货需要按顺序处理，但订单A和订单B之间没有先后关系。
| 方案 | 顺序保证 | 吞吐量 | 适用场景 |
|---|---|---|---|
| 全局有序（1分区） | 全局 | 极低 | 极少数强顺序场景，如 Binlog 同步 |
| 分区有序（按Key分区） | 同Key有序 | 高 | 大部分业务场景，如订单状态流转 |
| 无序 | 无 | 最高 | 日志、监控指标 |
怕浪猫见过最离谱的案例：一个团队为了"保证顺序"，把Topic设成1个分区，然后抱怨Kafka吞吐量太低。改成按订单ID分区后，吞吐量直接翻了20倍。还有团队把所有消息都用空Key发送，结果消息均匀分布到各分区，完全无序，然后又抱怨消费端处理乱序。这都是对Kafka分区机制理解不深导致的。
3.2 顺序消费实现
要实现顺序消费，需要两个条件：
- 生产端：相同业务Key的消息发到同一分区
- 消费端：同一分区的消息串行处理，不能并发
生产端用HashPartitioner（Sarama默认就是），关键是消费端不能并发处理同一分区的消息：
go
type OrderedConsumer struct {
handler func(*sarama.ConsumerMessage) error
}
func (c *OrderedConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
// 串行处理，不启动goroutine并发
// 这是顺序消费的关键：必须串行
for message := range claim.Messages() {
if err := c.handler(message); err != nil {
// 处理失败，不提交offset，暂停消费等待重试
// 注意：这里会阻塞整个分区的后续消息
// 这是顺序消费的代价：一条消息卡住，后面全部排队
log.Printf("ordered consume failed: %v, will retry", err)
// 简单退避后重试
time.Sleep(time.Second)
continue
}
session.MarkMessage(message, "")
}
return nil
}但如果处理逻辑里有耗时操作（比如调外部API、等数据库锁），串行处理会导致吞吐量暴跌。解法是引入顺序队列加异步worker——每个分区一个worker goroutine，分区内串行，分区间并行：
go
type PartitionOrderedHandler struct {
workers map[int32]chan *sarama.ConsumerMessage
wg sync.WaitGroup
handler func(*sarama.ConsumerMessage) error
}
func NewPartitionOrderedHandler(partitionCount int,
handler func(*sarama.ConsumerMessage) error) *PartitionOrderedHandler {
h := &PartitionOrderedHandler{
workers: make(map[int32]chan *sarama.ConsumerMessage),
handler: handler,
}
// 为每个分区创建一个有序处理channel
for i := 0; i < partitionCount; i++ {
ch := make(chan *sarama.ConsumerMessage, 1000)
h.workers[int32(i)] = ch
h.wg.Add(1)
go h.processOrdered(ch)
}
return h
}
func (h *PartitionOrderedHandler) processOrdered(ch <-chan *sarama.ConsumerMessage) {
defer h.wg.Done()
// 单goroutine消费channel，保证分区内有序
for msg := range ch {
if err := h.handler(msg); err != nil {
log.Printf("process failed: %v", err)
// 重新入队或走死信队列
}
}
}
func (h *PartitionOrderedHandler) ConsumeClaim(session sarama.ConsumerGroupSession,
claim sarama.ConsumerGroupClaim) error {
// 获取当前分区ID，找到对应的worker channel
partition := claim.Partition()
ch, ok := h.workers[partition]
if !ok {
return fmt.Errorf("no worker for partition %d", partition)
}
for message := range claim.Messages() {
ch <- message
session.MarkMessage(message, "")
}
return nil
}顺序和性能是天生的矛盾体。先想清楚你需要的是"全局有序"还是"局部有序"，90%的场景答案都是后者。不要为了简单而用全局有序，那是在用吞吐量换代码简单度，不值。
四、延迟消息实现
4.1 延迟消息的使用场景
延迟消息在业务中很常见，几乎每个后端系统都会用到。怕浪猫整理了常见的延迟消息场景：
- 订单30分钟未支付自动关闭：电商系统的标配功能。用户下单后如果不支付，系统需要在30分钟后自动关闭订单并释放库存。用延迟消息实现非常优雅——下单时发一条延迟30分钟的消息，消费时检查订单状态，已支付则忽略，未支付则关闭
- 会议开始前15分钟发提醒：日历系统中，用户创建会议后，需要在会议开始前提醒参与者
- 用户注册后7天发召回推送：运营策略中常见的用户留存手段，新用户注册后间隔不同时间点发送不同内容的推送
- 重试任务按递增间隔执行：外部接口调用失败后，按1秒、2秒、4秒、8秒的间隔重试，避免短时间内大量重试打垮下游
- 预约任务定时执行：用户预约了某个时间点的服务，系统需要在指定时间点触发
实现延迟消息有多种方案，各有优劣。用Go的time.AfterFunc或time.NewTimer最简单，但进程重启后任务就丢了。用Redis的ZSET可以持久化，但需要自己实现调度逻辑。用RocketMQ自带延迟消息功能最方便，但引入新中间件成本高。如果已经在用Kafka，可以用多级Topic方案模拟延迟消息，虽然只支持固定延迟级别，但覆盖了大部分业务场景。
4.2 时间轮算法原理
时间轮（Hashed Wheel Timer）是高效的延迟任务调度算法。想象一个时钟表盘：
- 表盘有N个槽位，每个槽位是一个链表
- 指针每隔一个tick间隔转动一个槽位
- 添加任务时，计算延迟时间对应的槽位，放入链表
- 指针转到某个槽位时，检查链表中的任务是否到期，到期则执行
时间轮的优势在于：添加任务O(1)，取消任务O(1)，到期检查O(1)。相比优先队列（添加O(logN)）在海量延迟任务场景下性能优势明显。
Go实现简化版时间轮：
go
type DelayTask struct {
ID string
ExecuteAt time.Time
Callback func()
}
type TimeWheel struct {
slots [][]*DelayTask
tick time.Duration
slotsNum int
current int
stopCh chan struct{}
mu sync.Mutex
}
func NewTimeWheel(tick time.Duration, slotsNum int) *TimeWheel {
tw := &TimeWheel{
slots: make([][]*DelayTask, slotsNum),
tick: tick,
slotsNum: slotsNum,
stopCh: make(chan struct{}),
}
go tw.run()
return tw
}
func (tw *TimeWheel) AddTask(delay time.Duration, callback func()) string {
task := &DelayTask{
ID: uuid.New().String(),
ExecuteAt: time.Now().Add(delay),
Callback: callback,
}
tw.mu.Lock()
defer tw.mu.Unlock()
// 计算任务应该放在哪个槽位
ticks := int(delay / tw.tick)
if ticks < 1 {
ticks = 1
}
slot := (tw.current + ticks) % tw.slotsNum
tw.slots[slot] = append(tw.slots[slot], task)
return task.ID
}
func (tw *TimeWheel) run() {
ticker := time.NewTicker(tw.tick)
defer ticker.Stop()
for {
select {
case <-ticker.C:\n\t\t\ttw.tickHandler()\n\t\tcase <-tw.stopCh:
return
}
}
}
func (tw *TimeWheel) tickHandler() {
tw.mu.Lock()
defer tw.mu.Unlock()
// 取出当前槽位的所有任务
tasks := tw.slots[tw.current]
tw.slots[tw.current] = nil
now := time.Now()
for _, task := range tasks {
if now.Before(task.ExecuteAt) {
// 还没到执行时间，重新放入对应槽位
// 这处理了tick精度问题和多轮时间轮的场景
remaining := task.ExecuteAt.Sub(now)
ticks := int(remaining / tw.tick)
if ticks < 1 {
ticks = 1
}
slot := (tw.current + ticks) % tw.slotsNum
tw.slots[slot] = append(tw.slots[slot], task)
} else {
// 执行任务，用goroutine避免阻塞时间轮
go task.Callback()
}
}
// 指针前进一步
tw.current = (tw.current + 1) % tw.slotsNum
}
func (tw *TimeWheel) Stop() {
close(tw.stopCh)
}时间轮的精髓在于：用空间换时间。把O(n)的扫描降级为O(1)的槽位定位，海量延迟任务的调度成本极低。Netty的HashedWheelTimer、Kafka的延迟操作都用了这个算法。
4.3 基于Kafka的延迟消息实现
Kafka本身不支持延迟消息（这是RocketMQ的优势），但可以用多级Topic来模拟。思路是：创建多个不同延迟级别的Topic，消息先发到延迟Topic，延迟消费者到时间后转发到目标Topic：
go
// 延迟消息服务
type DelayMessageService struct {
producer sarama.SyncProducer
delayTopics map[int]string // 延迟级别 -> topic名
}
// 预定义延迟级别，类似RocketMQ的设计
// 固定级别比任意延迟更高效，因为不需要排序
var delayLevels = map[int]time.Duration{
1: 1 * time.Second,
2: 5 * time.Second,
3: 10 * time.Second,
4: 30 * time.Second,
5: 1 * time.Minute,
6: 5 * time.Minute,
7: 10 * time.Minute,
8: 30 * time.Minute,
9: 1 * time.Hour,
10: 2 * time.Hour,
}
func (s *DelayMessageService) SendDelayMessage(topic string, key, value string,
delay time.Duration) error {
// 找到最接近的延迟级别
// 只能向上取整，不能向下（否则延迟不够）
level := s.findDelayLevel(delay)
delayTopic := s.delayTopics[level]
// 将原始topic和消息内容封装到消息头
// 延迟消费者到时间后根据header转发到目标topic
msg := &sarama.ProducerMessage{
Topic: delayTopic,
Key: sarama.StringEncoder(key),
Value: sarama.StringEncoder(value),
Headers: []sarama.RecordHeader{
{Key: []byte("targetTopic"), Value: []byte(topic)},
{Key: []byte("delayLevel"), Value: []byte(fmt.Sprintf("%d", level))},
{Key: []byte("sendTime"), Value: []byte(time.Now().Format(time.RFC3339Nano))},
},
}
_, _, err := s.producer.SendMessage(msg)
return err
}
func (s *DelayMessageService) findDelayLevel(delay time.Duration) int {
for i := 1; i <= 10; i++ {
if delayLevels[i] >= delay {
return i
}
}
return 10 // 最大延迟级别
}
// 延迟Topic消费者：到时间后转发到目标Topic
func (s *DelayMessageService) StartDelayConsumer(brokers []string, level int) error {
topic := s.delayTopics[level]
delay := delayLevels[level]
config := sarama.NewConfig()
config.Version = sarama.V3_5_0_0
config.Consumer.Offsets.Initial = sarama.OffsetOldest
consumer, err := sarama.NewConsumer(brokers, config)
if err != nil {
return err
}
partitions, err := consumer.Partitions(topic)
if err != nil {
return err
}
for _, partition := range partitions {
pc, err := consumer.ConsumePartition(topic, partition, sarama.OffsetOldest)
if err != nil {
return err
}
go func(pc sarama.PartitionConsumer) {
for msg := range pc.Messages() {
// 解析发送时间
sendTimeStr := ""
targetTopic := ""
for _, h := range msg.Headers {
if string(h.Key) == "sendTime" {
sendTimeStr = string(h.Value)
}
if string(h.Key) == "targetTopic" {
targetTopic = string(h.Value)
}
}
sendTime, _ := time.Parse(time.RFC3339Nano, sendTimeStr)
// 如果还没到延迟时间，等待
elapsed := time.Since(sendTime)
if elapsed < delay {
time.Sleep(delay - elapsed)
}
// 转发到目标Topic
if targetTopic == "" {
continue
}
forwardMsg := &sarama.ProducerMessage{
Topic: targetTopic,
Key: msg.Key,
Value: msg.Value,
}
s.producer.SendMessage(forwardMsg)
}
}(pc)
}
return nil
}RocketMQ的延迟消息就是用类似的多级Topic方案实现的，开源版支持18个固定延迟级别。商业版才支持任意时间延迟，原理是基于时间轮加RocksDB存储。如果你的项目主要用Kafka，上面的方案就能满足大部分延迟消息需求。
五、实战项目：日志收集与监控系统
理论讲够了，现在来干一个完整的实战项目：日志收集与监控系统。这个项目会用到前面讲的所有知识点——Kafka生产者、消费者、批量处理、告警等。
5.1 系统架构设计
在开始写代码之前，先想清楚架构。好的架构设计能让后面的开发事半功倍，差的架构会让你在运维中疲于奔命。
整体架构如下，数据从左到右流动。每个组件都有明确的职责边界，组件之间通过Kafka解耦，任何一层出问题都不会影响其他层：
[应用服务器] -> [日志Agent] -> [Kafka集群] -> [日志消费者] -> [Elasticsearch]
|
v
[查询API服务]
|
v
[告警引擎] -> [通知渠道]
|
v
[Prometheus+Grafana大盘]技术选型理由，每一项都有明确的Why：
- 日志采集：自研Go Agent而非Filebeat，因为我们需要对日志内容做预处理（解析JSON、提取字段、打标签），Go的并发模型和低资源占用适合常驻Agent。Filebeat虽然成熟但扩展性差，自定义解析逻辑需要写Lua脚本，调试困难
- 消息中间件：Kafka，高吞吐、持久化、支持回溯。Kafka的持久化特性意味着即使消费者挂了，消息也不会丢，恢复后可以继续消费。这是日志系统可靠性的基础
- 日志存储：Elasticsearch，全文搜索能力强，支持复杂聚合查询。ES的倒排索引让关键词搜索极快，聚合分析能力让日志统计变得简单。相比Loki等轻量方案，ES更重但功能更强
- 查询API：Go + gin，轻量高效。直接对接ES，封装查询接口，前端不需要直接操作ES
- 告警引擎：Go规则引擎 + 多渠道通知。独立于消费链路，不会因为告警逻辑拖慢日志消费
- 监控大盘：Prometheus采集指标 + Grafana可视化，业界标准方案。监控指标包括采集速率、消费Lag、ES写入延迟、告警触发次数等
架构设计的关键原则：
- 单一职责：每个组件只做一件事，Agent只管采集，Consumer只管消费存储，AlertEngine只管告警
- 解耦：组件之间通过Kafka或HTTP通信，任何一层可以独立升级或替换
- 可观测：每个组件都暴露Prometheus指标，系统自身的健康状态也要被监控
- 容错：任何一层挂了不影响其他层。Agent挂了Kafka消息不增加，Consumer挂了Kafka消息积压但不丢，ES挂了Consumer可以暂停消费
5.2 设计日志收集Agent
日志Agent是整个系统的入口，运行在每台应用服务器上。核心功能：读取日志文件增量内容，解析后发送到Kafka。关键设计点包括文件跟踪、断点续传、多格式解析。Agent必须是轻量的——CPU占用不超过10%，内存占用不超过100MB，不能影响应用本身的运行。
go
package agent
import (
"bufio"
"context"
"encoding/json"
"fmt"
"io"
"log"
"os"
"path/filepath"
"strings"
"sync"
"time"
"github.com/IBM/sarama"
"github.com/hpcloud/tail"
)
// LogEntry 日志条目结构
type LogEntry struct {
Timestamp string `json:"timestamp"`
Level string `json:"level"`
Source string `json:"source"`
Host string `json:"host"`
Path string `json:"path"`
Message string `json:"message"`
Tags map[string]string `json:"tags,omitempty"`
}
// FileCollector 文件采集器
type FileCollector struct {
filePath string
source string
host string
tags map[string]string
producer sarama.AsyncProducer
topic string
offsetStore *OffsetStore
}
func NewFileCollector(filePath, source, topic string,
producer sarama.AsyncProducer, offsetStore *OffsetStore) *FileCollector {
host, _ := os.Hostname()
return &FileCollector{
filePath: filePath,
source: source,
host: host,
tags: make(map[string]string),
producer: producer,
topic: topic,
offsetStore: offsetStore,
}
}
func (c *FileCollector) Start(ctx context.Context) error {
// 使用tail库跟踪文件变化
// tail库底层使用inotify（Linux）或kqueue（macOS）监听文件变化
// 比轮询高效得多
tailConfig := tail.Config{
Follow: true, // 持续跟踪新内容
ReOpen: true, // 文件被rotate后重新打开
MustExist: true, // 文件必须存在
Poll: false, // 使用inotify而非轮询
Location: &tail.SeekInfo{
Offset: c.offsetStore.Load(c.filePath),
Whence: io.SeekStart, // 从上次记录的位置开始
},
}
t, err := tail.TailFile(c.filePath, tailConfig)
if err != nil {
return fmt.Errorf("tail file %s failed: %w", c.filePath, err)
}
log.Printf("Start collecting: %s from offset %d",
c.filePath, c.offsetStore.Load(c.filePath))
var lineCount int64
lastSave := time.Now()
for {
select {
case line, ok := <-t.Lines:
if !ok {
return nil
}
if line.Err != nil {
log.Printf("tail error: %v", line.Err)
continue
}
c.processLine(line.Text)
lineCount++
// 定期保存offset，避免重启后重复采集太多
if time.Since(lastSave) > 5*time.Second {
pos, err := t.Seek(0, io.SeekCurrent)
if err == nil {
c.offsetStore.Save(c.filePath, pos)
}
lastSave = time.Now()
}
case <-ctx.Done():
// 退出前保存offset
pos, err := t.Seek(0, io.SeekCurrent)
if err == nil {
c.offsetStore.Save(c.filePath, pos)
}
t.Stop()
return ctx.Err()
}
}
}
func (c *FileCollector) processLine(line string) {
start := time.Now()
// 解析日志行，支持多种格式
entry := c.parseLine(line)
if entry == nil {
metrics.LogsCollectErrors.WithLabelValues(
c.source, c.host, "parse_error").Inc()
return
}
// 序列化并发送到Kafka
data, err := json.Marshal(entry)
if err != nil {
metrics.LogsCollectErrors.WithLabelValues(
c.source, c.host, "marshal_error").Inc()
return
}
msg := &sarama.ProducerMessage{
Topic: c.topic,
// 按主机名分区，同一主机的日志进同一分区
// 这样可以保证同一主机的日志在Kafka中是有序的
Key: sarama.StringEncoder(c.host),
Value: sarama.ByteEncoder(data),
Timestamp: time.Now(),
Headers: []sarama.RecordHeader{
{Key: []byte("source"), Value: []byte(c.source)},
{Key: []byte("host"), Value: []byte(c.host)},
},
}
// 异步发送，不阻塞采集
c.producer.Input() <- msg
// 记录指标
metrics.LogsCollectedTotal.WithLabelValues(c.source, c.host).Inc()
metrics.LogsCollectedDuration.WithLabelValues(c.source).
Observe(time.Since(start).Seconds())
}
// parseLine 支持多种日志格式的解析
func (c *FileCollector) parseLine(line string) *LogEntry {
entry := &LogEntry{
Host: c.host,
Source: c.source,
Path: c.filePath,
Tags: c.tags,
}
// 尝试解析JSON格式日志（推荐）
// 很多现代日志库（如zap、logrus）都支持JSON输出
if strings.HasPrefix(line, "{") {
if err := json.Unmarshal([]byte(line), entry); err == nil {
return entry
}
}
// 尝试解析标准日志格式: 2024-01-15 10:30:00 [INFO] message
// 这是Go标准库log包的默认格式
parts := strings.SplitN(line, " ", 4)
if len(parts) >= 4 {
entry.Timestamp = parts[0] + " " + parts[1]
levelPart := strings.Trim(parts[2], "[]")
entry.Level = levelPart
entry.Message = parts[3]
} else {
// 无法解析格式，整体作为message
// 至少保证日志不丢
entry.Timestamp = time.Now().Format("2006-01-02 15:04:05")
entry.Level = "INFO"
entry.Message = line
}
return entry
}断点续传实现——这是Agent的关键功能，重启后从上次位置继续读取：
go
// OffsetStore 文件读取位置持久化
type OffsetStore struct {
path string
mu sync.Mutex
}
func NewOffsetStore(path string) *OffsetStore {
return &OffsetStore{path: path}
}
func (s *OffsetStore) Save(filePath string, offset int64) error {
s.mu.Lock()
defer s.mu.Unlock()
records := s.load()
records[filePath] = offset
data, err := json.Marshal(records)
if err != nil {
return err
}
// 原子写入：先写临时文件再rename
// 避免写一半进程挂了导致文件损坏
tmpPath := s.path + ".tmp"
if err := os.WriteFile(tmpPath, data, 0644); err != nil {
return err
}
return os.Rename(tmpPath, s.path)
}
func (s *OffsetStore) Load(filePath string) int64 {
records := s.load()
if offset, ok := records[filePath]; ok {
return offset
}
return 0
}
func (s *OffsetStore) load() map[string]int64 {
data, err := os.ReadFile(s.path)
if err != nil {
return make(map[string]int64)
}
var records map[string]int64
json.Unmarshal(data, &records)
if records == nil {
records = make(map[string]int64)
}
return records
}Agent主程序，支持多文件采集和优雅退出：
go
func main() {
// 加载配置
config := loadConfig("agent.yaml")
// 创建Kafka异步生产者
producer, err := createAsyncProducer(config.Kafka.Brokers)
if err != nil {
log.Fatalf("create producer failed: %v", err)
}
defer producer.Close()
// 创建offset存储
offsetStore := NewOffsetStore(config.Agent.OffsetFile)
// 启动Prometheus指标端点
go startMetricsServer(config.Metrics.Port)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// 为每个日志文件启动一个采集器
var wg sync.WaitGroup
for _, fileConfig := range config.Files {
collector := NewFileCollector(
fileConfig.Path,
fileConfig.Source,
config.Kafka.Topic,
producer,
offsetStore,
)
wg.Add(1)
go func() {
defer wg.Done()
if err := collector.Start(ctx); err != nil {
log.Printf("collector error: %v", err)
}
}()
}
// 优雅退出
sigCh := make(chan os.Signal, 1)
signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
<-sigCh
log.Println("Shutting down agent...")
cancel()
wg.Wait()
log.Println("Agent stopped")
}日志采集Agent的关键设计点：断点续传。Agent重启后需要从上次读取的位置继续，不能丢数据也不能重复。实现方式是定期把读取到的offset持久化到本地文件。怕浪猫踩过的坑：offset写太频繁导致磁盘IO高，写太少导致重启后重复采集多。5秒一次是经验值。
5.3 实现日志消费与存储
消费端从Kafka读取日志，批量写入Elasticsearch。批量写入是性能的关键——逐条写入ES的QPS只有几百，Bulk批量写入可以达到上万：
go
package consumer
import (
"bytes"
"context"
"encoding/json"
"fmt"
"log"
"strings"
"sync"
"time"
"github.com/IBM/sarama"
"github.com/elastic/go-elasticsearch/v8"
)
// LogConsumer 日志消费者
type LogConsumer struct {
esClient *elasticsearch.Client
esIndex string
batchSize int
batch []*sarama.ConsumerMessage
batchMu sync.Mutex
}
func NewLogConsumer(esURLs []string, esIndex string, batchSize int) *LogConsumer {
esClient, err := elasticsearch.NewClient(elasticsearch.Config{
URLs: esURLs,
Username: "elastic",
Password: "changeme",
})
if err != nil {
log.Fatalf("create ES client failed: %v", err)
}
c := &LogConsumer{
esClient: esClient,
esIndex: esIndex,
batchSize: batchSize,
}
// 初始化索引模板
c.ensureIndexTemplate()
return c
}
// ensureIndexTemplate 创建ES索引模板
// 模板会自动应用到匹配的索引名，统一mapping和settings
func (c *LogConsumer) ensureIndexTemplate() {
template := `{
"index_patterns": ["logs-*"],
"template": {
"settings": {
"number_of_shards": 3,
"number_of_replicas": 1,
"refresh_interval": "5s"
},
"mappings": {
"properties": {
"timestamp": {"type": "date"},
"level": {"type": "keyword"},
"source": {"type": "keyword"},
"host": {"type": "keyword"},
"path": {"type": "keyword"},
"message": {"type": "text", "analyzer": "ik_max_word"},
"tags": {"type": "object", "enabled": false}
}
}
}
}`
res, err := c.esClient.Indices.PutTemplate(
"logs-template", strings.NewReader(template))
if err != nil {
log.Printf("put index template failed: %v", err)
}
res.Body.Close()
}
func (c *LogConsumer) ConsumeClaim(session sarama.ConsumerGroupSession,
claim sarama.ConsumerGroupClaim) error {
c.batchMu.Lock()
c.batch = make([]*sarama.ConsumerMessage, 0, c.batchSize)
c.batchMu.Unlock()
ticker := time.NewTicker(2 * time.Second)
defer ticker.Stop()
for {
select {
case msg, ok := <-claim.Messages():
if !ok {
c.flush()
return nil
}
c.batchMu.Lock()
c.batch = append(c.batch, msg)
shouldFlush := len(c.batch) >= c.batchSize
c.batchMu.Unlock()
if shouldFlush {
c.flush()
}
session.MarkMessage(msg, "")
case <-ticker.C:\n\t\t\t// 定时flush，避免低流量时消息积压在batch里\n\t\t\tc.flush()\n\n\t\tcase <-session.Context().Done():
c.flush()
return nil
}
}
}
// flush 批量写入ES
func (c *LogConsumer) flush() {
c.batchMu.Lock()
if len(c.batch) == 0 {
c.batchMu.Unlock()
return
}
batch := c.batch
c.batch = make([]*sarama.ConsumerMessage, 0, c.batchSize)
c.batchMu.Unlock()
start := time.Now()
// 使用Bulk API批量写入ES
// Bulk API格式：每两行一组，第一行是操作指令，第二行是文档内容
var buf bytes.Buffer
for _, msg := range batch {
var entry LogEntry
if err := json.Unmarshal(msg.Value, &entry); err != nil {
log.Printf("unmarshal failed: %v", err)
continue
}
// 按日期生成索引名，每天一个索引
// 便于按时间范围查询和过期清理
indexName := fmt.Sprintf("logs-%s", time.Now().Format("2006.01.02"))
// Bulk action line（指定操作和目标索引）
meta := fmt.Sprintf(`{"index":{"_index":"%s"}}`, indexName)
buf.WriteString(meta)
buf.WriteByte('\n')
// 文档内容
data, _ := json.Marshal(entry)
buf.Write(data)
buf.WriteByte('\n')
}
if buf.Len() == 0 {
return
}
batchSize := buf.Len()
res, err := c.esClient.Bulk(bytes.NewReader(buf.Bytes()))
if err != nil {
log.Printf("ES bulk insert failed: %v", err)
return
}
defer res.Body.Close()
if res.IsError() {
log.Printf("ES bulk error: %s", res.String())
}
// 记录指标
duration := time.Since(start).Seconds()
metrics.ESBulkInsertDuration.WithLabelValues(c.esIndex).Observe(duration)
metrics.ESBulkInsertSize.WithLabelValues(c.esIndex).Observe(float64(batchSize))
metrics.LogsConsumedTotal.WithLabelValues(c.esIndex, "all").Add(float64(len(batch)))
}
func (c *LogConsumer) Setup(sarama.ConsumerGroupSession) error { return nil }
func (c *LogConsumer) Cleanup(sarama.ConsumerGroupSession) error { return nil }ES Bulk API是写入性能的关键。逐条写入ES的QPS大约在500左右，Bulk批量写入可以轻松达到10000以上。批大小建议500到1000条，太大会增加内存压力和GC停顿。怕浪猫在日志高峰期踩过batch太大导致OOM的坑，后来加了byte size限制才解决。
5.4 实现日志查询API
查询API基于gin框架，支持关键词搜索、时间范围过滤、日志级别过滤和分页。这是日志系统面向用户的核心接口：
go
package api
import (
"bytes"
"encoding/json"
"fmt"
"net/http"
"strconv"
"time"
"github.com/gin-gonic/gin"
"github.com/elastic/go-elasticsearch/v8"
)
type LogQueryAPI struct {
es *elasticsearch.Client
}
func NewLogQueryAPI(es *elasticsearch.Client) *LogQueryAPI {
return &LogQueryAPI{es: es}
}
func (a *LogQueryAPI) RegisterRoutes(r *gin.Engine) {
api := r.Group("/api/v1/logs")
{
api.GET("/search", a.Search) // 日志搜索
api.GET("/tail", a.Tail) // 实时日志流（SSE）
api.GET("/histogram", a.Histogram) // 日志分布统计
}
}
// Search 日志搜索
// GET /api/v1/logs/search?q=error&level=ERROR&source=order-service
// &start=2024-01-01T00:00:00Z&end=2024-01-02T00:00:00Z&page=1&size=20
func (a *LogQueryAPI) Search(c *gin.Context) {
page, _ := strconv.Atoi(c.DefaultQuery("page", "1"))
size, _ := strconv.Atoi(c.DefaultQuery("size", "20"))
if page < 1 {
page = 1
}
if size > 100 {
size = 100 // 限制最大返回条数，防止大结果集拖慢ES
}
from := (page - 1) * size
// 构建ES查询
query := a.buildSearchQuery(c, from, size)
// 搜索所有匹配的日志索引（按日期分索引的模式）
indexPattern := "logs-*"
res, err := a.es.Search(
a.es.Search.WithContext(c.Request.Context()),
a.es.Search.WithIndex(indexPattern),
a.es.Search.WithBody(query),
a.es.Search.WithPretty(),
)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
defer res.Body.Close()
var result map[string]interface{}
if err := json.NewDecoder(res.Body).Decode(&result); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
// 解析结果
hits := result["hits"].(map[string]interface{})["hits"].([]interface{})
total := result["hits"].(map[string]interface{})["total"].
(map[string]interface{})["value"].(float64)
logs := make([]map[string]interface{}, 0, len(hits))
for _, hit := range hits {
h := hit.(map[string]interface{})
source := h["_source"].(map[string]interface{})
source["_id"] = h["_id"]
source["_index"] = h["_index"]
logs = append(logs, source)
}
c.JSON(http.StatusOK, gin.H{
"total": int64(total),
"page": page,
"size": size,
"logs": logs,
})
}
// buildSearchQuery 构建ES查询体
func (a *LogQueryAPI) buildSearchQuery(c *gin.Context, from, size int) *bytes.Reader {
// 构建bool query
boolQuery := map[string]interface{}{
"query": map[string]interface{}{
"bool": map[string]interface{}{
"must": []interface{}{},
"filter": []interface{}{},
},
},
"from": from,
"size": size,
"sort": []map[string]interface{}{
{"timestamp": map[string]string{"order": "desc"}},
},
}
must := boolQuery["query"].(map[string]interface{})["bool"].
(map[string]interface{})["must"].([]interface{})
filter := boolQuery["query"].(map[string]interface{})["bool"].
(map[string]interface{})["filter"].([]interface{})
// 关键词搜索（全文检索）
if q := c.Query("q"); q != "" {
must = append(must, map[string]interface{}{
"match": map[string]interface{}{
"message": q,
},
})
}
// 日志级别过滤（精确匹配）
if level := c.Query("level"); level != "" {
filter = append(filter, map[string]interface{}{
"term": map[string]interface{}{"level": level},
})
}
// 来源过滤
if source := c.Query("source"); source != "" {
filter = append(filter, map[string]interface{}{
"term": map[string]interface{}{"source": source},
})
}
// 时间范围过滤
timeRange := map[string]interface{}{}
if start := c.Query("start"); start != "" {
timeRange["gte"] = start
}
if end := c.Query("end"); end != "" {
timeRange["lte"] = end
}
if len(timeRange) > 0 {
filter = append(filter, map[string]interface{}{
"range": map[string]interface{}{"timestamp": timeRange},
})
}
boolQuery["query"].(map[string]interface{})["bool"].
(map[string]interface{})["must"] = must
boolQuery["query"].(map[string]interface{})["bool"].
(map[string]interface{})["filter"] = filter
data, _ := json.Marshal(boolQuery)
return bytes.NewReader(data)
}
// Histogram 日志分布统计
// 返回按时间分桶的日志数量和各级别分布，用于绘制日志趋势图
func (a *LogQueryAPI) Histogram(c *gin.Context) {
interval := c.DefaultQuery("interval", "minute")
// 支持的interval: second, minute, hour, day
validIntervals := map[string]bool{
"second": true, "minute": true, "hour": true, "day": true,
}
if !validIntervals[interval] {
interval = "minute"
}
query := map[string]interface{}{
"size": 0, // 不返回文档，只要聚合结果
"query": map[string]interface{}{
"bool": map[string]interface{}{
"filter": a.buildTimeFilter(c),
},
},
"aggs": map[string]interface{}{
"log_over_time": map[string]interface{}{
"date_histogram": map[string]interface{}{
"field": "timestamp",
"calendar_interval": interval,
},
"aggs": map[string]interface{}{
"levels": map[string]interface{}{
"terms": map[string]interface{}{
"field": "level",
},
},
},
},
},
}
data, _ := json.Marshal(query)
res, err := a.es.Search(
a.es.Search.WithContext(c.Request.Context()),
a.es.Search.WithIndex("logs-*"),
a.es.Search.WithBody(bytes.NewReader(data)),
)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
defer res.Body.Close()
var result map[string]interface{}
json.NewDecoder(res.Body).Decode(&result)
c.JSON(http.StatusOK, result)
}
// Tail 实时日志流（Server-Sent Events）
// 前端通过EventSource连接，持续接收最新日志
func (a *LogQueryAPI) Tail(c *gin.Context) {
c.Header("Content-Type", "text/event-stream")
c.Header("Cache-Control", "no-cache")
c.Header("Connection", "keep-alive")
flusher, ok := c.Writer.(http.Flusher)
if !ok {
c.JSON(http.StatusInternalServerError,
gin.H{"error": "streaming not supported"})
return
}
ticker := time.NewTicker(1 * time.Second)
defer ticker.Stop()
var lastTimestamp time.Time
for {
select {
case <-c.Request.Context().Done():
return
case <-ticker.C:\n\t\t\tlogs := a.fetchLatestLogs(lastTimestamp)
for _, logEntry := range logs {
data, _ := json.Marshal(logEntry)
fmt.Fprintf(c.Writer, "data: %s\n\n", data)
flusher.Flush()
}
if len(logs) > 0 {
lastTimestamp = time.Now()
}
}
}
}
func (a *LogQueryAPI) fetchLatestLogs(after time.Time) []map[string]interface{} {
query := map[string]interface{}{
"size": 50,
"query": map[string]interface{}{
"range": map[string]interface{}{
"timestamp": map[string]interface{}{
"gt": after.Format(time.RFC3339Nano),
},
},
},
"sort": []map[string]interface{}{
{"timestamp": map[string]string{"order": "asc"}},
},
}
data, _ := json.Marshal(query)
res, err := a.es.Search(
a.es.Search.WithIndex("logs-*"),
a.es.Search.WithBody(bytes.NewReader(data)),
)
if err != nil {
return nil
}
defer res.Body.Close()
var result map[string]interface{}
json.NewDecoder(res.Body).Decode(&result)
hits, ok := result["hits"].(map[string]interface{})["hits"].([]interface{})
if !ok {
return nil
}
logs := make([]map[string]interface{}, 0, len(hits))
for _, hit := range hits {
h := hit.(map[string]interface{})
logs = append(logs, h["_source"].(map[string]interface{}))
}
return logs
}查询API的性能瓶颈通常在ES。记得给timestamp、level、source等常用过滤字段建好mapping和索引，避免动态mapping导致的类型推断错误。对于高频查询，加一层Redis缓存，命中率80%以上时查询延迟能降一个数量级。
5.5 实现告警引擎
告警引擎定期查询ES中的日志，匹配预设规则，触发后通过多种渠道通知。这是把"被动排查"变成"主动发现"的核心组件：
go
package alert
import (
"bytes"
"context"
"encoding/json"
"fmt"
"log"
"net/http"
"sync"
"time"
"github.com/elastic/go-elasticsearch/v8"
)
// Rule 告警规则定义
type Rule struct {
ID string `json:"id"`
Name string `json:"name"`
Description string `json:"description"`
Enabled bool `json:"enabled"`
// 查询条件
IndexPattern string `json:"indexPattern"`
Level string `json:"level"` // 日志级别过滤
Keyword string `json:"keyword"` // 关键词匹配
// 触发条件
Threshold int `json:"threshold"` // 匹配数量阈值
TimeWindow time.Duration `json:"timeWindow"` // 时间窗口
// 通知配置
Channels []Channel `json:"channels"`
// 运行时状态
lastFired time.Time
mu sync.Mutex
}
// Channel 通知渠道
type Channel struct {
Type string `json:"type"` // webhook, dingtalk, feishu, email
Config map[string]string `json:"config"`
}
// AlertEngine 告警引擎
type AlertEngine struct {
es *elasticsearch.Client
rules []*Rule
stopCh chan struct{}
aggregator *AlertAggregator // 告警聚合器
}
func NewAlertEngine(es *elasticsearch.Client) *AlertEngine {
return &AlertEngine{
es: es,
stopCh: make(chan struct{}),
aggregator: NewAlertAggregator(5 * time.Minute),
}
}
func (e *AlertEngine) AddRule(rule *Rule) {
e.rules = append(e.rules, rule)
}
func (e *AlertEngine) Start() {
log.Println("Alert engine started")
// 每30秒检查一次规则
// 间隔太短会给ES造成压力，太长会导致告警延迟
ticker := time.NewTicker(30 * time.Second)
defer ticker.Stop()
for {
select {
case <-ticker.C:\n\t\t\te.checkAllRules()\n\t\tcase <-e.stopCh:
log.Println("Alert engine stopped")
return
}
}
}
func (e *AlertEngine) Stop() {
close(e.stopCh)
}
// checkAllRules 并行检查所有规则
func (e *AlertEngine) checkAllRules() {
var wg sync.WaitGroup
for _, rule := range e.rules {
if !rule.Enabled {
continue
}
wg.Add(1)
go func(r *Rule) {
defer wg.Done()
e.checkRule(r)
}(rule)
}
wg.Wait()
}
func (e *AlertEngine) checkRule(rule *Rule) {
rule.mu.Lock()
defer rule.mu.Unlock()
// 检查冷却期，避免频繁告警
// 同一规则在TimeWindow内只触发一次
if time.Since(rule.lastFired) < rule.TimeWindow {
return
}
// 构建ES查询，统计时间窗口内匹配的日志数量
query := e.buildRuleQuery(rule)
res, err := e.es.Search(
e.es.Search.WithIndex(rule.IndexPattern),
e.es.Search.WithBody(query),
e.es.Search.WithSize(0), // 只需要count，不需要文档
)
if err != nil {
log.Printf("rule %s query failed: %v", rule.ID, err)
return
}
defer res.Body.Close()
var result map[string]interface{}
json.NewDecoder(res.Body).Decode(&result)
total, ok := result["hits"].(map[string]interface{})["total"].
(map[string]interface{})["value"].(float64)
if !ok {
return
}
// 超过阈值，触发告警
if int(total) >= rule.Threshold {
alert := Alert{
RuleName: rule.Name,
Description: rule.Description,
Count: int(total),
Threshold: rule.Threshold,
TimeWindow: rule.TimeWindow.String(),
FiredAt: time.Now(),
}
// 获取匹配的样本日志，附在告警里方便排查
samples := e.getSampleLogs(rule, 5)
alert.Samples = samples
// 告警聚合检查：避免告警风暴
shouldSend, state := e.aggregator.ShouldSend(rule.ID, rule.TimeWindow)
if shouldSend {
alert.AggregatedCount = state.Count
// 发送通知到所有配置的渠道
for _, channel := range rule.Channels {
if err := e.sendNotification(channel, alert); err != nil {
metrics.AlertNotificationErrors.WithLabelValues(
rule.ID, channel.Type).Inc()
log.Printf("send notification failed: %v", err)
}
}
metrics.AlertsFiredTotal.WithLabelValues(
rule.ID, rule.Name).Inc()
}
rule.lastFired = time.Now()
log.Printf("Alert fired: %s (count=%d, threshold=%d, aggregated=%d)",
rule.Name, int(total), rule.Threshold, state.Count)
}
}
func (e *AlertEngine) buildRuleQuery(rule *Rule) *bytes.Reader {
boolQuery := map[string]interface{}{
"query": map[string]interface{}{
"bool": map[string]interface{}{
"filter": []interface{}{
map[string]interface{}{
"range": map[string]interface{}{
"timestamp": map[string]interface{}{
"gte": fmt.Sprintf("now-%ds",
int(rule.TimeWindow.Seconds())),
},
},
},
},
},
},
"size": 0,
}
filters := boolQuery["query"].(map[string]interface{})["bool"].
(map[string]interface{})["filter"].([]interface{})
if rule.Level != "" {
filters = append(filters, map[string]interface{}{
"term": map[string]interface{}{"level": rule.Level},
})
}
if rule.Keyword != "" {
filters = append(filters, map[string]interface{}{
"match": map[string]interface{}{"message": rule.Keyword},
})
}
boolQuery["query"].(map[string]interface{})["bool"].
(map[string]interface{})["filter"] = filters
data, _ := json.Marshal(boolQuery)
return bytes.NewReader(data)
}
// Alert 告警事件
type Alert struct {
RuleName string `json:"ruleName"`
Description string `json:"description"`
Count int `json:"count"`
Threshold int `json:"threshold"`
TimeWindow string `json:"timeWindow"`
FiredAt time.Time `json:"firedAt"`
Samples []map[string]interface{} `json:"samples"`
AggregatedCount int `json:"aggregatedCount"`
}
// sendNotification 根据渠道类型发送通知
func (e *AlertEngine) sendNotification(channel Channel, alert Alert) error {
switch channel.Type {
case "dingtalk":
return e.sendDingTalk(channel.Config["webhook"], alert)
case "feishu":
return e.sendFeishu(channel.Config["webhook"], alert)
case "webhook":
return e.sendWebhook(channel.Config["url"], alert)
case "email":
return e.sendEmail(channel.Config["to"], alert)
default:
return fmt.Errorf("unknown channel type: %s", channel.Type)
}
}
// sendDingTalk 钉钉机器人通知
func (e *AlertEngine) sendDingTalk(webhook string, alert Alert) error {
message := fmt.Sprintf("【告警】%s\n\n描述: %s\n触发条件: %d次/%s\n当前值: %d次\n"+
"聚合次数: %d\n触发时间: %s\n\n请尽快处理！",
alert.RuleName,
alert.Description,
alert.Threshold,
alert.TimeWindow,
alert.Count,
alert.AggregatedCount,
alert.FiredAt.Format("2006-01-02 15:04:05"),
)
payload := map[string]interface{}{
"msgtype": "text",
"text": map[string]string{"content": message},
"at": map[string]interface{}{
"isAtAll": true, // @所有人，生产环境建议按需@值班人
},
}
data, _ := json.Marshal(payload)
resp, err := http.Post(webhook, "application/json", bytes.NewReader(data))
if err != nil {
return err
}
defer resp.Body.Close()
return nil
}
// sendFeishu 飞书机器人通知（富文本卡片）
func (e *AlertEngine) sendFeishu(webhook string, alert Alert) error {
card := map[string]interface{}{
"msg_type": "interactive",
"card": map[string]interface{}{
"header": map[string]interface{}{
"title": map[string]interface{}{
"tag": "plain_text",
"content": fmt.Sprintf("告警: %s", alert.RuleName),
},
"template": "red", // 红色卡片表示严重告警
},
"elements": []interface{}{
map[string]interface{}{
"tag": "div",
"text": map[string]interface{}{
"tag": "lark_md",
"content": fmt.Sprintf(
"**描述**: %s\n**阈值**: %d次/%s\n**实际**: %d次\n"+
"**聚合**: %d次\n**时间**: %s",
alert.Description,
alert.Threshold,
alert.TimeWindow,
alert.Count,
alert.AggregatedCount,
alert.FiredAt.Format("2006-01-02 15:04:05"),
),
},
},
// 如果有样本日志，展示前几条
map[string]interface{}{
"tag": "div",
"text": map[string]interface{}{
"tag": "lark_md",
"content": fmt.Sprintf("**样本日志**: 共%d条，详情见附件",
len(alert.Samples)),
},
},
},
},
}
data, _ := json.Marshal(card)
resp, err := http.Post(webhook, "application/json", bytes.NewReader(data))
if err != nil {
return err
}
defer resp.Body.Close()
return nil
}
func (e *AlertEngine) sendWebhook(url string, alert Alert) error {
data, _ := json.Marshal(alert)
resp, err := http.Post(url, "application/json", bytes.NewReader(data))
if err != nil {
return err
}
defer resp.Body.Close()
return nil
}
func (e *AlertEngine) sendEmail(to string, alert Alert) error {
// 邮件通知实现省略，可以用gomail等库
return nil
}
// getSampleLogs 获取匹配规则的样本日志
func (e *AlertEngine) getSampleLogs(rule *Rule, size int) []map[string]interface{} {
query := e.buildRuleQuery(rule)
var m map[string]interface{}
json.NewDecoder(query).Decode(&m)
m["size"] = size
m["sort"] = []map[string]interface{}{
{"timestamp": map[string]string{"order": "desc"}},
}
data, _ := json.Marshal(m)
res, err := e.es.Search(
e.es.Search.WithIndex(rule.IndexPattern),
e.es.Search.WithBody(bytes.NewReader(data)),
)
if err != nil {
return nil
}
defer res.Body.Close()
var result map[string]interface{}
json.NewDecoder(res.Body).Decode(&result)
hits, ok := result["hits"].(map[string]interface{})["hits"].([]interface{})
if !ok {
return nil
}
samples := make([]map[string]interface{}, 0, len(hits))
for _, hit := range hits {
h := hit.(map[string]interface{})
samples = append(samples, h["_source"].(map[string]interface{}))
}
return samples
}
// AlertAggregator 告警聚合器
// 解决告警风暴问题：同一规则在冷却期内只发一次通知
type AlertAggregator struct {
cache map[string]*AlertState
mu sync.Mutex
ttl time.Duration
}
type AlertState struct {
FirstFired time.Time
Count int
LastSent time.Time
}
func NewAlertAggregator(ttl time.Duration) *AlertAggregator {
return &AlertAggregator{
cache: make(map[string]*AlertState),
ttl: ttl,
}
}
func (a *AlertAggregator) ShouldSend(ruleID string,
cooldown time.Duration) (bool, *AlertState) {
a.mu.Lock()
defer a.mu.Unlock()
state, exists := a.cache[ruleID]
if !exists {
a.cache[ruleID] = &AlertState{
FirstFired: time.Now(),
Count: 1,
LastSent: time.Now(),
}
return true, a.cache[ruleID]
}
state.Count++
if time.Since(state.LastSent) >= cooldown {
state.LastSent = time.Now()
return true, state
}
return false, state
}告警规则配置示例：
go
func setupAlertRules(engine *AlertEngine) {
// 规则1: ERROR日志5分钟内超过10条
// 适用于所有服务，可能是某个服务出了问题
engine.AddRule(&Rule{
ID: "error-spike",
Name: "ERROR日志激增",
Description: "5分钟内ERROR级别日志超过10条，可能存在服务异常",
Enabled: true,
IndexPattern: "logs-*",
Level: "ERROR",
Threshold: 10,
TimeWindow: 5 * time.Minute,
Channels: []Channel{
{
Type: "dingtalk",
Config: map[string]string{
"webhook": "https://oapi.dingtalk.com/robot/send?access_token=xxx",
},
},
},
})
// 规则2: panic关键词出现立即告警
// panic意味着程序崩溃，需要立即处理
engine.AddRule(&Rule{
ID: "panic-detected",
Name: "Panic检测",
Description: "检测到panic关键字，程序可能崩溃",
Enabled: true,
IndexPattern: "logs-*",
Keyword: "panic",
Threshold: 1,
TimeWindow: 1 * time.Minute,
Channels: []Channel{
{Type: "feishu", Config: map[string]string{
"webhook": "https://open.feishu.cn/open-apis/bot/v2/hook/xxx"}},
{Type: "webhook", Config: map[string]string{
"url": "https://internal.example.com/alert/panic"}},
},
})
// 规则3: 某服务连接数据库失败
engine.AddRule(&Rule{
ID: "db-connection-failed",
Name: "数据库连接失败",
Description: "order-service连接数据库失败，可能影响下单",
Enabled: true,
IndexPattern: "logs-*",
Keyword: "database connection refused",
Threshold: 3,
TimeWindow: 2 * time.Minute,
Channels: []Channel{
{Type: "dingtalk", Config: map[string]string{
"webhook": "https://oapi.dingtalk.com/robot/send?access_token=xxx"}},
{Type: "feishu", Config: map[string]string{
"webhook": "https://open.feishu.cn/open-apis/bot/v2/hook/xxx"}},
},
})
}告警引擎最大的敌人是告警风暴。一条服务抖动可能在1分钟内触发上百条告警。必须设置告警冷却期和聚合策略，否则告警本身就会成为故障——你会花时间处理告警通知而不是处理故障。
5.6 实现监控大盘：Prometheus + Grafana集成
日志系统自身的运行状态也需要监控。你要知道Agent有没有在采集、Consumer的Lag有多大、ES写入延迟是否正常。这需要暴露指标到Prometheus，再用Grafana可视化。
首先定义指标——每个指标都有明确的含义和标签维度：
go
package metrics
import (
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
)
var (
// ===== Agent指标 =====
LogsCollectedTotal = promauto.NewCounterVec(
prometheus.CounterOpts{
Name: "log_agent_collected_total",
Help: "Total number of log lines collected",
},
[]string{"source", "host"},
)
LogsCollectErrors = promauto.NewCounterVec(
prometheus.CounterOpts{
Name: "log_agent_collect_errors_total",
Help: "Total number of collection errors",
},
[]string{"source", "host", "error_type"},
)
LogsCollectedDuration = promauto.NewHistogramVec(
prometheus.HistogramOpts{
Name: "log_agent_collect_duration_seconds",
Help: "Time spent collecting and sending logs",
Buckets: prometheus.ExponentialBuckets(0.001, 2, 10),
},
[]string{"source"},
)
// ===== Consumer指标 =====
LogsConsumedTotal = promauto.NewCounterVec(
prometheus.CounterOpts{
Name: "log_consumer_consumed_total",
Help: "Total number of log entries consumed from Kafka",
},
[]string{"topic", "partition"},
)
LogsConsumedErrors = promauto.NewCounterVec(
prometheus.CounterOpts{
Name: "log_consumer_errors_total",
Help: "Total number of consumption errors",
},
[]string{"topic", "error_type"},
)
// ConsumerLag是最关键的运维指标
// Lag持续增长说明消费速度跟不上生产速度
ConsumerLag = promauto.NewGaugeVec(
prometheus.GaugeOpts{
Name: "log_consumer_lag",
Help: "Consumer lag (number of messages behind)",
},
[]string{"topic", "partition", "consumer_group"},
)
// ===== ES写入指标 =====
ESBulkInsertDuration = promauto.NewHistogramVec(
prometheus.HistogramOpts{
Name: "es_bulk_insert_duration_seconds",
Help: "Time spent on ES bulk insert",
Buckets: prometheus.ExponentialBuckets(0.01, 2, 10),
},
[]string{"index"},
)
ESBulkInsertSize = promauto.NewHistogramVec(
prometheus.HistogramOpts{
Name: "es_bulk_insert_size_bytes",
Help: "Size of ES bulk insert payload",
Buckets: prometheus.ExponentialBuckets(100, 2, 15),
},
[]string{"index"},
)
// ===== 告警引擎指标 =====
AlertsFiredTotal = promauto.NewCounterVec(
prometheus.CounterOpts{
Name: "alert_engine_fired_total",
Help: "Total number of alerts fired",
},
[]string{"rule_id", "rule_name"},
)
AlertNotificationErrors = promauto.NewCounterVec(
prometheus.CounterOpts{
Name: "alert_notification_errors_total",
Help: "Total number of notification errors",
},
[]string{"rule_id", "channel_type"},
)
)Consumer Lag监控——这是Kafka运维最重要的指标。Lag持续增长说明消费速度跟不上生产速度，最终会导致消息延迟越来越大：
go
func MonitorConsumerLag(client sarama.Client, topic, groupID string) {
ticker := time.NewTicker(30 * time.Second)
defer ticker.Stop()
for range ticker.C {
partitions, err := client.Partitions(topic)
if err != nil {
continue
}
for _, partition := range partitions {
// 获取分区最新offset（生产者写入的位置）
latestOffset, err := client.GetOffset(topic, partition, sarama.OffsetNewest)
if err != nil {
continue
}
// 获取消费者组当前提交的offset
offsetManager, err := sarama.NewOffsetManagerFromClient(groupID, client)
if err != nil {
continue
}
partitionOffsetManager, err := offsetManager.ManagePartition(topic, partition)
if err != nil {
offsetManager.Close()
continue
}
consumedOffset, _ := partitionOffsetManager.NextOffset()
partitionOffsetManager.Close()
offsetManager.Close()
lag := latestOffset - consumedOffset
if lag < 0 {
lag = 0
}
// 设置Gauge指标
ConsumerLag.WithLabelValues(
topic,
fmt.Sprintf("%d", partition),
groupID,
).Set(float64(lag))
}
}
}暴露Prometheus指标端点：
go
package main
import (
"net/http"
"github.com/prometheus/client_golang/prometheus/promhttp"
)
func startMetricsServer(port string) {
mux := http.NewServeMux()
mux.Handle("/metrics", promhttp.Handler())
go func() {
log.Printf("Metrics server listening on :%s", port)
if err := http.ListenAndServe(":"+port, mux); err != nil {
log.Printf("metrics server error: %v", err)
}
}()
}Grafana仪表盘配置，以下JSON可以直接导入Grafana：
json
{
"dashboard": {
"title": "日志监控系统大盘",
"tags": ["logging", "kafka", "elasticsearch"],
"timezone": "browser",
"panels": [
{
"title": "日志采集速率 (lines/s)",
"type": "graph",
"datasource": "Prometheus",
"gridPos": {"h": 8, "w": 12, "x": 0, "y": 0},
"targets": [{
"expr": "rate(log_agent_collected_total[1m])",
"legendFormat": "{{source}} - {{host}}"
}],
"yAxes": [{"label": "lines/s"}]
},
{
"title": "Kafka消费Lag",
"type": "graph",
"datasource": "Prometheus",
"gridPos": {"h": 8, "w": 12, "x": 12, "y": 0},
"targets": [{
"expr": "log_consumer_lag",
"legendFormat": "partition {{partition}}"
}],
"alert": {
"name": "Consumer Lag High",
"conditions": [{
"type": "query",
"query": {"params": ["A", "5m", "now"]},
"evaluator": {"params": [10000], "type": "gt"},
"operator": {"type": "and"}
}]
}
},
{
"title": "ES写入延迟",
"type": "graph",
"datasource": "Prometheus",
"gridPos": {"h": 8, "w": 12, "x": 0, "y": 8},
"targets": [{
"expr": "rate(es_bulk_insert_duration_seconds_sum[5m]) / rate(es_bulk_insert_duration_seconds_count[5m])",
"legendFormat": "avg duration"
}]
},
{
"title": "告警触发次数 (1h)",
"type": "stat",
"datasource": "Prometheus",
"gridPos": {"h": 4, "w": 6, "x": 12, "y": 8},
"targets": [{
"expr": "increase(alert_engine_fired_total[1h])",
"legendFormat": "{{rule_name}}"
}],
"fieldConfig": {
"defaults": {"thresholds": {
"mode": "absolute",
"steps": [
{"color": "green", "value": null},
{"color": "yellow", "value": 1},
{"color": "red", "value": 10}
]
}}
}
},
{
"title": "采集错误率",
"type": "graph",
"datasource": "Prometheus",
"gridPos": {"h": 8, "w": 12, "x": 0, "y": 16},
"targets": [{
"expr": "rate(log_agent_collect_errors_total[5m])",
"legendFormat": "{{source}} - {{error_type}}"
}]
}
],
"time": {"from": "now-6h", "to": "now"},
"refresh": "30s"
}
}监控大盘的价值不在于好看，在于缩短MTTR（平均故障恢复时间）。当告警来了，你第一眼看到的就是哪个环节出了问题——是Agent没采集、Kafka积压了、还是ES写入慢了——而不是满世界查日志。
5.7 系统部署与运维
部署架构推荐使用Docker Compose，开发和测试环境一键启动：
yaml
version: '3.8'
services:
kafka:
image: bitnami/kafka:3.5
ports:
- "9092:9092"
environment:
KAFKA_CFG_NODE_ID: 0
KAFKA_CFG_PROCESS_ROLES: controller,broker
KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://127.0.0.1:9092
KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 0@kafka:9093
KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
volumes:
- kafka_data:/bitnami/kafka
elasticsearch:
image: docker.elastic.co/elasticsearch/elasticsearch:8.11.0
ports:
- "9200:9200"
environment:
- discovery.type=single-node
- xpack.security.enabled=false
- "ES_JAVA_OPTS=-Xms512m -Xmx512m"
volumes:
- es_data:/usr/share/elasticsearch/data
kibana:
image: docker.elastic.co/kibana/kibana:8.11.0
ports:
- "5601:5601"
environment:
- ELASTICSEARCH_HOSTS=http://elasticsearch:9200
depends_on:
- elasticsearch
prometheus:
image: prom/prometheus:latest
ports:
- "9090:9090"
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
grafana:
image: grafana/grafana:latest
ports:
- "3000:3000"
environment:
- GF_SECURITY_ADMIN_PASSWORD=admin
depends_on:
- prometheus
volumes:
kafka_data:
es_data:Prometheus配置文件：
yaml
global:
scrape_interval: 15s
scrape_configs:
- job_name: 'log-agent'
static_configs:
- targets: ['host.docker.internal:2112']
labels:
service: 'log-agent'
- job_name: 'log-consumer'
static_configs:
- targets: ['host.docker.internal:2113']
labels:
service: 'log-consumer'
- job_name: 'alert-engine'
static_configs:
- targets: ['host.docker.internal:2114']
labels:
service: 'alert-engine'5.8 压测与优化
系统搭建完成后，怕浪猫做了一轮压测。压测不是可选项而是必选项——你以为系统能扛10万QPS，实际可能只扛了2万。生产环境的流量峰值永远比你预想的猛，不做压测就上线等于裸奔。压测要模拟真实场景：真实的日志格式、真实的流量曲线、真实的查询并发。
压测环境：3台4C8G机器，Kafka 3节点，ES单节点。这个配置接近生产环境的中等规模，压测结果有参考价值。
压测方法：用Go写了一个日志生成器，按不同速率向日志文件写入模拟日志（包含JSON格式和纯文本格式），同时模拟多用户并发查询。压测持续30分钟，观察各组件的CPU、内存、IO和网络指标。
压测结果：
| 组件 | 指标 | 数值 | 备注 |
|---|---|---|---|
| Agent采集 | 单机吞吐 | 15,000 lines/s | 使用inotify |
| Agent采集 | CPU占用 | 约15% (1核) | 含序列化 |
| Agent采集 | 内存占用 | 约80MB | 含Kafka缓冲 |
| Kafka | 生产TPS | 50,000 msg/s | LZ4压缩 |
| Kafka | 消费TPS | 40,000 msg/s | 批量消费 |
| ES写入 | Bulk QPS | 12,000 docs/s | batch=500 |
| ES查询 | 关键词搜索 | 50ms | 100万文档 |
| ES查询 | 范围+聚合 | 200ms | 100万文档 |
| 告警引擎 | 规则检查延迟 | 小于2s | 10条规则 |
优化经验清单——这些是压测过程中发现并解决的问题：
Agent侧优化：
- 文件tail使用inotify而非轮询，CPU降低80%。轮询模式下CPU占用高达60%，换成inotify后降到15%
- Kafka生产者开启LZ4压缩，网络带宽减少60%。在千兆网卡环境下，无压缩时网络是瓶颈，开启压缩后网络利用率从95%降到35%
- 批量发送间隔设为200ms，吞吐量提升5倍。逐条发送时TPS只有3000，批量后达到15000
- 异步生产者替代同步生产者，发送不再阻塞采集。同步模式下采集速度受限于Kafka确认速度
Consumer侧优化：
- 批量消费100条一批，ES写入TPS提升12倍。逐条写入ES时TPS只有1000，Bulk批量后达到12000
- ES refresh_interval从1s改为5s，写入性能提升30%。ES默认1秒刷新一次索引让文档可搜索，但日志场景不需要这么高的实时性
- 消费者goroutine数等于分区数，最大化并行度。多出的goroutine会空闲，少了则浪费分区并行能力
- ES索引按天分，查询时按时间范围路由，减少扫描量。查询7天数据只扫7个索引而不是一个巨大的索引
查询侧优化：
- 热查询加Redis缓存，命中率80%时查询延迟降90%。常见查询如"最近1小时ERROR日志"可以缓存30秒
- 对message字段使用ik_max_word分词器，中文搜索效果更好。默认分词器对中文支持差，搜"订单"搜不到"下单流程"
- 限制单次查询最大返回100条，防止大结果集拖慢ES。用户要看更多数据用分页，不要一次查几千条
压测不是可选项。你以为系统能扛10万QPS，实际可能只扛了2万。生产环境的流量峰值永远比你预想的猛，不做压测就上线等于裸奔。
六、踩坑总结
这一章的项目，怕浪猫在生产环境实际跑过两年。两年里大大小小的故障出过十几次，每次都让人刻骨铭心。这里挑几个最有代表性的坑分享出来，每个都是用故障换来的教训。希望你看完之后能在自己的项目中提前规避，不要等生产事故来教你。
坑1：Kafka消费者Rebalance导致重复消费
场景：消费者处理消息耗时较长，超过session.timeout，触发Rebalance，分区被重新分配。新消费者从上次提交的offset开始消费，导致一批消息被重复处理。某次导致2000个订单被重复处理，花了两天时间修数据。当时消费者在处理消息时调用了外部库存接口，接口响应慢，单条消息处理时间超过了30秒的session timeout。
根因分析：Sarama的消费者组通过心跳维持成员关系。如果消费者处理消息时阻塞太久，没有及时发送心跳，Coordinator会认为这个消费者挂了，触发Rebalance。Rebalance后分区被重新分配，新的消费者从上次提交的offset开始消费，但原来的消费者可能还在处理之前的消息，导致重复处理。
解法有三个层面：
- 调大session timeout，给消费者更多处理时间
- 使用CooperativeRebalanceStrategy，减少Rebalance影响范围
- 把耗时操作异步化，消息处理只做入队，后台worker慢慢处理
go
// 使用Cooperative策略，只rebalance变化的分区
// 传统的Eager策略会暂停所有消费者，Cooperative只暂停受影响的分区
config.Consumer.Group.Rebalance.GroupStrategies = []sarama.BalanceStrategy{
sarama.NewBalanceStrategyCooperativeSticky(),
}
config.Consumer.Group.Session.Timeout = 30 * time.Second
// 处理消息的goroutine要控制并发，避免处理时间超过session timeout坑2：ES Bulk写入OOM
场景：批量写入时batch size设太大（10000条），单条日志平均2KB，一次Bulk请求20MB。高流量时段内存暴涨触发OOM，消费者进程被kill。更麻烦的是Kafka消费offset已经提交了，但ES没写进去，这批日志彻底丢了。
根因分析：batch size只限制了条数没限制字节数。如果日志内容很长（比如包含完整请求和响应体），单条日志可能几十KB，10000条就是几百MB。Go的GC在这种大对象分配上表现不好，容易导致内存暴涨。
解法：限制batch的byte size而非只限制count；使用背压机制，ES写入慢时减慢消费速度；对日志内容做截断，单条日志超过10KB只保留前10KB。
go
const maxBatchBytes = 5 * 1024 * 1024 // 5MB上限
func (c *LogConsumer) flush() {
// 检查batch总大小，超过上限则分批写入
totalBytes := 0
for _, msg := range c.batch {
totalBytes += len(msg.Value)
}
if totalBytes > maxBatchBytes {
// 分批写入，每批不超过maxBatchBytes
c.flushInChunks(maxBatchBytes)
return
}
c.flushAll()
}坑3：告警风暴
场景：一个核心服务异常，ERROR日志暴增，30秒内触发了200多条告警，钉钉群被刷屏，反而看不到关键信息。值班同学直接把钉钉群静音了，结果后面的真正重要的告警也被淹没了。这个问题的影响比日志积压还严重——积压至少能发现，告警被静音后你可能完全不知道出了问题。
根因分析：告警引擎每30秒检查一次规则，每次检查都发通知。没有聚合和冷却机制，导致同一规则的告警反复发送。
解法：告警聚合加冷却期加分级路由。同一规则的告警5分钟内只发一次，附带聚合统计（"过去5分钟触发了200次"）；P0级别走电话通知，P1走钉钉，P2走邮件。不同级别的告警走不同渠道，确保重要告警不被淹没。
坑4：日志文件被rotate时丢数据
场景：应用使用logrotate切割日志，Agent的tail在文件被rotate时丢失了最后的几行。日志量少的时候没发现，日志量大时每次rotate都丢几百行。某次排查线上问题发现刚好缺了故障发生时间段的日志，因为那个时间点logrotate正好在切割日志。
根因分析：logrotate默认使用create模式——先重命名旧文件，再创建新文件。tail库跟踪的是文件描述符，文件被重命名后，tail继续读旧文件（已重命名），新文件不会被自动跟踪。虽然tail库配置了ReOpen=true，但在检测到文件变化和重新打开之间有一个短暂的窗口期，这期间写入的日志可能丢失。
解法：配置logrotate使用copytruncate模式——先复制旧文件，再清空原文件。原文件的inode不变，tail不受影响。代价是复制期间有一次完整的文件拷贝IO。另一个方案是Agent同时监听新旧文件名，但这需要Agent感知logrotate的命名规则。
坑5：Sarama AsyncProducer死锁
场景：业务高峰期AsyncProducer突然不再发送消息，程序不报错也不退出。排查发现Successes channel满了（配置了Return.Successes=true但没消费Successes channel），导致Input channel也满了，所有发送操作阻塞。更隐蔽的是，这个阻塞不会触发任何错误或告警，程序看起来正常运行但其实已经停止工作。
根因分析：Sarama的AsyncProducer内部使用channel传递消息。Input channel写入消息，Successes和Errors channel返回结果。如果配置了Return.Successes=true但没有goroutine消费Successes channel，当内部缓冲区满后，Input channel的写入会阻塞，进而导致所有调用发送的代码阻塞。
解法：如果不需要逐条确认成功，就设Return.Successes=false（默认就是false）；如果需要成功回调，必须用goroutine持续消费Successes channel。生产环境的最佳实践是无论是否需要成功回调，都启动goroutine消费Successes和Errors channel，前者用于更新指标，后者用于错误处理和重试。
每一个坑都是用故障换来的经验。你在设计阶段多想一步，生产环境就少一次半夜被叫起来。怕浪猫的建议：上面这些坑，在设计阶段就提前规避，不要等生产事故来教你。最好的故障是别人的故障，第二好的是已经预防的故障，最差的是自己亲身经历的故障。
七、完整的系统组装
最后，把所有组件组装到一起。main函数就是整个系统的入口，负责初始化所有组件并管理生命周期。实际生产中，Agent、Consumer、API、AlertEngine可以部署在同一进程中（适合小规模部署），也可以拆分成独立服务（适合大规模部署）。这里展示的是单进程模式，方便理解各组件之间的关系：
go
package main
import (
"context"
"log"
"os"
"os/signal"
"syscall"
"github.com/IBM/sarama"
"github.com/elastic/go-elasticsearch/v8"
)
func main() {
// 加载配置
config := LoadConfig("config.yaml")
log.Printf("Starting log system with config: %+v", config)
// 启动Prometheus指标端点
// 每个组件都通过这个端点暴露自己的运行指标
go startMetricsServer(config.Metrics.Port)
// 创建Kafka异步生产者
// Agent用这个生产者把日志发送到Kafka
producer := createAsyncProducer(config.Kafka.Brokers)
defer producer.Close()
// 创建ES客户端
// Consumer和AlertEngine都需要操作ES
esClient, _ := elasticsearch.NewClient(elasticsearch.Config{
URLs: config.ES.URLs,
})
// 创建offset存储
// Agent用它在本地持久化文件读取位置，支持断点续传
offsetStore := NewOffsetStore(config.Agent.OffsetFile)
// 启动日志采集Agent
// 为每个配置的日志文件启动一个采集goroutine
ctx, cancel := context.WithCancel(context.Background())
for _, fileCfg := range config.Agent.Files {
collector := NewFileCollector(
fileCfg.Path, fileCfg.Source,
config.Kafka.Topic, producer, offsetStore,
)
go collector.Start(ctx)
}
// 启动Kafka消费者，消费日志写入ES
// 消费者使用消费者组模式，支持多实例水平扩展
go func() {
consumer := NewLogConsumer(
config.ES.URLs, "logs", config.Consumer.BatchSize,
)
StartConsumerGroup(
config.Kafka.Brokers,
config.Kafka.Topic,
config.Kafka.ConsumerGroup,
consumer,
)
}()
// 启动告警引擎
// 独立goroutine运行，定期查询ES检查告警规则
alertEngine := NewAlertEngine(esClient)
setupAlertRules(alertEngine)
go alertEngine.Start()
// 启动查询API
// 对外提供日志搜索、实时流、统计聚合接口
go startQueryAPI(esClient, config.API.Port)
// 启动Consumer Lag监控
// 这是最重要的运维指标，Lag持续增长意味着消费速度跟不上生产速度
kafkaClient := createKafkaClient(config.Kafka.Brokers)
go MonitorConsumerLag(kafkaClient, config.Kafka.Topic,
config.Kafka.ConsumerGroup)
log.Println("Log system fully started")
// 优雅退出
// 收到SIGINT或SIGTERM后，先停止采集，再等消费者处理完积压消息
// 最后关闭生产者和ES连接
sigCh := make(chan os.Signal, 1)
signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
<-sigCh
log.Println("Shutting down...")
cancel() // 停止所有采集器
alertEngine.Stop() // 停止告警引擎
producer.Close() // 关闭Kafka生产者
log.Println("Log system stopped")
}部署这个系统时，建议把配置文件外置，支持不同环境使用不同配置。配置文件应该包含Kafka地址、ES地址、监听的日志文件路径、告警规则等。生产环境建议用配置中心（如Nacos、Apollo）管理配置，避免改配置就要重新部署。
写在最后
这一章我们从Kafka客户端使用讲到消息可靠性保障，从延迟消息讲到日志监控系统完整实现。内容量很大，代码很多，但每一行都是生产验证过的。
怕浪猫想说，消息队列和可观测性不是两个独立的东西。当你把日志采集、Kafka传输、ES存储、告警引擎这条链路打通后，你的系统就有了"自我感知"能力。出问题之前告警先到，出问题之后三分钟定位根因——这就是可观测性的价值，也是这一章整个项目的意义。
如果你正在搭建类似系统，建议按这个顺序来：先把Kafka跑通，再把Agent和Consumer跑通，最后接ES和告警。每一步都压测确认性能达标再进入下一步。不要一口气全搭起来再调试，那样出问题都不知道是哪个环节的。
下一章我们进入工程化与CI/CD，聊聊怎么把代码从开发环境安全可靠地送到生产环境。包括多环境配置管理、Docker化、CI流水线、CD策略、灰度发布等内容。
如果这篇文章帮到了你，点个收藏，以后排查问题能快速翻出来。代码量这么大，总有用得上的时候。
你在消息队列和日志监控方面踩过什么坑？评论区聊聊，怕浪猫在线答疑。
这是Go实战手册第14章，系列进度 14/16。关注我，下一章「工程化与CI/CD」更精彩。
怕浪猫说：消息队列是系统的血管，日志监控是系统的神经。血管堵了系统会停，神经断了你会变成瞎子——出了事还在傻笑。把这两套基础设施建好，你的系统才真正具备生产级生存能力。