Docker 安裝 RocketMQ及簡單使用
檢索鏡像
docker search rocketmq
顯示以下信息：
NAME DESCRIPTION STARS OFFICIAL AUTOMATED
rocketmqinc/rocketmq Image repository for Apache RocketMQ 35
styletang/rocketmq-console-ng rocketmq-console-ng 27
foxiswho/rocketmq rocketmq 23
laoyumi/rocketmq 10 [OK]
apacherocketmq/rocketmq Docker Image for Apache RocketMQ 8
xlxwhy/rocketmq alibaba's rocketmq 4
rocketmqinc/rocketmq-namesrv Customized RocketMQ Name Server Image for Ro… 2
huanwei/rocketmq-broker 2
rocketmqinc/rocketmq-broker Customized RocketMQ Broker Image for RocketM… 1
apacherocketmq/rocketmq-operator RocketMQ Operator is to manage RocketMQ serv… 1
pangliang/rocketmq-console-ng 1
2019liurui/rocketmq-namesrv RocketMQ name service image for RocketMQ-Ope… 1
2019liurui/rocketmq-broker RocketMQ broker image for RocketMQ-Operator 1
coder4/rocketmq rocketmq 1 [OK]
pengzu/rocketmq-console-ng web console for rocketmq ,this code is from … 0
huanwei/rocketmq 0
huanwei/rocketmq-broker-k8s 0
rocketmqinc/rocketmq-operator The Kubernetes operator for RocketMQ 0
king019/rocketmq rocketmq 0
slpcat/rocketmq-console-ng 0
2019liurui/rocketmq-operator Kubernetes Operator for RocketMQ ! 0
huanwei/rocketmq-operator 0
fengzt/rocketmq-broker apache rocketmq 4.2.0 broker server(官方文檔… 0
fengzt/rocketmq-nameserver apache rocketmq 4.2.0 nameserver 0
slpcat/rocketmq 0
檢索具體版本
我就隨便選一個吧，如foxiswho/rocketmq，以下是一個查看當前鏡像所有的版本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)}}'
顯示以下信息：
% Total % Received % Xferd Average Speed Time Time Time Current
Dload Upload Total Spent Left Speed
100 934 0 934 0 0 310 0 --:--:-- 0:00:03 --:--:-- 310
foxiswho/rocketmq:4.7.0
foxiswho/rocketmq:base
foxiswho/rocketmq:base-4.3.2
foxiswho/rocketmq:base-4.4.0
foxiswho/rocketmq:base-4.5.0
foxiswho/rocketmq:base-4.5.1
foxiswho/rocketmq:base-4.5.2
foxiswho/rocketmq:base-4.6.1
foxiswho/rocketmq:broker
foxiswho/rocketmq:broker-4.3.2
foxiswho/rocketmq:broker-4.4.0
foxiswho/rocketmq:broker-4.5.0
foxiswho/rocketmq:broker-4.5.1
foxiswho/rocketmq:broker-4.5.2
foxiswho/rocketmq:broker-4.6.1
foxiswho/rocketmq:broker-4.7.0
foxiswho/rocketmq:server
foxiswho/rocketmq:server-4.3.2
foxiswho/rocketmq:server-4.4.0
foxiswho/rocketmq:server-4.5.0
foxiswho/rocketmq:server-4.5.1
foxiswho/rocketmq:server-4.5.2
foxiswho/rocketmq:server-4.6.1
foxiswho/rocketmq:server-4.7.0
啟動NameServer
docker run -d -p 9876:9876 --name rmqserver foxiswho/rocketmq:server-4.5.1
啟動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容器中默認的配置文件的路徑為：
/etc/rocketmq/broker.conf
也可以通過-v參數指定本機的配置文件：
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" -v /conf/broker.conf:/etc/rocketmq/broker.conf foxiswho/rocketmq:broker-4.5.1
-
注意事項：
broker.conf的配置內容如下：
brokerClusterName = DefaultCluster brokerName = broker-a brokerId = 0 deleteWhen = 04 fileReservedTime = 48 brokerRole = ASYNC_MASTER flushDiskType = ASYNC_FLUSH在最后一行后面新增
brokerIP1 = 192.168.3.27ip改成你的linux宿主機的ip
安裝 rocketmq console
如果一切正常，NameServer和Broker一會兒就會安裝好，為了管理上的方便，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
然后通過如下命令檢查一下啟動情況：
docker ps|grep rocketmq
結果如下：
614a3d0ed22b styletang/rocketmq-console-ng "sh -c 'java $JAVA_O…" 2 hours ago Up 2 hours 0.0.0.0:8180->8080/tcp rmqconsole
a76891fbfd35 foxiswho/rocketmq:broker-4.5.1 "/bin/sh -c 'cd ${RO…" 2 hours ago Up 16 minutes 0.0.0.0:10909->10909/tcp, 0.0.0.0:10911->10911/tcp rmqbroker
01d29f8f68c8 foxiswho/rocketmq:server-4.5.1 "/bin/sh -c 'cd ${RO…" 2 hours ago Up 2 hours 0.0.0.0:9876->9876/tcp
-
其他docker命令：
重啟網絡
systemctl restart network停止、刪除所有docker容器
docker stop $(docker ps -a -q) #停止所有docker容器 docker rm $(docker ps -a -q) #刪除所有
簡單Java示例
maven依賴
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.5.0</version>
</dependency>
生產者
import org.apache.rocketmq.client.exception.MQClientException;
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 Producer {
public static void main(String[] args) throws MQClientException, InterruptedException {
// 需要一個producer group名字作為構造方法的參數，這里為producer1
DefaultMQProducer producer = new DefaultMQProducer("producer1");
// 設置NameServer地址,此處應改為實際NameServer地址，多個地址之間用；分隔
// NameServer的地址必須有，但是也可以通過環境變量的方式設置，不一定非得寫死在代碼里
producer.setNamesrvAddr("192.168.3.27:9876");
producer.setVipChannelEnabled(false);
// 為避免程序啟動的時候報錯，添加此代碼，可以讓rocketMq自動創建topickey
producer.setCreateTopicKey("AUTO_CREATE_TOPIC_KEY");
producer.start();
for (int i = 0; i < 10; i++) {
try {
// topic 主題名稱
// pull 臨時值 在消費者消費的時候 可以根據msg類型進行消費
// body 內容
Message message = new Message("producer-topic", "msg", ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
SendResult sendResult = producer.send(message);
System.out.println("發送的消息ID:" + sendResult.getMsgId() + "--- 發送消息的狀態：" + sendResult.getSendStatus());
} catch (Exception e) {
e.printStackTrace();
Thread.sleep(1000);
}
}
producer.shutdown();
}
}
消費者
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.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
public class Consumer {
private static final String ADDR = "192.168.3.27:9876";
public static void main(String[] args) throws MQClientException {
// 設置消費者組
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group");
consumer.setVipChannelEnabled(false);
consumer.setNamesrvAddr(ADDR);
// 設置消費者端消息拉取策略，表示從哪里開始消費
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
// 設置消費者拉取消息的策略，*表示消費該topic下的所有消息，也可以指定tag進行消息過濾
consumer.subscribe("producer-topic", "msg");
// 消費者端啟動消息監聽，一旦生產者發送消息被監聽到，就打印消息，和rabbitmq中的handlerDelivery類似
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
for (MessageExt messageExt : msgs) {
String topic = messageExt.getTopic();
String tag = messageExt.getTags();
String msg = new String(messageExt.getBody());
System.out.println("*********************************");
System.out.println("消費響應：msgId : " + messageExt.getMsgId() + ", msgBody : " + msg + ", tag:" + tag + ", topic:" + topic);
System.out.println("*********************************");
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
// 調用start()方法啟動consumer
consumer.start();
System.out.println("Consumer Started....");
}
}