使用docker安装RocketMQ

男娘i 2022-10-30 15:27 169阅读 0赞

# 使用docker安装RocketMQ #

> [官网地址][Link 1]

## 1、下载 ##

# git clone https://github.com/apache/rocketmq-docker.git
    # cd rocketmq-docker/
    # ls
    CONTRIBUTING.md  image-build  LICENSE  NOTICE  product  README.md  stage.sh  templates
    # cd image-build/
    # ls
    build-image.sh  Dockerfile-alpine  Dockerfile-centos  scripts  update.sh

## 2、创建RocketMQ镜像 ##

> 构建命令： `sh build-image.sh RMQ-VERSION BASE-IMAGE`  
> 可选[RMQ-VERSION][]

我们使用 4.7.1 alpine

➜  image-build git:(master)  sh build-image.sh 4.7.1 alpine
    Version = 4.7.1
    Sending build context to Docker daemon  27.14kB
    Step 1/22 : FROM openjdk:8-alpine
     ---> a3562aa0b991
    
    //..........构建时间较长请耐心等待
    
    //构建完成
    Successfully built 128108c2e50d
    Successfully tagged apacherocketmq/rocketmq:4.7.1-alpine

构建完成后，我们就能查询到镜像

# docker images |grep mq
    apacherocketmq/rocketmq  4.7.1-alpine   128108c2e50d     4   9 seconds ago      145MB

## 生成配置 ##

# cd ..
    # ls
    CONTRIBUTING.md  image-build  LICENSE  NOTICE  product  README.md  stage.sh  templates
    # sh stage.sh 4.7.1 (这里的4.7.1对应之前的镜像版本)
    Stage version = 4.7.1
    mkdir /root/rocketmq/rocketmq-docker/stages/4.7.1
    staged templates into folder /root/rocketmq/rocketmq-docker/stages/4.7.1
    # ls
    CONTRIBUTING.md  image-build  LICENSE  NOTICE  product  README.md  stages  stage.sh  templates

生成了stages目录，里面存放配置模板文件

# cd stages/
    # ls
    4.7.1
    # cd 4.7.1/
    # ls
    data            kubernetes        play-docker-compose.sh  play-docker.sh      play-kubernetes.sh  ssl
    docker-compose  play-consumer.sh  play-docker-dledger.sh  play-docker-tls.sh  play-producer.sh

部署方式有以下5种

1.  单机

./play-docker.sh alpine

1.  docker-composer

./play-docker-compose.sh

1.  kubernetes集群

./play-kubernetes.sh

1.  Cluster of Dledger storage(RocketMQ需要4.4.0版本以上)

./play-docker-dledger.sh

1.  TLS

./play-docker-tls.sh
    ./play-producer.sh
    ./play-consumer.sh

我这里选择的是单机部署，可以看到生成了两个容器

# docker ps |grep mq
    5b557ea1e6be        apacherocketmq/rocketmq:4.7.1-alpine                         "sh mqbroker"            25 seconds ago                                                                                      Up 24 seconds       0.0.0.0:10909->10909/tcp, 9876/tcp, 0.0.0.0:10911-10912->10911-10912/tcp   rmqbroker
    8b1318aee5d6        apacherocketmq/rocketmq:4.7.1-alpine                         "sh mqnamesrv"           26 seconds ago                                                                                      Up 25 seconds       10909/tcp, 0.0.0.0:9876->9876/tcp, 10911-10912/tcp                         rmqnamesrv

验证RocketMQ启动成功

1、使用命令 `docker ps|grep rmqbroker`找到RocketMQ broker的容器id

2、使用命令 `docker exec -it 5b557ea1e6be ./mqadmin clusterList -n nameserver_ip:9876` 验证RocketMQ broker工作正常

> nameserver\_ip请进入容器rmqnamesrv内部查看

# docker exec -it 5b557ea1e6be ./mqadmin clusterList -n 172.17.0.2:9876
    RocketMQLog:WARN No appenders could be found for logger (io.netty.util.internal.PlatformDependent0).
    RocketMQLog:WARN Please initialize the logger system properly.
    #Cluster Name #Broker Name #BID #Addr #Version #InTPS(LOAD) #OutTPS(LOAD) #PCWait(ms) #Hour #SPACE
    DefaultCluster    5b557ea1e6be            0     172.17.0.8:10911       V4_7_1                   0.00(0,0ms)         0.00(0,0ms)          0 447225.46 -1.0000

如需升级执行以下命令

cd image-build
    ./update.sh

## 安装RocketMQ控制台 ##

//拉取镜像
    docker pull apacherocketmq/rocketmq-console:2.0.0
    //运行容器，这里172.15.65.xx为宿主机ip
    docker run -e "JAVA_OPTS=-Drocketmq.namesrv.addr=172.15.65.xx:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false" -p 6881:8080 -t apacherocketmq/rocketmq-console:2.0.0

启动容器后可以通过 http://localhost:6881/\#/ 登录控制台

## golang client使用问题 ##

由于使用的docker服务启动，broker的地址是内网地址，需要将地址修改为外网地址

# docker ps |grep mq
    8abb966542a3        apacherocketmq/rocketmq-console:2.0.0                        "sh -c 'java $JAVA..."   17 hours ago        Up 17 hours         0.0.0.0:6881->8080/tcp                                                     dazzling_tesla
    5b557ea1e6be        apacherocketmq/rocketmq:4.7.1-alpine                         "sh mqbroker"            18 hours ago        Up 18 hours         0.0.0.0:10909->10909/tcp, 9876/tcp, 0.0.0.0:10911-10912->10911-10912/tcp   rmqbroker
    8b1318aee5d6        apacherocketmq/rocketmq:4.7.1-alpine                         "sh mqnamesrv"           18 hours ago        Up 18 hours         10909/tcp, 0.0.0.0:9876->9876/tcp, 10911-10912/tcp                         rmqnamesrv
    # docker exec -it 5b557ea1e6be bash // 进入到容器内部修改配置
    # vi ../conf/broker.conf

在文件中添加外网地址 `brokerIP1=172.15.65.xx`

# Licensed to the Apache Software Foundation (ASF) under one or more
    # contributor license agreements. See the NOTICE file distributed with
    # this work for additional information regarding copyright ownership.
    # The ASF licenses this file to You under the Apache License, Version 2.0
    # (the "License"); you may not use this file except in compliance with
    # the License. You may obtain a copy of the License at
    #
    # http://www.apache.org/licenses/LICENSE-2.0
    #
    # Unless required by applicable law or agreed to in writing, software
    # distributed under the License is distributed on an "AS IS" BASIS,
    # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    # See the License for the specific language governing permissions and
    # limitations under the License.
    
    brokerClusterName = DefaultCluster
    brokerName = broker-a
    brokerId = 0
    deleteWhen = 04
    fileReservedTime = 48
    brokerRole = ASYNC_MASTER
    flushDiskType = ASYNC_FLUSH
    
    # add your IP
    brokerIP1=172.15.65.xx

然后重启broker, `docker restart 5b557ea1e6be`

> 这里需要去修改启动脚本 ./play-docker.sh 里的start\_namesrv\_broker() 函数中的docker启动命令，在mybroker后面添加-c …/conf/broker.conf

这样查看cluster会发现Address变成了外网地址。

## go中运行 ##

获取`go-client`：go get -u -v github.com/apache/rocketmq-client-go/v2@v2.0.0

client-go producer-simple

> https://github.com/apache/rocketmq-client-go/blob/v2.0.0/examples/producer/simple/main.go

/*
    Licensed to the Apache Software Foundation (ASF) under one or more
    contributor license agreements.  See the NOTICE file distributed with
    this work for additional information regarding copyright ownership.
    The ASF licenses this file to You under the Apache License, Version 2.0
    (the "License"); you may not use this file except in compliance with
    the License.  You may obtain a copy of the License at
        http://www.apache.org/licenses/LICENSE-2.0
    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
    */
    
    package main
    
    import (
    	"context"
    	"fmt"
    	"os"
    	"strconv"
    
    	"github.com/apache/rocketmq-client-go/v2"
    	"github.com/apache/rocketmq-client-go/v2/primitive"
    	"github.com/apache/rocketmq-client-go/v2/producer"
    )
    
    // Package main implements a simple producer to send message.
    func main() {
    	p, _ := rocketmq.NewProducer(
    		producer.WithNameServer([]string{"127.0.0.1:9876"}),
    		producer.WithRetry(2),
    	)
    	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.Message{
    			Topic: topic,
    			Body:  []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())
    	}
    }

consumer

> https://github.com/apache/rocketmq-client-go/blob/v2.0.0/examples/consumer/simple/main.go

/*
    Licensed to the Apache Software Foundation (ASF) under one or more
    contributor license agreements.  See the NOTICE file distributed with
    this work for additional information regarding copyright ownership.
    The ASF licenses this file to You under the Apache License, Version 2.0
    (the "License"); you may not use this file except in compliance with
    the License.  You may obtain a copy of the License at
        http://www.apache.org/licenses/LICENSE-2.0
    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
    */
    
    package main
    
    import (
    	"context"
    	"fmt"
    	"os"
    	"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.WithGroupName("testGroup"),
    		consumer.WithNameServer([]string{"127.0.0.1:9876"}),
    	)
    	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)
    	}
    	time.Sleep(time.Hour)
    	err = c.Shutdown()
    	if err != nil {
    		fmt.Printf("shutdown Consumer error: %s", err.Error())
    	}
    }

[Link 1]: https://github.com/apache/rocketmq-docker
[RMQ-VERSION]: https://archive.apache.org/dist/rocketmq/