Docker安装RocketMQ以及使用

2023-12-30 09:24:04

一、RocketMQ简介

Apache RocketMQ是一个分布式消息传递和流媒体平台，具有低延迟，高性能和可靠性，万亿级容量和灵活的可伸缩性。它由四个部分组成：nameserver，broker，生产者和使用者。

二、docker安装RocketMQ

1、搜索RocketMQ镜像

docker search rocketmq

2、启动NameServer

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

3、启动broker

编辑broker.conf文件

vim /home/rocketmq/broker.conf

内容为:

# 所属集群名称，如果节点较多可以配置多个
brokerClusterName = DefaultCluster
#broker名称，master和slave使用相同的名称，表明他们的主从关系
brokerName = broker-a
#0表示Master，大于0表示不同的slave
brokerId = 0
#表示几点做消息删除动作，默认是凌晨4点
deleteWhen = 04
#在磁盘上保留消息的时长，单位是小时
fileReservedTime = 48
#有三个值：SYNC_MASTER，ASYNC_MASTER，SLAVE；同步和异步表示Master和Slave之间同步数据的机制；
brokerRole = ASYNC_MASTER
#刷盘策略，取值为：ASYNC_FLUSH，SYNC_FLUSH表示同步刷盘和异步刷盘；SYNC_FLUSH消息写入磁盘后才返回成功状态，ASYNC_FLUSH不需要；
flushDiskType = ASYNC_FLUSH
# 设置broker节点所在服务器的ip地址
brokerIP1 = 192.168.220.76

执行命令:

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

4、启动rocketmq console

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

5、可视化页面,地址为 http://192.168.220.76:8080/

三、java使用RocketMQ

1、pom.xml中添加依赖

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.0.4</version>
</dependency>

2.创建生产者

// 1 创建消息生产者，指定生成组名
DefaultMQProducer defaultMQProducer = new DefaultMQProducer("dzp-producer-group");
// 2 指定NameServer的地址
defaultMQProducer.setNamesrvAddr("192.168.220.76:9876");
// 3 启动生产者
defaultMQProducer.start();
// 4 构建消息对象，主要是设置消息的主题、标签、内容
Message message = new Message("dzp-topic", "dzp-tag", "dzp-key", ("dzp测试消息发送").getBytes());
// 5 发送消息
SendResult result = defaultMQProducer.send(message);
System.out.println("SendResult-->" + result);
// 6 关闭生产者
defaultMQProducer.shutdown();

3、创建消费者

// 1 创建消费者，指定所属的消费者组名
DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer("dzp-consumer-group");
// 2 指定NameServer的地址
defaultMQPushConsumer.setNamesrvAddr("192.168.220.76:9876");
// 3 指定消费者订阅的主题和标签
defaultMQPushConsumer.subscribe("dzp-topic", "*");
// 4 进行订阅：注册回调函数，编写处理消息的逻辑
defaultMQPushConsumer.registerMessageListener((List<MessageExt> list, ConsumeConcurrentlyContext context) -> {
    //并且返回ConsumeConcurrentlyStatus.RECONSUME_LATER
    try {
      for (MessageExt messageExt : list) {

          String topic = messageExt.getTopic();
          System.out.println("topic-->" + topic);

          String tags = messageExt.getTags();
          System.out.println("tags-->" + tags);

          String keys = messageExt.getKeys();
          System.out.println("keys-->" + keys);

          String body = new String(messageExt.getBody());
          System.out.println("body-->" + body);
        }

    } catch (Throwable throwable) {
        throwable.printStackTrace();
    }

       return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    });

    // 5 启动消费者
    defaultMQPushConsumer.start();
}

码农公寓

一、RocketMQ简介

二、docker安装RocketMQ

三、java使用RocketMQ

相关文章