docker安装RocketMQ的实现步骤

Anne ·

更新时间:2024-11-14

· 1335 次阅读

一、检索镜像

二、创建Broker Server

三、创建broker

四、创建rocketmq console

五、测试

六、java样例

七、其他

参考：

一、检索镜像


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)}}'

查看当前镜像所有的版本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)}}'

二、创建Broker Server


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.conf

进入rmqbroker容器


docker exec -it rmqbroker /bin/bash
cd /etc/rocketmq/
vi broker.conf

四、创建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

http://192.168.23.131:8180/

五、测试

由于我在宿主机安装rocketmq,测试如下：


#测试消息发送命令
sh tools.sh org.apache.rocketmq.example.quickstart.Producer


#测试消息接收命令
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>

消息发送（Producer端发送同步消息）


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 {
    // 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("zhuzeqing-1");
    // 设置NameServer的地址
    producer.setNamesrvAddr("192.168.23.131:9876");
    // 启动Producer实例
        producer.start();
    for (int i = 0; i < 1; i++) {
        // 创建消息，并指定Topic，Tag和消息体
        Message msg = new Message("zhuzeqing-1-Topic" /* Topic */,
        "TagA" /* Tag */,
        ("hello，rocketmq --" + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
        );
        // 发送消息到一个Broker
            SendResult sendResult = producer.send(msg);
            // 通过sendResult返回消息是否成功送达
            System.out.printf("%s%n", sendResult);
    }
    // 如果不再发送消息，关闭Producer实例。
    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.message.MessageExt;
public class Consumer {
public static void main(String[] args) throws InterruptedException, MQClientException {
    // 实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("zhuzeqing-1");
    // 设置NameServer的地址
        consumer.setNamesrvAddr("192.168.23.131:9876");
    // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
        consumer.subscribe("zhuzeqing-1-Topic", "*");
    // 注册回调实现类来处理从broker拉取回来的消息
        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);
                // 标记该消息已经被成功消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者实例
        consumer.start();
        System.out.printf("Consumer Started.%n");
}
}

七、其他

查看rmqbroker IP