centos7 install rocketmq 宿主机快速搭建RocketMQ单机开发环境

为什么采用宿主机而不采用 Docker 方式快速搭建

在搭建 RocketMQ 测试环境时，我们可以选择在宿主机上直接安装和配置，也可以使用 Docker 容器来快速搭建。然而，为什么我们选择了在宿主机上安装而不是使用 Docker 方式呢？

调整配置的灵活性：在 Docker 方式下，虽然可以使用环境变量来设置配置项，但是有时候需要对容器内部的配置文件进行修改，（默认RocketMQ broker 吃内存8个G电脑配置差docker更本起不来）这可能不太方便。在宿主机上直接安装，则可以直接编辑配置文件，更加灵活地调整配置。
调试和排查问题的方便性：在宿主机上安装的 RocketMQ 实例，可以直接在宿主机上查看日志、调试程序，更容易排查问题。而在 Docker 容器中，需要通过 Docker logs 命令或者进入容器内部来查看日志，相对来说更加繁琐。

综上所述，虽然 Docker 提供了快速部署和便捷管理的优势，但在某些情况下，直接在宿主机上安装 RocketMQ 可能更加灵活、性能更好，并且更容易调试和排查问题。因此，在选择搭建测试环境时，我们根据实际情况选择了在宿主机上安装 RocketMQ。

第一步：源码下载和安装

wget https://archive.apache.org/dist/rocketmq/4.9.5/rocketmq-all-4.9.5-bin-release.zip

unzip rocketmq-all-4.9.5-bin-release.zip

mv rocketmq-all-4.9.5-bin-release /app/rocketmq

查看RocketMQ的更多版本链接

第二步：修改初始内存

# 修改初始内存【修改两个启动脚本的jvm】
cd /app/rocketmq/bin/
vim runserver.sh
# 修改初始堆内存大小，要不测试环境没那么多内存 
原：JAVA_OPT="${JAVA_OPT} -server -Xms4g -Xmx4g -Xmn2g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
改：JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn512m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

vim runbroker.sh
# 修改初始堆内存大小，要不测试环境没那么多内存 
原:  JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g"
改:  JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m"

第三步：启动RocketMQ单机

# 启动NameServer，bin目录前台启动，确定没报错再后台启动  【sh bin/mqnamesrv &】
[root@slavenode1 bin]# ./mqnamesrv
The Name Server boot success. serializeType=JSON

# 启动broker，bin目录前台启动，确定没报错再后台启动
[root@masternode1 bin]# ./mqbroker -n localhost:9876
The broker[masternode1, 192.168.1.160:10911] boot success. serializeType=JSON and name server is localhost:9876

# 查看进程
[root@masternode1 ~]# jps
2214 Jps
2071 BrokerStartup
1947 NamesrvStartup

第四步：发送消息测试

消费着启动

 export NAMESRV_ADDR=localhost:9876
 ./tools.sh org.apache.rocketmq.example.quickstart.Consumer

发送测试消息

 export NAMESRV_ADDR=localhost:9876
 ./tools.sh org.apache.rocketmq.example.quickstart.Producer

第五步：docker控制台搭建rocketmq-dashboard

docker run -d \
    --name rocketmq-dashboard \
    -p 8060:8080 \
    -v /apps/rocketmq-dashboard/data:/tmp/rocketmq-console/data \
    -e NAMESRV_ADDR=0.0.0.0:9876 \
    apacherocketmq/rocketmq-dashboard:1.0.0

-v /apps/rocketmq-dashboard/data:/tmp/rocketmq-console/data 参数用于将主机的 /apps/rocketmq-dashboard/data 目录映射到容器的 /tmp/rocketmq-console/data
目录，这样可以持久化保存 RocketMQ Dashboard 的数据。
-e NAMESRV_ADDR=0.0.0.0:9876 参数用于设置环境变量 NAMESRV_ADDR，指定 RocketMQ 的 Namesrv 地址。

第六步：rocketmq-client-go 测试延迟消息

延迟5秒代码测试示例
在这里插入图片描述

consumer.go

package main

import (
	"context"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"os"
	"time"
)

func main() {
	// 设置推送消费者
	c, _ := rocketmq.NewPushConsumer(
		//消费组
		consumer.WithGroupName("testGroup"),
		// namesrv地址
		consumer.WithNameServer([]string{"127.0.0.1:9876"}),
	)
	// 必须先在 开始前
	err := c.Subscribe("Topic-test", consumer.MessageSelector{}, func(ctx context.Context, ext ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		for i := range ext {
			// 获取消息正文
			body := ext[i].Body
			fmt.Printf("获取到值:%s \n", body)
		}
		return consumer.ConsumeSuccess, nil
	})
	if err != nil {
		fmt.Println(err.Error())
	}
	err = c.Start()
	if err != nil {
		fmt.Println(err.Error())
		os.Exit(-1)
	}
	time.Sleep(time.Hour)
	err = c.Shutdown()
	if err != nil {
		fmt.Printf("shutdown Consumer error:%s", err.Error())
	}
}

export ROCKETMQ_GO_LOG_LEVEL=warn
go run   consumer.go

producer_delay.go


func main() {
	p, _ := rocketmq.NewProducer(
		// 设置 nameSrvAddr
		// nameSrvAddr 是 Topic 路由注册中心
		producer.WithNameServer([]string{"0.0.0.0:9876"}),
		// 指定发送失败时的重试时间
		producer.WithRetry(3),
		// 设置 Group
		producer.WithGroupName("testGroup"),
	)
	// 开始连接
	err := p.Start()
	if err != nil {
		fmt.Printf("start producer error: %s", err.Error())
		os.Exit(2)
	}

	// 设置节点名称
	topic := "Topic-test"
	// 循环发送信息 (同步发送)
	for i := 0; i < 10; i++ {
		msg := &primitive.Message{
			Topic: topic,
			Body:  []byte("RocketMQ Go producer_delay.go [" + strconv.Itoa(i) + "] " + time.Now().Format("2006-01-02 15:04:05")),
		}
		time.Sleep(time.Second)
		// WithDelayTimeLevel set message delay time to consume.
		// reference delay level definition: 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
		// delay level starts from 1. for example, if we set param level=1, then the delay time is 1s.
		// 设置消息的延时级别为 2，对应 RocketMQ 配置中的延时时间 5s
		msg.WithDelayTimeLevel(2)
		// 发送信息
		res, err := p.SendSync(context.Background(), msg)
		if err != nil {
			fmt.Printf("发送失败！:%s\n", err)
		} else {
			fmt.Printf("发送成功 : result=%s\n", res.String())
		}
	}
	// 关闭生产者
	err = p.Shutdown()
	if err != nil {
		fmt.Printf("shutdown producer error:%s", err.Error())
	}
}