likes
comments
collection
share

Rocket MQ 安装与使用

作者站长头像
站长
· 阅读数 14

什么是 MQ

用一个例子来讲解什么是 MQ，比如在添加用户时，在成功添加用户后给用户发短信

把这个功能拆解一下，可以分为两个步骤：

  1. 添加用户
  2. 发送短信

如果现在并发量比较大时，同步完成这样两个步骤，会影响性能

这时候可以把发送短信的步骤放到一个队列中 queue，然后异步处理，这样就不会影响性能

当添加到 queue 中，能够马上给出响应，无需等到短信发送成功后才给出响应

如下图所示：

Rocket MQ 安装与使用

现在你在本地开发环境中，这样做没有太的问题

但如果你把发送短信的服务部署到另一台服务器上，queue 就无法使用了

你需要一个第三方组件来传递消息

比如 MySQL、Redis、MQ 等，这些都可以用来传递消息

Rocket MQ 安装与使用

MQ 的应用场景：

  1. 应用解耦 Rocket MQ 安装与使用
  2. 削峰
    • 当流量比较大时，数据库容易崩溃，数据库崩溃了也就意味着服务不可用了，我们可以通过 MQ 容量来控制流量，从而保证数据库的正常运行
      • 数据库有并发上限，一味的扩容除了增加经济负担外，并不能解决问题
      • 你的服务只能处理 2000 并发，就设置 MQ 每次处理 2000 个，一个个排队
  3. 数据分发

缺点：

  1. 系统可用性降低
    • 引入外部以来越多，稳定性越差
  2. 系统复杂度提高
    • 同步处理变成异步处理
  3. 一致性问题
    • 处理失败，如果保证一致性

安装 Rocket MQ

Rocket MQ 安装的工具有点多：

  • foxiswho/rcokermq:server
  • foxiswho/rocketmq:broker
  • styletang/rocketmq-console-ng
  1. 配置网络：
    • 使用 docker-compose 创建各个容器，需要保证个容器之间的通信问题
      • 如果你的 docker 中已经创建了一个网络，docker network create xxxx 
        networks:
  network1:
    external: true
      • 如果没有创建网络，可以使用 docker-compose 创建网络 
        networks:
  network1:
    name: network1
    driver: bridge
  2. 配置 foxiswho/rocketmq:broker 中的访问地址
    • 配置文件在 /etc/rocketmq/broker.conf，将 brokerIP1=192.168.0.2 设置为宿主机的IP 地址
  3. 运行 docker-compose up -d 启动容器
version: "3.5"
services:
  go-rmqnamesrv:
    image: foxiswho/rocketmq:server
    container_name: go-rmqnamesrv
    ports:
      - 9876:9876
    networks:
      network1:
        aliases:
          - go-rmqnamesrv

  go-rmqbroker:
    image: foxiswho/rocketmq:broker
    container_name: go-rmqbroker
    ports:
      - 10909:10909
      - 10911:10911
    environment:
      NAMESRV_ADDR: "go-rmqnamesrv:9876"
      JAVA_OPTS: " -Duser.home=/opt"
      JAVA_OPT_EXT: "-server -Xms256m -Xmx256m -Xmn256m"
    command: mqbroker -c /etc/rocketmq/broker.conf
    depends_on:
      - go-rmqnamesrv
    volumes:
      - ./conf/broker.conf:/etc/rocketmq/broker.conf
    networks:
      network1:
        aliases:
          - go-rmqbroker

  go-rmqconsole:
    image: styletang/rocketmq-console-ng
    container_name: go-rmqconsole
    ports:
      - 8080:8080
    environment:
      JAVA_OPTS: "-Drocketmq.namesrv.addr=go-rmqnamesrv:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false"
    depends_on:
      - go-rmqnamesrv
    networks:
      network1:
        aliases:
          - go-rmqconsole
networks:
  network1:
    external: true

Rocket MQ 基本概念

  • Producer：生产者
  • Consumer: 消费者
  • Name Server: 用来同步数据节点中的信息
  • Broker: 信息存储节点
    • 读取信息时需要通过 Name Server 获取 Broker 的信息
  • Topic: 区分消息的种类
  • Message Queue: 相当于是 Topic 的分区

按发送特点分

同步发送

  1. 同步发送，线程阻塞，投递 completes 阻塞
  2. 如果发送失败，会在默认的超时时间 3 秒内重试，最多重试 2 次
  3. 投递 completes 不代表成功，要通过检查 SendResult.sendStatus 是否成功
  4. SendResult 里面有发送状态的枚举，SendStatus，有四种状态：
    • SEND_OK：发送成功
    • FLUSH_DISK_TIMEOUT：刷盘超时
    • FLUSH_SLAVE_TIMEOUT：同步超时
    • SLAVE_NOT_AVAILABLE：没有可用的 slave 节点
  5. 在重试时，可以 ack 的 SendStatus=SEND_OK 才会停止重试

ps: 发送同步消息且 ack 为 Send_OK，只有代表消息成功写入 MQ，不代表该消息被成功消费

异步发送

  1. 异步调用的话，当前线程一定要等待异步线程回调结束再关闭 producer，因为是异步，不会阻塞，提前关闭 producer 会导致断开连接
    • 异步调用时，它会立马给你一个响应，这个响应不代表是你调用成功了
  2. 异步消息不重试，投递失败回调 onException() 方法，只有同步消息才会重试
  3. 异步消息发送一般用于耗时较长的场景，比如：视频上传后转码

单向发送

  1. 消息不可靠，性能高，只负责向服务器里发送一条消息，不重试也不关心是否发送成功
  2. 此方式发送消息的过程耗时非常短，一般在微妙级别

按使用功能特点分

普通消息

普通消息是业务中使用最多的消息，生产者关注消息发送成功即可，消费者消费到消息即可，不保证消息的顺序，所以并发量很高

顺序消息

顺序消息分为 2 种：

  • 全局顺序消息（先进先出）
  • 分区顺序消息（1 个 Broker 内的消息先进先出）

延迟消息

经过一段时间后才会投递，Rocket MQ 定义了一些级别，只能选择这些 queue： 1s、5s、10s、30s、1m、2m、3m、4m、5m、6m、7m、8m、9m、10m、20m、30m、1h、2h

msg.SetDelayTimeLevel(2);
sendResult := producer.send(msg);

实现方式：

  • 发送消息的时，如果消息设置了 DelayTimeLevel，那么该消息会被丢到 ScheduleMessageService.SCHEDULE_TOPIC 中
  • 根据 DelayTimeLevel 选择对应的 queue
  • 再把真实的 topic 和 queue 信息封装起来，set 到 msg 中
  • 然后每个 SCHEDULE_TOPIC_XXXX 的每个 DelayTimeLevelQueue，有定时任务去刷新，是否有待投递的消息
  • 每 10s 定时持久化发送进度

事务消息

  • 事务消息：达到分布式事务的最终一致
  • 半事务消息：暂不能投递的消息，发送方已经成功地将消息发送到了消息队列 RocketMQ 版服务端，但是服务端未收到生产者对该消息的二次确认，此时该消息被标记成“暂不能投递”状态，处于该种状态下的消息即半事务消息
  • 消息回查：由于网络闪断、生产者应用重启等原因，导致某条事务消息的二次确认丢失，消息队列 RocketMQ 版服务端通过扫描发现某条消息长期处于“半事务消息”时，需要主动向消息生产者询问该消息的最终状态（Commit 或是 Rollback），该询问过程即消息回查

使用 go 连接 Rocket MQ

发送普通消息

文档

import (
  "context"
  "fmt"

  "github.com/apache/rocketmq-client-go/v2"
  "github.com/apache/rocketmq-client-go/v2/primitive"
  "github.com/apache/rocketmq-client-go/v2/producer"
)

func main() {
  p, err := rocketmq.NewProducer(producer.WithNameServer([]string{"http://go-rmqnamesrv:9876"}))
  if err != nil {
    panic("生成 producer 失败")
  }

  if err = p.Start(); err != nil {
    panic("启动 producer 失败")
  }

  res, err := p.SendSync(context.Background(), &primitive.Message{
    Topic: "uccs",
    Body:  []byte("Hello RocketMQ Go Client!"),
  })

  if err != nil {
    fmt.Printf("发送消息失败: %s\n", err)
    return
  }

  fmt.Printf("发送消息成功: %s\n", res.String())
  if err := p.Shutdown(); err != nil {
    panic("关闭 producer 失败")
  }
}

响应

INFO[0000] the topic route info changed                  changeTo="{\"OrderTopicConf\":\"\",\"queueDatas\":[{\"brokerName\":\"broker-a\",\"readQueueNums\":4,\"writeQueueNums\":4,\"perm\":6,\"topicSynFlag\":0}],\"brokerDatas\":[{\"cluster\":\"DefaultCluster\",\"brokerName\":\"broker-a\",\"brokerAddrs\":{\"0\":\"172.21.0.46:10911\"}}]}" changedFrom="<nil>" topic=uccs
发送消息成功: SendResult [sendStatus=0, msgIds=AC120002240C000000002c89fe280001, offsetMsgId=AC15002E00002A9F000000000000028C, queueOffset=4, messageQueue=MessageQueue [topic=uccs, brokerName=broker-a, queueId=1]]
INFO[0000] will remove client from clientMap             clientID=172.18.0.2@74764

消费消息

文档

它有一个 NewPullConsumer 方法和 NewPushConsumer 方法，区别在于：

  • NewPullConsumer：主动拉取消息，消费者主动去拉取消息
  • NewPushConsumer：被动接收消息，RocketMQ 会推送消息给消费者

消费者在消费时，需要设置一个 groupName，因为当消费者是集群时，某个消费者消费了，其他消费者就不会重复消费了，达到负载均衡的效果

Rocket MQ 安装与使用

import (
  "context"
  "fmt"
  "time"

  "github.com/apache/rocketmq-client-go/v2"
  "github.com/apache/rocketmq-client-go/v2/consumer"
  "github.com/apache/rocketmq-client-go/v2/primitive"
)

func main() {
  c, _ := rocketmq.NewPushConsumer(
    consumer.WithNameServer([]string{"http://go-rmqnamesrv:9876"}),
    consumer.WithGroupName("testGroup"),
  )
  if err := c.Subscribe("uccs", consumer.MessageSelector{}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
    for i := range msgs {
      fmt.Printf("收到消息: %s\n", msgs[i].Body)
    }
    return consumer.ConsumeSuccess, nil
  }); err != nil {
    panic(err)
  }
  c.Start()

  time.Sleep(time.Hour)
  _ = c.Shutdown()
}

延迟消息

  • 延迟消息只需设置 DelayTimeLevel 即可： 
    msg := &primitive.Message{
  Topic: "uccs",
  Body: []byte("this is a delay message 2"),
  }
msg.WithDelayTimeLevel(4)

import (
  "context"
  "fmt"

  "github.com/apache/rocketmq-client-go/v2"
  "github.com/apache/rocketmq-client-go/v2/primitive"
  "github.com/apache/rocketmq-client-go/v2/producer"
)

func main() {
  p, err := rocketmq.NewProducer(producer.WithNameServer([]string{"http://go-rmqnamesrv:9876"}))
  if err != nil {
    panic("生成 producer 失败")
  }

  if err = p.Start(); err != nil {
    panic("启动 producer 失败")
  }

  msg := &primitive.Message{
    Topic: "uccs",
    Body:  []byte("this is a delay message 2"),
  }
  msg.WithDelayTimeLevel(4)
  res, err := p.SendSync(context.Background(), msg)
  if err != nil {
    fmt.Printf("发送消息失败: %s\n", err)
    return
  }
  fmt.Printf("发送消息成功: %s\n", res.String())
  if err := p.Shutdown(); err != nil {
    panic("关闭 producer 失败")
  }
}

事务消息

  • 正常投递，不会回查（不执行 CheckLocalTransaction） 
    func (l *Listener) ExecuteLocalTransaction(msg *primitive.Message) primitive.LocalTransactionState {
  fmt.Println("开始执行本地逻辑")
  time.Sleep(time.Second * 3)
  fmt.Println("执行本地逻辑成功")
  return primitive.CommitMessageState
}
func (l *Listener) CheckLocalTransaction(msg *primitive.MessageExt) primitive.LocalTransactionState {
  return primitive.RollbackMessageState
}
  • 执行失败，进行 rollback（不执行 CheckLocalTransaction） 
    func (l *Listener) ExecuteLocalTransaction(msg *primitive.Message) primitive.LocalTransactionState {
  fmt.Println("开始执行本地逻辑")
  time.Sleep(time.Second * 3)
  fmt.Println("执行本地逻辑失败")
  return primitive.RollbackMessageState
}
func (l *Listener) CheckLocalTransaction(msg *primitive.MessageExt) primitive.LocalTransactionState {
  return primitive.RollbackMessageState
}
  • 宕机、服务挂了、代码异常等，会回查（执行 CheckLocalTransaction） 
    func (l *Listener) ExecuteLocalTransaction(msg *primitive.Message) primitive.LocalTransactionState {
  fmt.Println("开始执行本地逻辑")
  time.Sleep(time.Second * 3)
  fmt.Println("执行本地逻辑失败")
  // 本地执行逻辑无缘无故失败，比如代码异常，宕机等
  return primitive.UnknowState
}
func (l *Listener) CheckLocalTransaction(msg *primitive.MessageExt) primitive.LocalTransactionState {
  fmt.Println("开始回查")
  time.Sleep(time.Second * 15)
  return primitive.CommitMessageState
}