引入依赖
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.1.1</version>
</dependency>
1. 定义RocketMQTemplateService工具类
@Service
public class RocketMQTemplateService {
private static Logger logger = LoggerFactory.getLogger(RocketMQTemplateService.class);
@Autowired
private RocketMQTemplate rocketMQTemplate;
/**
* @Description: 功能描述
* 当发送的消息不重要时,采用OneWay方式,以提升吞吐量,一般用户日志存储
*
* @param: 参数描述
* @param destination
* @param payload
* @throws: 异常描述
*/
public void sendOneWay(String destination, Object payload) {
rocketMQTemplate.sendOneWay(destination, payload);
}
/**
* @Description: 功能描述
* 默认使用同步发送syncSend, 但拿不到回执;convertAndSend和send等价
*
* @param: 参数描述
* @param destination
* @param payload
* @throws: 异常描述
*
*/
public void convertAndSend(String destination, Object payload) {
rocketMQTemplate.convertAndSend(destination, payload);
}
/**
* @Description: 功能描述
* 同步发送,需设置延迟等级
*
* @param: 参数描述
* @param destination
* @param message
* @param timeout
* @param delayLevel
* @return
* @throws: 异常描述
*
*/
public SendResult syncSend(String destination, Object message, long timeout, int delayLevel) {
return rocketMQTemplate.syncSend(destination, MessageBuilder.withPayload(message).build(), timeout, delayLevel);
}
/**
* @Description: 功能描述
* 异步发送,需设置延迟等级
*
* @param: 参数描述
* @param destination
* @param message
* @param timeout
* @param delayLevel
* @throws: 异常描述
*
*/
public void asyncSend(String destination, Object message, long timeout, int delayLevel) {
rocketMQTemplate.asyncSend(destination, MessageBuilder.withPayload(message).build(), new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
}
@Override
public void onException(Throwable e) {
logger.error("rocketmq发送异步消息异常:destination = " + destination + ";timeout = " + timeout
+ ";delayLevel = " + delayLevel + ";message = " + message + ";error = " + e.getMessage());
}
}, timeout, delayLevel);
}
/**
* @Description: 功能描述
* 异步发送,需设置延迟等级
*
* @param: 参数描述
* @param destination
* @param message
* @param sendCallback
* @param timeout
* @param delayLevel
* @throws: 异常描述
*
*/
public void asyncSend(String destination, Object message, SendCallback sendCallback, long timeout, int delayLevel) {
rocketMQTemplate.asyncSend(destination, MessageBuilder.withPayload(message).build(), sendCallback, timeout, delayLevel);
}
部分适用场景
当发送的消息不重要时,采用one-way方式,以提升吞吐量
当发送的消息很重要时,且对响应时间不敏感的时候采用sync方式
当发送的消息很重要时,且对响应时间很是敏感的时候采用async方式
2. 使用
@Autowired
private RocketMQTemplateService rocketMQTemplateService;
public void 方法名(参数 参数) {
// 以异步消息为例
rocketMQTemplateService.asyncSend("TopicDetailId", 参数, 300000, RocketMqDelayLevel.getDelayLevel(30));
}
监听类文章来源:https://www.toymoban.com/news/detail-809357.html
@Component
@RocketMQMessageListener(topic = BizRocketMq.TopicDetailId,
consumeMode = ConsumeMode.CONCURRENTLY,//分为:有序模式和无序模式,设置为无序模式(并发模式)
messageModel = MessageModel.CLUSTERING,//分为:集群模式和广播模式;设置为集群模式;广播模式只能与无序模式匹配设置,并且广播模式只执行一次,切记切记!
consumerGroup = BizRocketMq.GroupDetailId)
public class JiFenExChangeServiceListener implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {
@Override
public void onMessage(MessageExt messageExt) {
你的业务逻辑
}
}
消息中两id必须一致文章来源地址https://www.toymoban.com/news/detail-809357.html
到了这里,关于RockerMQ发送消息流程的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!