4.6 调用RocketMQ

4.6.1 简介

client/rocketmq 包是对github.com/apache/rocketmq-client-go/v2 进行二次封装

4.6.2 配置规范

配置说明4.6 调用RocketMQ - 图1 (opens new window)

4.6.3 用法

访问rocketmq示例4.6 调用RocketMQ - 图2 (opens new window)

  1. // run: go run main.go --config=config.toml
  3. package main
  5. import (
  6.     "context"
  7.     "fmt"
  8.     "strconv"
  10.     "github.com/apache/rocketmq-client-go/v2/primitive"
  11.     "github.com/douyu/jupiter"
  12.     "github.com/douyu/jupiter/pkg/client/rocketmq"
  13.     "github.com/douyu/jupiter/pkg/xlog"
  14. )
  16. // run: go run main.go -config=config.toml
  17. type Engine struct {
  18.     jupiter.Application
  19. }
  21. func NewEngine() *Engine {
  22.     eng := &Engine{}
  23.     if err := eng.Startup(
  24.         eng.exampleRocketMQProducer,
  25.         eng.exampleRocketMQConsumer,
  26.     ); err != nil {
  27.         xlog.Panic("startup", xlog.Any("err", err))
  28.     }
  29.     return eng
  30. }
  32. func main() {
  33.     app := NewEngine()
  34.     if err := app.Run(); err != nil {
  35.         panic(err)
  36.     }
  37. }
  39. func (eng *Engine) exampleRocketMQConsumer() (err error) {
  40.     consumerClient := rocketmq.StdPushConsumerConfig("configName").Build()
  41.     defer func() {
  42.         if consumerClient.Enable {
  43.             _ = consumerClient.Close()
  44.         }
  45.     }()
  46.     consumerClient.Subscribe(consumerClient.ConsumerConfig.Topic, func(ctx context.Context, ext *primitive.MessageExt) error {
  47.         fmt.Println("msg...", string(ext.Message.Body))
  48.         fmt.Println("msg topic...", string(ext.Message.Topic))
  49.         fmt.Println("msg topic tag...", string(ext.Message.GetTags()))
  50.         return nil
  51.     })
  52.     err = consumerClient.Start()
  53.     return
  54. }
  56. func (eng *Engine) exampleRocketMQProducer() (err error) {
  57.     producerClient := rocketmq.StdProducerConfig("configName").Build()
  58.     defer func() {
  59.         _ = producerClient.Close()
  60.     }()
  62.     err = producerClient.Start()
  63.     if err != nil {
  64.         return
  65.     }
  67.     for i := 0; i < 10; i++ {
  68.         msg := "a" + strconv.Itoa(i)
  69.         err = producerClient.Send([]byte(msg))
  70.     }
  71.     return
  72. }

执行 go run main.go —config=config.toml,可以看到如下图结果 image