标签：RocketMq5.Dashboard；
一、简介
RocketMQ因其架构简单、业务功能丰富、具备极强可扩展性等特点被广泛应用，比如金融业务、互联网、大数据、物联网等领域的业务场景；
二、环境部署
1、编译打包
1、下载5.0版本源码包
rocketmq-all-5.0.0-source-release.zip
2、解压后进入目录，编译打包
mvn -Prelease-all -DskipTests -Dspotbugs.skip=true clean install -U
2、修改配置
在distribution/target/rocketmq-5.0.0/rocketmq-5.0.0/bin/runserver.sh
distribution/target/rocketmq-5.0.0/rocketmq-5.0.0/bin/runbroker.sh
3、服务启动
1、该目录下
distribution/target/rocketmq-5.0.0/rocketmq-5.0.0/bin/
2、启动NameServer
sh mqnamesrv
输出日志
The Name Server boot success. serializeType=JSON
3、启动Broker+Proxy
sh mqbroker -n localhost:9876 --enable-proxy
输出日志
rocketmq-proxy startup successfully
4、关闭服务
sh mqshutdown namesrv
Send shutdown request to mqnamesrv(18636) OK
sh mqshutdown broker
Send shutdown request to mqbroker with proxy enable OK(18647)
4、控制台安装
1、下载master源码包
rocketmq-dashboard-master
2、解压后进入目录，编译打包
mvn clean package -Dmaven.test.skip=true
3、启动服务
java -jar target/rocketmq-dashboard-1.0.1-SNAPSHOT.jar
4、输出日志
INFO main - Tomcat started on port(s): 8080 (http) with context path ''
5、访问服务：localhost:8080
三、工程搭建
1、工程结构
2、依赖管理
在rocketmq-starter组件中，实际上依赖的是rocketmq-client组件的5.0版本，由于两个新版框架间的兼容问题，需要添加相关配置解决该问题；
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>${rocketmq-starter.version}</version>
</dependency>
3、配置文件
配置RocketMq服务地址，消息生产者和消费者；
rocketmq:
name-server: 127.0.0.1:9876
# 生产者
producer:
group: boot_group_1
# 消息发送超时时间
send-message-timeout: 3000
# 消息最大长度4M
max-message-size: 4096
# 消息发送失败重试次数
retry-times-when-send-failed: 3
# 异步消息发送失败重试次数
retry-times-when-send-async-failed: 2
# 消费者
consumer:
group: boot_group_1
# 每次提取的最大消息数
pull-batch-size: 5
4、配置类
在配置类中主要定义两个Bean的加载，即RocketMQTemplate和DefaultMQProducer，主要是提供消息发送的能力，即生产消息；
@Configuration
public class RocketMqConfig {
@Value("${rocketmq.name-server}")
private String nameServer;
@Value("${rocketmq.producer.group}")
private String producerGroup;
@Value("${rocketmq.producer.send-message-timeout}")
private Integer sendMsgTimeout;
@Value("${rocketmq.producer.max-message-size}")
private Integer maxMessageSize;
@Value("${rocketmq.producer.retry-times-when-send-failed}")
private Integer retryTimesWhenSendFailed ;
@Value("${rocketmq.producer.retry-times-when-send-async-failed}")
private Integer retryTimesWhenSendAsyncFailed ;
@Bean
public RocketMQTemplate rocketMqTemplate(){
RocketMQTemplate rocketMqTemplate = new RocketMQTemplate();
rocketMqTemplate.setProducer(defaultMqProducer());
return rocketMqTemplate;
}
@Bean
public DefaultMQProducer defaultMqProducer() {
DefaultMQProducer producer = new DefaultMQProducer();
producer.setNamesrvAddr(this.nameServer);
producer.setProducerGroup(this.producerGroup);
producer.setSendMsgTimeout(this.sendMsgTimeout);
producer.setMaxMessageSize(this.maxMessageSize);
producer.setRetryTimesWhenSendFailed(this.retryTimesWhenSendFailed);
producer.setRetryTimesWhenSendAsyncFailed(this.retryTimesWhenSendAsyncFailed);
return producer;
}
}
四、基础用法
1、消息生产
编写一个生产者接口类，分别使用RocketMQTemplate和DefaultMQProducer实现消息发送的功能，然后可以通过Dashboard控制面板查看消息详情；
@RestController
public class ProducerWeb {
private static final Logger log = LoggerFactory.getLogger(ProducerWeb.class);
@Autowired
private RocketMQTemplate rocketMqTemplate;
@GetMapping("/send/msg1")
public String sendMsg1 (){
try {
// 构建消息主体
JsonMapper jsonMapper = new JsonMapper();
String msgBody = jsonMapper.writeValueAsString(new MqMsg(1,"boot_mq_msg"));
// 发送消息
rocketMqTemplate.convertAndSend("boot-mq-topic",msgBody);
} catch (Exception e) {
e.printStackTrace();
}
return "OK" ;
}
@Autowired
private DefaultMQProducer defaultMqProducer ;
@GetMapping("/send/msg2")
public String sendMsg2 (){
try {
// 构建消息主体
JsonMapper jsonMapper = new JsonMapper();
String msgBody = jsonMapper.writeValueAsString(new MqMsg(2,"boot_mq_msg"));
// 构建消息对象
Message message = new Message();
message.setTopic("boot-mq-topic");
message.setTags("boot-mq-tag");
message.setKeys("boot-mq-key");
message.setBody(msgBody.getBytes());
// 发送消息，打印日志
SendResult sendResult = defaultMqProducer.send(message);
log.info("msgId:{},sendStatus:{}",sendResult.getMsgId(),sendResult.getSendStatus());
} catch (Exception e) {
e.printStackTrace();
}
return "OK" ;
}
}
2、消息消费
编写消息监听类，实现RocketMQListener接口，通过RocketMQMessageListener注解控制监听的具体信息；
@Service
@RocketMQMessageListener(consumerGroup = "boot_group_1",topic = "boot-mq-topic")
public class ConsumerListener implements RocketMQListener<String> {
private static final Logger log = LoggerFactory.getLogger(ConsumerListener.class);
@Override
public void onMessage(String message) {
log.info("n=====n message：{} n=====n",message);
}
}
五、参考源码
文档仓库：
https://gitee.com/cicadasmile/butte-java-note
源码仓库：
https://gitee.com/cicadasmile/butte-spring-parent
内容来源于网络如有侵权请私信删除
文章来源: 博客园
- 还没有人评论，欢迎说说您的想法！
客服
