go语言并发实战——日志收集系统(三) 利用sarama包连接KafKa实现消息的生产与消费

本文涉及的产品
日志服务 SLS,月写入数据量 50GB 1个月
简介: go语言并发实战——日志收集系统(三) 利用sarama包连接KafKa实现消息的生产与消费

环境的搭建

Kafka以及相关组件的下载

我们要实现今天的内容,不可避免的要进行对开发环境的配置,Kafka环境的配置比较繁琐,需要配置JDK,Scala,ZoopKeeper和Kafka,这里我们不做赘述,如果大家不知道如何配置环境,这里我们个大家找了一篇博文供大家参考:


sarama包的安装

今天我们所时机的内容需要用到go语言的第三方包sarama,由于1.19版本后添加了ztcd压缩算法,需要用到cgo,这里我们为了方便考虑选择下载sarama v1.19.0,所以这里我们不能直接使用go get'命令来安装第三方包,我们要使用/go mod文件来实现,下面是主要步骤:

  • 在项目中创建文件夹(博主的是Kafkademo)
  • 打开终端,输入go mod init,进行go.mod文件的初始化:

  • 我们在.mod文件内指定第三方包及其版本:
module Kafkademo
require (
  github.com/Shopify/sarama v1.19
)
go 1.21.6

其实这是已经可以使用命令go mod tidy了,但是博主在做的时候发现,这样会直接清除掉.mod文件里面的内容,所以建议先创建一个producer文件,在文件里面写:

package main
import (
  "fmt"
  "github.com/Shopify/sarama"
)
func main() {
  config := sarama.NewConfig()
  config.Producer.RequiredAcks = sarama.WaitForAll                                
}

这时候再打开终端输入go mod tidy

等待命令运行完毕,打开.mod文件,看到如下内容就OK了:

利用sarama向Kafka发送消息(消息的生产)

代码

package main
import (
  "fmt"
  "github.com/Shopify/sarama"
)
func main() {
  config := sarama.NewConfig()                              //创建config实例
  config.Producer.RequiredAcks = sarama.WaitForAll          //发送完数据需要leader和follow都确认
  config.Producer.Partitioner = sarama.NewRandomPartitioner //创建随机分区
  config.Producer.Return.Successes = true                   //成功交付的消息将在success channel返回
  //创建信息
  msg := &sarama.ProducerMessage{}
  msg.Topic = "web.log"
  msg.Value = sarama.StringEncoder("this is a test log")
  //连接KafKa
  client, err := sarama.NewSyncProducer([]string{"127.0.0.1:9092"}, config)
  if err != nil {
    fmt.Println("producer closed, err:", err)
    return
  }
  defer client.Close()
  //发送消息
  pid, offset, err := client.SendMessage(msg)
  if err != nil {
    fmt.Println("send msg failed,err:", err)
    return
  }
  fmt.Printf("pid:%v offset:%v\n", pid, offset)
}

运行过程

  • 首先我们打开终端开起ZooKepper服务
zkServer
  • 然后再Kafka所在文件夹下输入命令运行Kafka:
.\bin\windows\kafka-server-start.bat .\config\server.properties

最后运行程序即可,输出结果为:

补充:消息的消费

代码

package main
import (
  "fmt"
  "github.com/Shopify/sarama"
  "time"
)
func main() {
  customer, err := sarama.NewConsumer([]string{"127.0.0.1:9092"}, nil)
  if err != nil {
    fmt.Println("failed init customer,err:", err)
    return
  }
  partitionlist, err := customer.Partitions("web.log-0") //获取topic的所有分区
  if err != nil {
    fmt.Println("failed get partition list,err:", err)
    return
  }
  fmt.Println("partitions:", partitionlist)
  for partition := range partitionlist { // 遍历所有分区
    //根据消费者对象创建一个分区对象
    pc, err := customer.ConsumePartition("web.log", int32(partition), sarama.OffsetNewest)
    if err != nil {
      fmt.Println("failed get partition consumer,err:", err)
      return
    }
    defer pc.Close() // 移动到这里
    go func(consumer sarama.PartitionConsumer) {
      defer pc.AsyncClose() // 移除这行,因为已经在循环结束时关闭了
      for msg := range pc.Messages() {
        fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v", msg.Partition, msg.Offset, msg.Key, msg.Value)
      }
    }(pc)
    time.Sleep(time.Second * 10)
  }
}

不过这个不能实现直接的消费,后续我们会对这个进行补充,这里仅作介绍

相关文章
|
15天前
|
存储 缓存 安全
Go 语言中的 Sync.Map 详解:并发安全的 Map 实现
`sync.Map` 是 Go 语言中用于并发安全操作的 Map 实现,适用于读多写少的场景。它通过两个底层 Map(`read` 和 `dirty`)实现读写分离,提供高效的读性能。主要方法包括 `Store`、`Load`、`Delete` 等。在大量写入时性能可能下降,需谨慎选择使用场景。
|
23天前
|
算法 安全 Go
公司局域网管理系统里的 Go 语言 Bloom Filter 算法,太值得深挖了
本文探讨了如何利用 Go 语言中的 Bloom Filter 算法提升公司局域网管理系统的性能。Bloom Filter 是一种高效的空间节省型数据结构,适用于快速判断元素是否存在于集合中。文中通过具体代码示例展示了如何在 Go 中实现 Bloom Filter,并应用于局域网的 IP 访问控制,显著提高系统响应速度和安全性。随着网络规模扩大和技术进步,持续优化算法和结合其他安全技术将是企业维持网络竞争力的关键。
46 2
公司局域网管理系统里的 Go 语言 Bloom Filter 算法,太值得深挖了
|
24天前
|
前端开发 Go API
开箱即用的GO后台管理系统 Kratos Admin - 前端权限
Kratos Admin 是一个开箱即用的 GO 后台管理系统,前端权限管理分为路由和按钮权限。路由权限支持后端动态生成和前端固定配置两种方式,通过 `.env` 文件配置 `VITE_ROUTER_ACCESS_MODE` 实现。按钮权限可通过权限码或角色控制,支持组件、API 和指令三种方式。项目代码托管于 Gitee 和 Github。
24 0
|
23天前
|
SQL JSON 关系型数据库
开箱即用的GO后台管理系统 Kratos Admin - 列表查询规则
Kratos Admin 是一个开箱即用的GO后台管理系统,支持通用列表查询请求。通过 `page`、`pageSize`、`query`(AND过滤)、`or`(OR过滤)、`orderBy`(排序)、`noPaging`(不分页)和 `fieldMask`(字段掩码)等参数,灵活配置查询条件。过滤规则遵循Python ORM风格,支持多种查找类型如 `in`、`gte`、`icontains` 等,适用于不同数据库。
26 0
|
1月前
|
监控 关系型数据库 MySQL
【01】客户端服务端C语言-go语言-web端PHP语言整合内容发布-优雅草网络设备监控系统-硬件设备实时监控系统运营版发布-本产品基于企业级开源项目Zabbix深度二开-分步骤实现预计10篇合集-自营版
【01】客户端服务端C语言-go语言-web端PHP语言整合内容发布-优雅草网络设备监控系统-硬件设备实时监控系统运营版发布-本产品基于企业级开源项目Zabbix深度二开-分步骤实现预计10篇合集-自营版
32 0
|
2月前
|
存储 监控 算法
内网监控系统之 Go 语言布隆过滤器算法深度剖析
在数字化时代,内网监控系统对企业和组织的信息安全至关重要。布隆过滤器(Bloom Filter)作为一种高效的数据结构,能够快速判断元素是否存在于集合中,适用于内网监控中的恶意IP和违规域名筛选。本文介绍其原理、优势及Go语言实现,提升系统性能与响应速度,保障信息安全。
36 5
|
3月前
|
消息中间件 缓存 监控
go高并发之路——消息中间件kafka
本文介绍了高并发业务中的流量高峰应对措施,重点讲解了Kafka消息中间件的使用,包括常用的Go语言库sarama及其版本问题,以及Kafka的版本选择建议。文中还详细解释了Kafka生产者的四种分区策略:轮询、随机、按Key和指定分区,并提供了相应的代码示例。
go高并发之路——消息中间件kafka
|
4月前
|
缓存 监控 前端开发
在 Go 语言中实现 WebSocket 实时通信的应用,包括 WebSocket 的简介、Go 语言的优势、基本实现步骤、应用案例、注意事项及性能优化策略,旨在帮助开发者构建高效稳定的实时通信系统
本文深入探讨了在 Go 语言中实现 WebSocket 实时通信的应用,包括 WebSocket 的简介、Go 语言的优势、基本实现步骤、应用案例、注意事项及性能优化策略,旨在帮助开发者构建高效稳定的实时通信系统。
243 1
|
4月前
|
存储 负载均衡 监控
如何利用Go语言的高效性、并发支持、简洁性和跨平台性等优势,通过合理设计架构、实现负载均衡、构建容错机制、建立监控体系、优化数据存储及实施服务治理等步骤,打造稳定可靠的服务架构。
在数字化时代,构建高可靠性服务架构至关重要。本文探讨了如何利用Go语言的高效性、并发支持、简洁性和跨平台性等优势,通过合理设计架构、实现负载均衡、构建容错机制、建立监控体系、优化数据存储及实施服务治理等步骤,打造稳定可靠的服务架构。
96 1
|
4月前
|
数据库连接 Go 数据库
Go语言中的错误注入与防御编程。错误注入通过模拟网络故障、数据库错误等,测试系统稳定性
本文探讨了Go语言中的错误注入与防御编程。错误注入通过模拟网络故障、数据库错误等,测试系统稳定性;防御编程则强调在编码时考虑各种错误情况,确保程序健壮性。文章详细介绍了这两种技术在Go语言中的实现方法及其重要性,旨在提升软件质量和可靠性。
75 1