目录

一、RocketMQ简介

二、docker安装RocketMQ

 三、java使用RocketMQ 

一、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中添加依赖


org.apache.rocketmq
rocketmq-spring-boot-starter
2.0.4


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 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();
}