概述
rocketmq消息组件在springboot框架中的应用,使用rocketmq的整合包进行编码实现。
编码参考
- 引入rocketmq依赖jar, 最新版本参考GitHub - apache/rocketmq-spring: Apache RocketMQ Spring Integration
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.2.3</version>
</dependency>
- 基本配置
rocketmq:
name-server: 192.168.15.175:9876;192.168.15.77:9876
producer:
group: ta-cipher-encode
rocketmq.name-server: rocketmq集群地址,单点或集群
rocketmq.producer.group : 生产者组名,用于标识一组具有相同功能的生产者
- 其他配置
Name Server相关配置:
rocketmq.namesrv.domain:Name Server的域名,用于自动发现Name Server地址。
Producer相关配置:
rocketmq.producer.group:生产者组名,用于标识一组具有相同功能的生产者。
rocketmq.producer.send-msg-timeout:发送消息的超时时间,默认为3秒。
rocketmq.producer.compress-msg-body-over-howmuch:消息体大于指定字节大小时启用压缩。
Consumer相关配置:
rocketmq.consumer.group:消费者组名,用于标识一组具有相同功能的消费者。
rocketmq.consumer.consume-thread-min:消费者线程池的最小线程数。
rocketmq.consumer.consume-thread-max:消费者线程池的最大线程数。
rocketmq.consumer.consume-message-batch-max-size:批量消费消息时每次拉取的最大消息数量。
rocketmq.consumer.pull-interval:拉取消息间隔时间,默认为0,表示尽可能快地拉取消息。
Message相关配置:
rocketmq.message.max-size:消息的最大大小,默认为4MB。
rocketmq.message.compress-level:消息压缩级别,可选值为0(不压缩)到9(最高压缩率)。
rocketmq.message.timeout:消息的过期时间,默认为3天。
集群模式相关配置:
rocketmq.broker.cluster.name:Broker集群的名称。
rocketmq.broker.cluster.slave-read-only:Slave节点是否只读,默认为true。文章来源:https://www.toymoban.com/news/detail-658499.html
- 生产端代码
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
@Slf4j
public class RmqProvdier{
@Autowired
private RocketMQTemplate rocketMQTemplate;
public boolean send(String message) {
try {
// 发送消息
rocketMQTemplate.convertAndSend("ta-cipher-persist", message);
} catch (Exception e) {
log.error("send message:{}", message, e);
return false;
}
return true;
}
}
- 消费端代码
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
@Component
@Slf4j
@RocketMQMessageListener(topic = "ta-cipher-persist", consumerGroup = "ta-cipher-encode")
public class RocketMQConsumer implements RocketMQListener<String> {
@Override
public void onMessage(String message) {
try {
log.info("message:{}", message);
} catch (Exception e) {
log.error("errorMessage:{}", message);
// 抛出异常会重新消费消息
throw new RuntimeException("Message processing failed", e);
}
}
}
@RocketMQMessageListener参数文章来源地址https://www.toymoban.com/news/detail-658499.html
- topic 消费这从那个topic中读取消息
- consumerGroup 当前消费者的消费组
到了这里,关于rocketmq消息注解基于springboot的简单应用及默认详细配置的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!