docker compose搭建rockermq

在根目录下创建docker-env目录

docker-env
  ├─conf
  ├─logs
  └─store

docker-env/conf/broker.conf文件内容

brokerClusterName = DefaultCluster
brokerName = broker-a
brokerId = 0
deleteWhen = 04
fileReservedTime = 48
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH
# 主机IP
brokerIP1 = 192.168.110.64

docker-compose.yaml内容

version: "3"
services:
  mqnamesrv:
    image: foxiswho/rocketmq:4.7.0
    container_name: mqnamesrv
    ports:
      - 9876:9876
    environment:
      JAVA_OPT: -server -Xms256m -Xmx256m
    command: sh mqnamesrv
  mqbroker:
    image: foxiswho/rocketmq:4.7.0
    container_name: mqbroker
    ports:
      - 10909:10909
      - 10911:10911
    volumes:
      - ./docker-env/logs:/opt/logs
      - ./docker-env/store:/opt/store
      - ./docker-env/conf/broker.conf:/etc/rocketmq/broker.conf
    environment:
      NAMESRV_ADDR: mqnamesrv:9876
      JAVA_OPT_EXT: -server -Xms256m -Xmx256m -Xmn128m
    command: sh mqbroker -n mqnamesrv:9876 -c /etc/rocketmq/broker.conf
  mqconsole:
    image: styletang/rocketmq-console-ng
    container_name: mqconsole
    ports:
      - 19876:8080
    environment:
      JAVA_OPTS: -Drocketmq.namesrv.addr=mqnamesrv:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=falses

基本概念

官方文档：https://rocketmq.apache.org/zh/docs/introduction/02concepts
本文介绍 Apache RocketMQ 的基本概念，以便您更好地理解和使用 Apache RocketMQ 。

主题（Topic）

Apache RocketMQ 中消息传输和存储的顶层容器，用于标识同一类业务逻辑的消息。主题通过TopicName来做唯一标识和区分。更多信息，请参见主题（Topic）。

消息类型（MessageType）

Apache RocketMQ 中按照消息传输特性的不同而定义的分类，用于类型管理和安全校验。 Apache RocketMQ 支持的消息类型有普通消息、顺序消息、事务消息和定时/延时消息。
信息
Apache RocketMQ 从5.0版本开始，支持强制校验消息类型，即每个主题Topic只允许发送一种消息类型的消息，这样可以更好的运维和管理生产系统，避免混乱。但同时保证向下兼容4.x版本行为，强制校验功能默认关闭，推荐通过服务端参数 enableTopicMessageTypeCheck 手动开启校验。

消息队列（MessageQueue）

队列是 Apache RocketMQ 中消息存储和传输的实际容器，也是消息的最小存储单元。 Apache RocketMQ 的所有主题都是由多个队列组成，以此实现队列数量的水平拆分和队列内部的流式存储。队列通过QueueId来做唯一标识和区分。更多信息，请参见队列（MessageQueue）。

消息（Message）

消息是 Apache RocketMQ 中的最小数据传输单元。生产者将业务数据的负载和拓展属性包装成消息发送到服务端，服务端按照相关语义将消息投递到消费端进行消费。更多信息，请参见消息（Message）。

消息视图（MessageView）

消息视图是 Apache RocketMQ 面向开发视角提供的一种消息只读接口。通过消息视图可以读取消息内部的多个属性和负载信息，但是不能对消息本身做任何修改。

消息标签（MessageTag）

消息标签是Apache RocketMQ 提供的细粒度消息分类属性，可以在主题层级之下做消息类型的细分。消费者通过订阅特定的标签来实现细粒度过滤。更多信息，请参见消息过滤。

消息位点（MessageQueueOffset）

消息是按到达Apache RocketMQ 服务端的先后顺序存储在指定主题的多个队列中，每条消息在队列中都有一个唯一的Long类型坐标，这个坐标被定义为消息位点。更多信息，请参见消费进度管理。

消费位点（ConsumerOffset）

一条消息被某个消费者消费完成后不会立即从队列中删除，Apache RocketMQ 会基于每个消费者分组记录消费过的最新一条消息的位点，即消费位点。更多信息，请参见消费进度管理。

消息索引（MessageKey）

消息索引是Apache RocketMQ 提供的面向消息的索引属性。通过设置的消息索引可以快速查找到对应的消息内容。

生产者（Producer）

生产者是Apache RocketMQ 系统中用来构建并传输消息到服务端的运行实体。生产者通常被集成在业务系统中，将业务消息按照要求封装成消息并发送至服务端。更多信息，请参见生产者（Producer）。

事务检查器（TransactionChecker）

Apache RocketMQ 中生产者用来执行本地事务检查和异常事务恢复的监听器。事务检查器应该通过业务侧数据的状态来检查和判断事务消息的状态。更多信息，请参见事务消息。

事务状态（TransactionResolution）

Apache RocketMQ 中事务消息发送过程中，事务提交的状态标识，服务端通过事务状态控制事务消息是否应该提交和投递。事务状态包括事务提交、事务回滚和事务未决。更多信息，请参见事务消息。

消费者分组（ConsumerGroup）

消费者分组是Apache RocketMQ 系统中承载多个消费行为一致的消费者的负载均衡分组。和消费者不同，消费者分组并不是运行实体，而是一个逻辑资源。在 Apache RocketMQ 中，通过消费者分组内初始化多个消费者实现消费性能的水平扩展以及高可用容灾。更多信息，请参见消费者分组（ConsumerGroup）。

消费者（Consumer）

消费者是Apache RocketMQ 中用来接收并处理消息的运行实体。消费者通常被集成在业务系统中，从服务端获取消息，并将消息转化成业务可理解的信息，供业务逻辑处理。更多信息，请参见消费者（Consumer）。

消费结果（ConsumeResult）

Apache RocketMQ 中PushConsumer消费监听器处理消息完成后返回的处理结果，用来标识本次消息是否正确处理。消费结果包含消费成功和消费失败。

订阅关系（Subscription）

订阅关系是Apache RocketMQ 系统中消费者获取消息、处理消息的规则和状态配置。订阅关系由消费者分组动态注册到服务端系统，并在后续的消息传输中按照订阅关系定义的过滤规则进行消息匹配和消费进度维护。更多信息，请参见订阅关系（Subscription）。

消息过滤

消费者可以通过订阅指定消息标签（Tag）对消息进行过滤，确保最终只接收被过滤后的消息合集。过滤规则的计算和匹配在Apache RocketMQ 的服务端完成。更多信息，请参见消息过滤。

重置消费位点

以时间轴为坐标，在消息持久化存储的时间范围内，重新设置消费者分组对已订阅主题的消费进度，设置完成后消费者将接收设定时间点之后，由生产者发送到Apache RocketMQ 服务端的消息。更多信息，请参见重置消费位点。

消息轨迹

在一条消息从生产者发出到消费者接收并处理过程中，由各个相关节点的时间、地点等数据汇聚而成的完整链路信息。通过消息轨迹，您能清晰定位消息从生产者发出，经由Apache RocketMQ 服务端，投递给消费者的完整链路，方便定位排查问题。

消息堆积

生产者已经将消息发送到Apache RocketMQ 的服务端，但由于消费者的消费能力有限，未能在短时间内将所有消息正确消费掉，此时在服务端保存着未被消费的消息，该状态即消息堆积。

事务消息

事务消息是Apache RocketMQ 提供的一种高级消息类型，支持在分布式场景下保障消息生产和本地事务的最终一致性。

定时/延时消息

定时/延时消息是Apache RocketMQ 提供的一种高级消息类型，消息被发送至服务端后，在指定时间后才能被消费者消费。通过设置一定的定时时间可以实现分布式场景的延时调度触发效果。

顺序消息

顺序消息是Apache RocketMQ 提供的一种高级消息类型，支持消费者按照发送消息的先后顺序获取消息，从而实现业务场景中的顺序处理。

example

安装第三方库：
go get github.com/apache/rocketmq-client-go/v2

发送普通消息

// 发送普通消息
func TestSendSimple(t *testing.T) {
    p, _ := rocketmq.NewProducer(
        producer.WithNameServer([]string{"127.0.0.1:9876"}),
    )
    err := p.Start()
    if err != nil {
        fmt.Printf("start producer error: %s", err.Error())
        os.Exit(1)
    }
    topic := "test"
    for i := 0; i < 10; i++ {
        msg := primitive.NewMessage(topic, []byte("Hello RocketMQ Go Client! "+strconv.Itoa(i)))
        res, err := p.SendSync(context.Background(), msg)
        if err != nil {
            fmt.Printf("send message error: %s\n", err)
        } else {
            fmt.Printf("send message success: result=%s\n", res.String())
        }
    }
    err = p.Shutdown()
    if err != nil {
        fmt.Printf("shutdown producer error: %s", err.Error())
    }
}

消息接收

订阅来自某个topic的消息 c.Subscribe(topic,consumer.MessageSelector{},func)

// 消息接收
func TestConsumer(t *testing.T) {
    sig := make(chan os.Signal)
    c, _ := rocketmq.NewPushConsumer(
        consumer.WithNameServer([]string{"127.0.0.1:9876"}),
        consumer.WithGroupName("test-group"),
    )
    err := c.Subscribe("test", consumer.MessageSelector{}, func(ctx context.Context,
                                                               msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
        for i := range msgs {
            fmt.Printf("subscribe callback: %v \n", msgs[i])
        }
        return consumer.ConsumeSuccess, nil
    })
    if err != nil {
        fmt.Println(err.Error())
    }
    // Note: start after subscribe
    err = c.Start()
    if err != nil {
        fmt.Println(err.Error())
        os.Exit(-1)
    }
    <-sig
    err = c.Shutdown()
    if err != nil {
        fmt.Printf("shutdown Consumer error: %s", err.Error())
    }
}

发送延迟消息

需要传递延迟消息的时间levelmsg.WithDelayTimeLevel()

// 发送延迟消息
// DelayTimeLevel 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
func TestSendDelay(t *testing.T) {
    p, _ := rocketmq.NewProducer(
        producer.WithNameServer([]string{"127.0.0.1:9876"}),
    )
    err := p.Start()
    if err != nil {
        fmt.Printf("start producer error: %s", err.Error())
        os.Exit(1)
    }
    topic := "test"
    for i := 0; i < 10; i++ {
        msg := primitive.NewMessage(topic, []byte("delay message: Hello RocketMQ Go Client! "+strconv.Itoa(i)))
        // 延迟
        msg.WithDelayTimeLevel(2)
        res, err := p.SendSync(context.Background(), msg)
        if err != nil {
            fmt.Printf("send message error: %s\n", err)
        } else {
            fmt.Printf("send message success: result=%s\n", res.String())
        }
    }
    err = p.Shutdown()
    if err != nil {
        fmt.Printf("shutdown producer error: %s", err.Error())
    }
}

发送事务消息

参数传递需要实现ExecuteLocalTransaction和CheckLocalTransaction方法的结构体参数

// 发送事务消息
func TestSendTransaction(t *testing.T) {
    p, _ := rocketmq.NewTransactionProducer(
        &OrderListener{},
        producer.WithNameServer([]string{"127.0.0.1:9876"}),
    )
    err := p.Start()
    if err != nil {
        fmt.Printf("start producer error: %s", err.Error())
        os.Exit(1)
    }
    topic := "test-transaction"
    for i := 0; i < 1; i++ {
        msg := primitive.NewMessage(topic, []byte("transaction message: Hello RocketMQ Go Client! "+strconv.Itoa(i)))
        res, err := p.SendMessageInTransaction(context.Background(), msg)
        if err != nil {
            fmt.Printf("send message error: %s\n", err)
        } else {
            fmt.Printf("send message success: result=%s\n", res.String())
        }
    }
    select {}
    err = p.Shutdown()
    if err != nil {
        fmt.Printf("shutdown producer error: %s", err.Error())
    }
}
type OrderListener struct {
}
// 正常发送消息执行
func (receiver *OrderListener) ExecuteLocalTransaction(msg *primitive.Message) primitive.LocalTransactionState {
    strings := make(chan string)
    go func() {
        strings <- "success"
        time.Sleep(time.Second * 1)
        strings <- "error"
        time.Sleep(time.Second * 2)
        strings <- "panic"
        close(strings)
    }()
    for s := range strings {
        if s == "success" {
            // 逻辑成功
            fmt.Println("success")
            // return primitive.CommitMessageState
        } else if s == "error" {
            // 逻辑失败
            fmt.Println("error")
            // return primitive.RollbackMessageState
        } else {
            // 故障 执行回查CheckLocalTransaction
            fmt.Println("panic")
        }
    }
    return primitive.CommitMessageState
}
// 代码异常、服务宕机、以及其他可能出现的意料之外的异常情况，会执行回查
func (receiver *OrderListener) CheckLocalTransaction(*primitive.MessageExt) primitive.LocalTransactionState {
    fmt.Println("rollback message")
    return primitive.RollbackMessageState
}