docker安裝RocketMQ的實(shí)現(xiàn)步驟

更新時(shí)間：2021年11月14日 15:48:08 作者：2014Team

本文主要介紹了docker安裝RocketMQ的實(shí)現(xiàn)步驟，文中通過(guò)示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下

一、檢索鏡像

docker search rocketmq

檢索具體版本

curl https://registry.hub.docker.com/v1/repositories/foxiswho/rocketmq/tags | tr -d '[\[\]" ]' | tr '}' '\n' | awk -F: -v image='foxiswho/rocketmq' '{if(NR!=NF && $3 != ""){printf("%s:%s\n",image,$3)}}'

查看當(dāng)前鏡像所有的版本shell命令

curl https://registry.hub.docker.com/v1/repositories/foxiswho/rocketmq/tags\
| tr -d '[\[\]" ]' | tr '}' '\n'\
| awk -F: -v image='foxiswho/rocketmq' '{if(NR!=NF && $3 != ""){printf("%s:%s\n",image,$3)}}'

二、創(chuàng)建Broker Server

docker run -d -p 9876:9876 --name rmqserver  foxiswho/rocketmq:server-4.5.1

三、創(chuàng)建broker

docker run -d -p 10911:10911 -p 10909:10909\
 --name rmqbroker --link rmqserver:namesrv\
 -e "NAMESRV_ADDR=namesrv:9876" -e "JAVA_OPTS=-Duser.home=/opt"\
 -e "JAVA_OPT_EXT=-server -Xms128m -Xmx128m"\
 foxiswho/rocketmq:broker-4.5.1

配置broker.conf

進(jìn)入rmqbroker容器

docker exec -it rmqbroker /bin/bash

cd /etc/rocketmq/

vi broker.conf

四、創(chuàng)建rocketmq console

docker run -d --name rmqconsole -p 8180:8080 --link rmqserver:namesrv\
 -e "JAVA_OPTS=-Drocketmq.namesrv.addr=namesrv:9876\
 -Dcom.rocketmq.sendMessageWithVIPChannel=false"\
 -t styletang/rocketmq-console-ng

然后通過(guò)如下命令檢查一下啟動(dòng)情況

docker ps|grep rocketmq

http://192.168.23.131:8180/

五、測(cè)試

由于我在宿主機(jī)安裝rocketmq,測(cè)試如下：

#測(cè)試消息發(fā)送命令
sh tools.sh org.apache.rocketmq.example.quickstart.Producer

#測(cè)試消息接收命令
sh tools.sh org.apache.rocketmq.example.quickstart.Consumer

六、java樣例

<dependency>
	    <groupId>org.apache.rocketmq</groupId>
	    <artifactId>rocketmq-client</artifactId>
	    <version>4.9.2</version>
	</dependency>

消息發(fā)送（Producer端發(fā)送同步消息）

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
 
 
public class SyncProducer {
	public static void main(String[] args) throws Exception {
    	// 實(shí)例化消息生產(chǎn)者Producer
        DefaultMQProducer producer = new DefaultMQProducer("zhuzeqing-1");
    	// 設(shè)置NameServer的地址
    	producer.setNamesrvAddr("192.168.23.131:9876");
    	// 啟動(dòng)Producer實(shí)例
        producer.start();
    	for (int i = 0; i < 1; i++) {
    	    // 創(chuàng)建消息，并指定Topic，Tag和消息體
    	    Message msg = new Message("zhuzeqing-1-Topic" /* Topic */,
        	"TagA" /* Tag */,
        	("hello，rocketmq --" + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
        	);
        	// 發(fā)送消息到一個(gè)Broker
            SendResult sendResult = producer.send(msg);
            // 通過(guò)sendResult返回消息是否成功送達(dá)
            System.out.printf("%s%n", sendResult);
    	}
    	// 如果不再發(fā)送消息，關(guān)閉Producer實(shí)例。
    	producer.shutdown();
    }
}

消費(fèi)消息

import java.util.List;
 
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
 
public class Consumer {
 
	public static void main(String[] args) throws InterruptedException, MQClientException {
 
    	// 實(shí)例化消費(fèi)者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("zhuzeqing-1");
 
    	// 設(shè)置NameServer的地址
        consumer.setNamesrvAddr("192.168.23.131:9876");
 
    	// 訂閱一個(gè)或者多個(gè)Topic，以及Tag來(lái)過(guò)濾需要消費(fèi)的消息
        consumer.subscribe("zhuzeqing-1-Topic", "*");
    	// 注冊(cè)回調(diào)實(shí)現(xiàn)類來(lái)處理從broker拉取回來(lái)的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                // 標(biāo)記該消息已經(jīng)被成功消費(fèi)
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 啟動(dòng)消費(fèi)者實(shí)例
        consumer.start();
        System.out.printf("Consumer Started.%n");
	}
}