Kafka消费服务孵化

项目开发过程中会有异步任务的场景，而kafka作为一款性能非常高效的队列服务，被广泛使用，因此我们的孵化器增加了Kafka消费服务孵化的功能。

消息如何生产？

我们专门提供了一个MQ消息代理中间件Pan,除kafka外，还支持rabbitmq、rocketmq、nsq等消息的生产
更多详细介绍请参考Pan的帮助文档

使用hera孵化kafka消费服务

新建一个main包main.go, 以下是一个通用模板，唯一变化的地方就是//注入处理方法

  1. package main
  3. import (
  4.     bs "github.com/tal-tech/hera/bootstrap"
  5.     "github.com/tal-tech/hera/kafkaconsumer"
  6.     "github.com/tal-tech/xesTools/confutil"
  7.     _ "github.com/tal-tech/xesTools/expvarutil"
  8.     "github.com/tal-tech/xesTools/flagutil"
  9.     "os"
  10.     "os/signal"
  11.     "syscall"
  12. )
  14. var (
  15.     exit = make(chan os.Signal, 2)
  16. )
  18. func main() {
  19.     //init config
  20.     confutil.InitConfig()
  22.     // watch exit signal
  23.     signal.Notify(exit, os.Interrupt, syscall.SIGTERM)
  26.     //生成kafka消费者服务
  27.     s := kafkaconsumer.NewServer()
  29.     s.AddBeforeServerStartFunc(bs.InitLogger(),
  30.         bs.InitTraceLogger("HS-Golang", "0.1"),
  31.         bs.InitPerfutil(),
  32.         bs.InitPprof(),
  33.     )
  34.     s.AddAfterServerStopFunc(bs.CloseLogger())
  36.     //注入处理方法
  37.     s.InjectHandleFuncs(handlerFuncs)
  38.     err := s.Start()
  39.     if err != nil {
  40.         panic(err)
  41.     }
  42.     // wait for exit signal
  43.     <-exit
  44.     s.Stop()
  45. }

配置书写

  1. [KafkaServer]
  2. groupName=
  3. consumerCnt=5
  4. ;多个用逗号分隔
  5. topic=test,test1
  6. failTopic=falitopic
  7. ;多个broker用逗号分隔
  8. kafkaHost=localhost:9092

消息处理方法

待注入的参数必须是[]kafkaconsumer.HandlerFunc类型，该类型的定义如下：

  1. type HandlerFunc struct {
  2.     //消息标识符
  3.     MessageKey string
  5.     //标识符所对应的处理方法
  6.     //参数[]byte 就是存储在kafka中的消息体
  7.     Func       func([]byte) error
  8. }

解释：

  • 标识符MessageKey 和Func 一一对应
  • 消息体的协议格式为：MessageKey+” “+实际消息，注意中间有一个空格，所以我们在定义标识符时一定不能包含空格，否则解析消息时会产生错误

创建消费函数 funcs.go

  1. package main
  3. import(
  4.     "fmt"
  5.     "git.100tal.com/wangxiao_go_lib/hera/kafkaconsumer"
  6. )
  8. var handlerFuncs = []kafkaconsumer.HandlerFunc{
  9.     {
  10.         MessageKey: "kafka_msg_test1",
  11.         Func:       daalFunc,
  12.     },
  13. }
  15. func daalFunc(v []byte) (err error) {
  16.     fmt.Println(string(v))
  17.     return 
  18. }

消费安全控制级别

At Least One 即至少消费一次