L'intégration de RocketMQ dans SpringBoot ne nécessite que quatre étapes simples :
1. Introduisez les dépendances pertinentes
<dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-spring-boot-starter</artifactId> </dependency>
2 Ajoutez les configurations pertinentes de RocketMQ
rocketmq:
consumer:
group: springboot_consumer_group
# 一次拉取消息最大值，注意是拉取消息的最大值而非消费最大值
pull-batch-size: 10
name-server: 10.5.103.6:9876
producer:
# 发送同一类消息的设置为同一个group，保证唯一
group: springboot_producer_group
# 发送消息超时时间，默认3000
sendMessageTimeout: 10000
# 发送消息失败重试次数，默认2
retryTimesWhenSendFailed: 2
# 异步消息重试此处，默认2
retryTimesWhenSendAsyncFailed: 2
# 消息最大长度，默认1024 * 1024 * 4(默认4M)
maxMessageSize: 4096
# 压缩消息阈值，默认4k(1024 * 4)
compressMessageBodyThreshold: 4096
# 是否在内部发送失败时重试另一个broker，默认false
retryNextServer: false3. Utilisez la classe d'outils de modèle fournie RocketMQTemplate pour envoyer des messages.
@RestController
public class NormalProduceController {
@Setter(onMethod_ = @Autowired)
private RocketMQTemplate rocketmqTemplate;
@GetMapping("/test")
public SendResult test() {
Message<String> msg = MessageBuilder.withPayload("Hello,RocketMQ").build();
SendResult sendResult = rocketmqTemplate.send(topic, msg);
}
}4. Implémentez l'interface RocketMQListener pour consommer des messages
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
@Component
@RocketMQMessageListener(topic = "your_topic_name", consumerGroup = "your_consumer_group_name")
public class MyConsumer implements RocketMQListener<String> {
@Override
public void onMessage(String message) {
// 处理消息的逻辑
System.out.println("Received message: " + message);
}
}Les 4 étapes ci-dessus peuvent réaliser l'intégration de SpringBoot et RocketMQ. Cette partie est une connaissance de base et ne sera pas trop expliquée.
Voici quelques problèmes couramment rencontrés lors de l'utilisation de RocketMQ dans SpringBoot. Nous allons maintenant les résoudre un par un pour vous.
Vous verrez l'alarme suivante dans le journal lors du démarrage du projet
RocketMQLog : WARN Aucun appender n'a pu être trouvé pour l'enregistreur (io.netty.util.internal.InternalThreadLocalMap) .
RocketMQLog : WARN Veuillez initialiser correctement le système d'enregistrement.
Pour le moment, il nous suffit de définir la variable d'environnement rocketmq.client.logUseSlf4j sur true dans la classe de démarrage pour spécifier clairement la journalisation RocketMQ. frameworkrocketmq.client.logUseSlf4j 为 true 明确指定RocketMQ的日志框架
@SpringBootApplication
public class RocketDemoApplication {
public static void main(String[] args) {
/*
* 指定使用的日志框架，否则将会告警
* RocketMQLog:WARN No appenders could be found for logger (io.netty.util.internal.InternalThreadLocalMap).
* RocketMQLog:WARN Please initialize the logger system properly.
*/
System.setProperty("rocketmq.client.logUseSlf4j", "true");
SpringApplication.run(RocketDemoApplication.class, args);
}
}同时还得在配置文件中调整日志级别，不然在控制台会一直看到broker的日志信息
logging:
level:
RocketmqClient: ERROR
io:
netty: ERROR
在使用Java8后经常会使用LocalDate/LocalDateTime这两个时间类型字段，然而RocketMQ原始配置并不支持Java时间类型，当我们发送的实体消息中包含上述两个字段时，消费端在消费时会出现如下所示的错误。
比如生产者的代码如下：
@GetMapping("/test")
public void test(){
//普通消息无返回值，只负责发送消息⽽不等待服务器回应且没有回调函数触发。
RocketMessage rocketMessage = RocketMessage.builder().
id(1111L).
message("hello,world")
.localDate(LocalDate.now())
.localDateTime(LocalDateTime.now())
.build();
rocketmqTemplate.convertAndSend(destination,rocketMessage);
}消费者的代码如下：
@Component
@RocketMQMessageListener(consumerGroup = "springboot_consumer_group",topic = "consumer_topic")
public class RocketMQConsumer implements RocketMQListener<RocketMessage> {
@Override
public void onMessage(RocketMessage message) {
System.out.println("消费消息-" + message);
}
}消费者开始消费时会出现类型转换异常错误Cannot construct instance of java.time.LocalDate，错误详情如下：
原因：RocketMQ内置使用的转换器是RocketMQMessageConverter，转换Json时使用的是MappingJackson2MessageConverter，但是这个转换器不支持时间类型。
解决办法：需要自定义消息转换器，将MappingJackson2MessageConverter进行替换，并添加支持时间模块
@Configuration
public class RocketMQEnhanceConfig {
/**
* 解决RocketMQ Jackson不支持Java时间类型配置
* 源码参考：{@link org.apache.rocketmq.spring.autoconfigure.MessageConverterConfiguration}
*/
@Bean
@Primary
public RocketMQMessageConverter enhanceRocketMQMessageConverter(){
RocketMQMessageConverter converter = new RocketMQMessageConverter();
CompositeMessageConverter compositeMessageConverter = (CompositeMessageConverter) converter.getMessageConverter();
List<MessageConverter> messageConverterList = compositeMessageConverter.getConverters();
for (MessageConverter messageConverter : messageConverterList) {
if(messageConverter instanceof MappingJackson2MessageConverter){
MappingJackson2MessageConverter jackson2MessageConverter = (MappingJackson2MessageConverter) messageConverter;
ObjectMapper objectMapper = jackson2MessageConverter.getObjectMapper();
objectMapper.registerModules(new JavaTimeModule());
}
}
return converter;
}
}在使用RocketMQ时，通常会在代码中直接指定消息主题(topic)，而且开发环境和测试环境可能共用一个RocketMQ环境。如果没有进行处理，在开发环境发送的消息就可能被测试环境的消费者消费，测试环境发送的消息也可能被开发环境的消费者消费，从而导致数据混乱的问题。
为了解决这个问题，我们可以根据不同的环境实现自动隔离。通过简单配置一个选项，如dev、test、prod等不同环境，所有的消息都会被自动隔离。例如，当发送的消息主题为consumer_topic时，可以自动在topic后面加上环境后缀，如consumer_topic_dev
rocketmq: enhance: # 启动隔离，用于激活配置类EnvironmentIsolationConfig # 启动后会自动在topic上拼接激活的配置文件，达到自动隔离的效果 enabledIsolation: true # 隔离环境名称，拼接到topic后，topic_dev，默认空字符串 environment: dev
logging:
niveau:netty: ERREUR
Après avoir utilisé Java8, les deux champs de type d'heure
LocalDate/LocalDateTime sont souvent utilisés. Cependant, la configuration d'origine de RocketMQ ne prend pas en charge le type d'heure Java. .Lorsque le message d'entité que nous envoyons contient les deux champs ci-dessus, l'erreur suivante se produira du côté du consommateur lors de la consommation. @Configuration
public class EnvironmentIsolationConfig implements BeanPostProcessor {
@Value("${rocketmq.enhance.enabledIsolation:true}")
private boolean enabledIsolation;
@Value("${rocketmq.enhance.environment:''}")
private String environmentName;
/**
* 在装载Bean之前实现参数修改
*/
@Override
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
if(bean instanceof DefaultRocketMQListenerContainer){
DefaultRocketMQListenerContainer container = (DefaultRocketMQListenerContainer) bean;
//拼接Topic
if(enabledIsolation && StringUtils.hasText(environmentName)){
container.setTopic(String.join("_", container.getTopic(),environmentName));
}
return container;
}
return bean;
}
}Le code du consommateur est le suivant :
/**
* 消息实体，所有消息都需要继承此类
* 公众号：JAVA日知录
*/
@Data
public abstract class BaseMessage {
/**
* 业务键，用于RocketMQ控制台查看消费情况
*/
protected String key;
/**
* 发送消息来源，用于排查问题
*/
protected String source = "";
/**
* 发送时间
*/
protected LocalDateTime sendTime = LocalDateTime.now();
/**
* 重试次数，用于判断重试次数，超过重试次数发送异常警告
*/
protected Integer retryTimes = 0;
}Impossible de construire une instance de java.time.LocalDate< /code>, erreur Les détails sont les suivants : <li><p><img src="https://img.php.cn/upload/article/000/887/227/168403074968475.jpg" alt="Quelle est la méthode pour SpringBoot pour intégrer RocketMQ" /></p></li>Raison : Le convertisseur intégré utilisé par RocketMQ est RocketMQMessageConverter, et MappingJackson2MessageConverter est utilisé lors de la conversion de Json, mais ce convertisseur ne prend pas en charge les types d'heure. </ul><li>Solution : vous devez personnaliser le convertisseur de messages, remplacer MappingJackson2MessageConverter et ajouter un module de temps de support<p><div class="code" style="position:relative; padding:0px; margin:0px;"><div class="code" style="position:relative; padding:0px; margin:0px;"><pre class="brush:java;">@Slf4j
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class RocketMQEnhanceTemplate {
private final RocketMQTemplate template;
@Resource
private RocketEnhanceProperties rocketEnhanceProperties;
public RocketMQTemplate getTemplate() {
return template;
}
/**
* 根据系统上下文自动构建隔离后的topic
* 构建目的地
*/
public String buildDestination(String topic, String tag) {
topic = reBuildTopic(topic);
return topic + ":" + tag;
}
/**
* 根据环境重新隔离topic
* @param topic 原始topic
*/
private String reBuildTopic(String topic) {
if(rocketEnhanceProperties.isEnabledIsolation() && StringUtils.hasText(rocketEnhanceProperties.getEnvironment())){
return topic +"_" + rocketEnhanceProperties.getEnvironment();
}
return topic;
}
/**
* 发送同步消息
*/
public <T extends BaseMessage> SendResult send(String topic, String tag, T message) {
// 注意分隔符
return send(buildDestination(topic,tag), message);
}
public <T extends BaseMessage> SendResult send(String destination, T message) {
// 设置业务键，此处根据公共的参数进行处理
// 更多的其它基础业务处理...
Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, message.getKey()).build();
SendResult sendResult = template.syncSend(destination, sendMessage);
// 此处为了方便查看给日志转了json，根据选择选择日志记录方式，例如ELK采集
log.info("[{}]同步消息[{}]发送结果[{}]", destination, JSONObject.toJSON(message), JSONObject.toJSON(sendResult));
return sendResult;
}
/**
* 发送延迟消息
*/
public <T extends BaseMessage> SendResult send(String topic, String tag, T message, int delayLevel) {
return send(buildDestination(topic,tag), message, delayLevel);
}
public <T extends BaseMessage> SendResult send(String destination, T message, int delayLevel) {
Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, message.getKey()).build();
SendResult sendResult = template.syncSend(destination, sendMessage, 3000, delayLevel);
log.info("[{}]延迟等级[{}]消息[{}]发送结果[{}]", destination, delayLevel, JSONObject.toJSON(message), JSONObject.toJSON(sendResult));
return sendResult;
}
}</pre><div class="contentsignin">Copier après la connexion</div></div><div class="contentsignin">Copier après la connexion</div></div></p>2.3 Isolation de l'environnement RockeMQ</li></ul>Lors de l'utilisation de RocketMQ, le sujet du message (sujet) est généralement spécifié directement dans le code, et le développement L'environnement et l'environnement de test peuvent partager un environnement RocketMQ. S'ils ne sont pas traités, les messages envoyés dans l'environnement de développement peuvent être consommés par les consommateurs dans l'environnement de test, et les messages envoyés dans l'environnement de test peuvent également être consommés par les consommateurs dans l'environnement de développement, entraînant une confusion des données. <p></p>Afin de résoudre ce problème, nous pouvons mettre en place une isolation automatique en fonction de différents environnements. En configurant simplement une option pour différents environnements tels que dev, test, prod, etc., tous les messages seront automatiquement isolés. Par exemple, lorsque le sujet du message envoyé est <code>consumer_topic, le suffixe d'environnement peut être automatiquement ajouté après le sujet, tel que consumer_topic_dev. Alors, comment y parvenir ? Vous pouvez écrire une classe de configuration pour implémenter BeanPostProcessor et remplacer la méthode postProcessBeforeInitialization pour modifier la rubrique correspondante avant l'initialisation de l'instance d'écoute. 🎜🎜BeanPostProcessor est une interface du framework Spring. Sa fonction est d'effectuer un travail de traitement supplémentaire avant et après l'initialisation du bean après l'instanciation et la configuration du conteneur Spring. 🎜🎜🎜🎜Plus précisément, l'interface BeanPostProcessor définit deux méthodes : 🎜🎜🎜🎜🎜postProcessBeforeInitialization(Object bean, String beanName) : traite avant l'initialisation du bean et vous pouvez apporter quelques modifications au bean. 🎜🎜🎜🎜postProcessAfterInitialization(Object bean, String beanName) : traité après l'initialisation du bean, certains nettoyages ou autres opérations peuvent être effectués. 🎜🎜🎜🎜🎜BeanPostProcessor peut intercepter et modifier le processus de création et d'initialisation de Bean dans l'application, et intervenir et exploiter le cycle de vie de Bean. Il peut améliorer toutes les instances de classe Bean, permettant aux développeurs de personnaliser certaines opérations avant et après l'initialisation de Bean pour répondre à leurs propres besoins commerciaux. Par exemple, vous pouvez utiliser BeanPostProcessor pour injecter certaines valeurs d'attribut nécessaires, ajouter un objet, etc. 🎜🎜🎜🎜Le plan de mise en œuvre est le suivant : 🎜🎜1. Ajoutez les configurations pertinentes au fichier de configuration🎜@Slf4j
public abstract class EnhanceMessageHandler<T extends BaseMessage> {
/**
* 默认重试次数
*/
private static final int MAX_RETRY_TIMES = 3;
/**
* 延时等级
*/
private static final int DELAY_LEVEL = EnhanceMessageConstant.FIVE_SECOND;
@Resource
private RocketMQEnhanceTemplate rocketMQEnhanceTemplate;
/**
* 消息处理
*
* @param message 待处理消息
* @throws Exception 消费异常
*/
protected abstract void handleMessage(T message) throws Exception;
/**
* 超过重试次数消息，需要启用isRetry
*
* @param message 待处理消息
*/
protected abstract void handleMaxRetriesExceeded(T message);
/**
* 是否需要根据业务规则过滤消息，去重逻辑可以在此处处理
* @param message 待处理消息
* @return true: 本次消息被过滤，false：不过滤
*/
protected boolean filter(T message) {
return false;
}
/**
* 是否异常时重复发送
*
* @return true: 消息重试，false：不重试
*/
protected abstract boolean isRetry();
/**
* 消费异常时是否抛出异常
* 返回true，则由rocketmq机制自动重试
* false：消费异常(如果没有开启重试则消息会被自动ack)
*/
protected abstract boolean throwException();
/**
* 最大重试次数
*
* @return 最大重试次数，默认5次
*/
protected int getMaxRetryTimes() {
return MAX_RETRY_TIMES;
}
/**
* isRetry开启时，重新入队延迟时间
* @return -1：立即入队重试
*/
protected int getDelayLevel() {
return DELAY_LEVEL;
}
/**
* 使用模板模式构建消息消费框架，可自由扩展或删减
*/
public void dispatchMessage(T message) {
// 基础日志记录被父类处理了
log.info("消费者收到消息[{}]", JSONObject.toJSON(message));
if (filter(message)) {
log.info("消息id{}不满足消费条件，已过滤。",message.getKey());
return;
}
// 超过最大重试次数时调用子类方法处理
if (message.getRetryTimes() > getMaxRetryTimes()) {
handleMaxRetriesExceeded(message);
return;
}
try {
long now = System.currentTimeMillis();
handleMessage(message);
long costTime = System.currentTimeMillis() - now;
log.info("消息{}消费成功，耗时[{}ms]", message.getKey(),costTime);
} catch (Exception e) {
log.error("消息{}消费异常", message.getKey(),e);
// 是捕获异常还是抛出，由子类决定
if (throwException()) {
//抛出异常，由DefaultMessageListenerConcurrently类处理
throw new RuntimeException(e);
}
//此时如果不开启重试机制，则默认ACK了
if (isRetry()) {
handleRetry(message);
}
}
}
protected void handleRetry(T message) {
// 获取子类RocketMQMessageListener注解拿到topic和tag
RocketMQMessageListener annotation = this.getClass().getAnnotation(RocketMQMessageListener.class);
if (annotation == null) {
return;
}
//重新构建消息体
String messageSource = message.getSource();
if(!messageSource.startsWith(EnhanceMessageConstant.RETRY_PREFIX)){
message.setSource(EnhanceMessageConstant.RETRY_PREFIX + messageSource);
}
message.setRetryTimes(message.getRetryTimes() + 1);
SendResult sendResult;
try {
// 如果消息发送不成功，则再次重新发送，如果发送异常则抛出由MQ再次处理(异常时不走延迟消息)
sendResult = rocketMQEnhanceTemplate.send(annotation.topic(), annotation.selectorExpression(), message, getDelayLevel());
} catch (Exception ex) {
// 此处捕获之后，相当于此条消息被消息完成然后重新发送新的消息
//由生产者直接发送
throw new RuntimeException(ex);
}
// 发送失败的处理就是不进行ACK，由RocketMQ重试
if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
throw new RuntimeException("重试消息发送失败");
}
}
}@Configuration
@EnableConfigurationProperties(RocketEnhanceProperties.class)
public class RocketMQEnhanceAutoConfiguration {
/**
* 注入增强的RocketMQEnhanceTemplate
*/
@Bean
public RocketMQEnhanceTemplate rocketMQEnhanceTemplate(RocketMQTemplate rocketMQTemplate){
return new RocketMQEnhanceTemplate(rocketMQTemplate);
}
/**
* 解决RocketMQ Jackson不支持Java时间类型配置
* 源码参考：{@link org.apache.rocketmq.spring.autoconfigure.MessageConverterConfiguration}
*/
@Bean
@Primary
public RocketMQMessageConverter enhanceRocketMQMessageConverter(){
RocketMQMessageConverter converter = new RocketMQMessageConverter();
CompositeMessageConverter compositeMessageConverter = (CompositeMessageConverter) converter.getMessageConverter();
List<MessageConverter> messageConverterList = compositeMessageConverter.getConverters();
for (MessageConverter messageConverter : messageConverterList) {
if(messageConverter instanceof MappingJackson2MessageConverter){
MappingJackson2MessageConverter jackson2MessageConverter = (MappingJackson2MessageConverter) messageConverter;
ObjectMapper objectMapper = jackson2MessageConverter.getObjectMapper();
objectMapper.registerModules(new JavaTimeModule());
}
}
return converter;
}
/**
* 环境隔离配置
*/
@Bean
@ConditionalOnProperty(name="rocketmq.enhance.enabledIsolation", havingValue="true")
public EnvironmentIsolationConfig environmentSetup(RocketEnhanceProperties rocketEnhanceProperties){
return new EnvironmentIsolationConfig(rocketEnhanceProperties);
}
}2023-03-23 17:04:59.726 [main] INFO o.a.r.s.support.DefaultRocketMQListenerContainer:290 - running container: DefaultRocketMQListenerContainer{consumerGroup='springboot_consumer_group', nameServer='10.5.103.6:9876', topic='consumer_topic_dev', consumeMode=CONCURRENTLY, selectorType=TAG, selectorExpression='*', messageModel=CLUSTERING}
在解释为什么要二次封装之前先来看看RocketMQ官方文档中推荐的最佳实践
消息发送成功或者失败要打印消息日志，用于业务排查问题。
如果消息量较少，建议在消费入口方法打印消息，消费耗时等，方便后续排查问题。
RocketMQ 无法避免消息重复（Exactly-Once），所以如果业务对消费重复非常敏感，务必要在业务层面进行去重处理。可以借助关系数据库进行去重。首先需要确定消息的唯一键，可以是msgId，也可以是消息内容中的唯一标识字段，例如订单Id等。
上面三个步骤基本每次发送消息或者消费消息都要实现，属于重复动作。
接下来讨论的是在RocketMQ中发送消息时选择何种消息类型最为合适。
在RocketMQ中有四种可选格式：
发送Json对象
发送转Json后的String对象
根据业务封装对应实体类
直接使用原生MessageExt接收。
对于如何选择消息类型，需要考虑到消费者在不查看消息发送者的情况下，如何获取消息的含义。因此，在这种情况下，使用第三种方式即根据业务封装对应实体类的方式最为合适，也是大多数开发者在发送消息时的常用方式。
有了上面两点结论以后我们来看看为什么要对RocketMQ二次封装。
按照上述最佳实践，一个完整的消息传递链路从生产到消费应包括 准备消息、发送消息、记录消息日志、处理发送失败、记录接收消息日志、处理业务逻辑、异常处理和异常重试 等步骤。
虽然使用原生RocketMQ可以完成这些动作，但每个生产者和消费者都需要编写大量重复的代码来完成相同的任务，这就是需要进行二次封装的原因。我们希望通过二次封装，**生产者只需准备好消息实体并调用封装后的工具类发送，而消费者只需处理核心业务逻辑，其他公共逻辑会得到统一处理。 **
在二次封装中，关键是找出框架在日常使用中所涵盖的许多操作，以及区分哪些操作是可变的，哪些是不变的。以上述例子为例，实际上只有生产者的消息准备和消费者的业务处理是可变的操作，需要根据需求进行处理，而其他步骤可以固定下来形成一个模板。
当然，本文提到的二次封装不是指对源代码进行封装，而是针对工具的原始使用方式进行的封装。可以将其与Mybatis和Mybatis-plus区分开来。这两者都能完成任务，只不过Mybatis-plus更为简单便捷。
实现二次封装需要创建一个自定义的starter，这样其他项目只需要依赖此starter即可使用封装功能。同时，在自定义starter中还需要解决文章第二部分中提到的一些问题。
代码结构如下所示：
/**
* 消息实体，所有消息都需要继承此类
* 公众号：JAVA日知录
*/
@Data
public abstract class BaseMessage {
/**
* 业务键，用于RocketMQ控制台查看消费情况
*/
protected String key;
/**
* 发送消息来源，用于排查问题
*/
protected String source = "";
/**
* 发送时间
*/
protected LocalDateTime sendTime = LocalDateTime.now();
/**
* 重试次数，用于判断重试次数，超过重试次数发送异常警告
*/
protected Integer retryTimes = 0;
}后面所有发送的消息实体都需要继承此实体类。
@Slf4j
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class RocketMQEnhanceTemplate {
private final RocketMQTemplate template;
@Resource
private RocketEnhanceProperties rocketEnhanceProperties;
public RocketMQTemplate getTemplate() {
return template;
}
/**
* 根据系统上下文自动构建隔离后的topic
* 构建目的地
*/
public String buildDestination(String topic, String tag) {
topic = reBuildTopic(topic);
return topic + ":" + tag;
}
/**
* 根据环境重新隔离topic
* @param topic 原始topic
*/
private String reBuildTopic(String topic) {
if(rocketEnhanceProperties.isEnabledIsolation() && StringUtils.hasText(rocketEnhanceProperties.getEnvironment())){
return topic +"_" + rocketEnhanceProperties.getEnvironment();
}
return topic;
}
/**
* 发送同步消息
*/
public <T extends BaseMessage> SendResult send(String topic, String tag, T message) {
// 注意分隔符
return send(buildDestination(topic,tag), message);
}
public <T extends BaseMessage> SendResult send(String destination, T message) {
// 设置业务键，此处根据公共的参数进行处理
// 更多的其它基础业务处理...
Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, message.getKey()).build();
SendResult sendResult = template.syncSend(destination, sendMessage);
// 此处为了方便查看给日志转了json，根据选择选择日志记录方式，例如ELK采集
log.info("[{}]同步消息[{}]发送结果[{}]", destination, JSONObject.toJSON(message), JSONObject.toJSON(sendResult));
return sendResult;
}
/**
* 发送延迟消息
*/
public <T extends BaseMessage> SendResult send(String topic, String tag, T message, int delayLevel) {
return send(buildDestination(topic,tag), message, delayLevel);
}
public <T extends BaseMessage> SendResult send(String destination, T message, int delayLevel) {
Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, message.getKey()).build();
SendResult sendResult = template.syncSend(destination, sendMessage, 3000, delayLevel);
log.info("[{}]延迟等级[{}]消息[{}]发送结果[{}]", destination, delayLevel, JSONObject.toJSON(message), JSONObject.toJSON(sendResult));
return sendResult;
}
}这里封装了一个消息发送类，实现了日志记录以及自动重建topic的功能（即生产者实现环境隔离），后面项目中只需要注入RocketMQEnhanceTemplate来实现消息的发送。
@Slf4j
public abstract class EnhanceMessageHandler<T extends BaseMessage> {
/**
* 默认重试次数
*/
private static final int MAX_RETRY_TIMES = 3;
/**
* 延时等级
*/
private static final int DELAY_LEVEL = EnhanceMessageConstant.FIVE_SECOND;
@Resource
private RocketMQEnhanceTemplate rocketMQEnhanceTemplate;
/**
* 消息处理
*
* @param message 待处理消息
* @throws Exception 消费异常
*/
protected abstract void handleMessage(T message) throws Exception;
/**
* 超过重试次数消息，需要启用isRetry
*
* @param message 待处理消息
*/
protected abstract void handleMaxRetriesExceeded(T message);
/**
* 是否需要根据业务规则过滤消息，去重逻辑可以在此处处理
* @param message 待处理消息
* @return true: 本次消息被过滤，false：不过滤
*/
protected boolean filter(T message) {
return false;
}
/**
* 是否异常时重复发送
*
* @return true: 消息重试，false：不重试
*/
protected abstract boolean isRetry();
/**
* 消费异常时是否抛出异常
* 返回true，则由rocketmq机制自动重试
* false：消费异常(如果没有开启重试则消息会被自动ack)
*/
protected abstract boolean throwException();
/**
* 最大重试次数
*
* @return 最大重试次数，默认5次
*/
protected int getMaxRetryTimes() {
return MAX_RETRY_TIMES;
}
/**
* isRetry开启时，重新入队延迟时间
* @return -1：立即入队重试
*/
protected int getDelayLevel() {
return DELAY_LEVEL;
}
/**
* 使用模板模式构建消息消费框架，可自由扩展或删减
*/
public void dispatchMessage(T message) {
// 基础日志记录被父类处理了
log.info("消费者收到消息[{}]", JSONObject.toJSON(message));
if (filter(message)) {
log.info("消息id{}不满足消费条件，已过滤。",message.getKey());
return;
}
// 超过最大重试次数时调用子类方法处理
if (message.getRetryTimes() > getMaxRetryTimes()) {
handleMaxRetriesExceeded(message);
return;
}
try {
long now = System.currentTimeMillis();
handleMessage(message);
long costTime = System.currentTimeMillis() - now;
log.info("消息{}消费成功，耗时[{}ms]", message.getKey(),costTime);
} catch (Exception e) {
log.error("消息{}消费异常", message.getKey(),e);
// 是捕获异常还是抛出，由子类决定
if (throwException()) {
//抛出异常，由DefaultMessageListenerConcurrently类处理
throw new RuntimeException(e);
}
//此时如果不开启重试机制，则默认ACK了
if (isRetry()) {
handleRetry(message);
}
}
}
protected void handleRetry(T message) {
// 获取子类RocketMQMessageListener注解拿到topic和tag
RocketMQMessageListener annotation = this.getClass().getAnnotation(RocketMQMessageListener.class);
if (annotation == null) {
return;
}
//重新构建消息体
String messageSource = message.getSource();
if(!messageSource.startsWith(EnhanceMessageConstant.RETRY_PREFIX)){
message.setSource(EnhanceMessageConstant.RETRY_PREFIX + messageSource);
}
message.setRetryTimes(message.getRetryTimes() + 1);
SendResult sendResult;
try {
// 如果消息发送不成功，则再次重新发送，如果发送异常则抛出由MQ再次处理(异常时不走延迟消息)
sendResult = rocketMQEnhanceTemplate.send(annotation.topic(), annotation.selectorExpression(), message, getDelayLevel());
} catch (Exception ex) {
// 此处捕获之后，相当于此条消息被消息完成然后重新发送新的消息
//由生产者直接发送
throw new RuntimeException(ex);
}
// 发送失败的处理就是不进行ACK，由RocketMQ重试
if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
throw new RuntimeException("重试消息发送失败");
}
}
}使用模版设计模式定义了消息消费的骨架，实现了日志打印，异常处理，异常重试等公共逻辑，消息过滤（查重）、业务处理则交由子类实现。
@Configuration
@EnableConfigurationProperties(RocketEnhanceProperties.class)
public class RocketMQEnhanceAutoConfiguration {
/**
* 注入增强的RocketMQEnhanceTemplate
*/
@Bean
public RocketMQEnhanceTemplate rocketMQEnhanceTemplate(RocketMQTemplate rocketMQTemplate){
return new RocketMQEnhanceTemplate(rocketMQTemplate);
}
/**
* 解决RocketMQ Jackson不支持Java时间类型配置
* 源码参考：{@link org.apache.rocketmq.spring.autoconfigure.MessageConverterConfiguration}
*/
@Bean
@Primary
public RocketMQMessageConverter enhanceRocketMQMessageConverter(){
RocketMQMessageConverter converter = new RocketMQMessageConverter();
CompositeMessageConverter compositeMessageConverter = (CompositeMessageConverter) converter.getMessageConverter();
List<MessageConverter> messageConverterList = compositeMessageConverter.getConverters();
for (MessageConverter messageConverter : messageConverterList) {
if(messageConverter instanceof MappingJackson2MessageConverter){
MappingJackson2MessageConverter jackson2MessageConverter = (MappingJackson2MessageConverter) messageConverter;
ObjectMapper objectMapper = jackson2MessageConverter.getObjectMapper();
objectMapper.registerModules(new JavaTimeModule());
}
}
return converter;
}
/**
* 环境隔离配置
*/
@Bean
@ConditionalOnProperty(name="rocketmq.enhance.enabledIsolation", havingValue="true")
public EnvironmentIsolationConfig environmentSetup(RocketEnhanceProperties rocketEnhanceProperties){
return new EnvironmentIsolationConfig(rocketEnhanceProperties);
}
}public class EnvironmentIsolationConfig implements BeanPostProcessor {
private RocketEnhanceProperties rocketEnhanceProperties;
public EnvironmentIsolationConfig(RocketEnhanceProperties rocketEnhanceProperties) {
this.rocketEnhanceProperties = rocketEnhanceProperties;
}
/**
* 在装载Bean之前实现参数修改
*/
@Override
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
if(bean instanceof DefaultRocketMQListenerContainer){
DefaultRocketMQListenerContainer container = (DefaultRocketMQListenerContainer) bean;
if(rocketEnhanceProperties.isEnabledIsolation() && StringUtils.hasText(rocketEnhanceProperties.getEnvironment())){
container.setTopic(String.join("_", container.getTopic(),rocketEnhanceProperties.getEnvironment()));
}
return container;
}
return bean;
}
}@ConfigurationProperties(prefix = "rocketmq.enhance")
@Data
public class RocketEnhanceProperties {
private boolean enabledIsolation;
private String environment;
}<dependency> <groupId>com.jianzh6</groupId> <artifactId>cloud-rocket-starter</artifactId> </dependency>
rocketmq:
...
enhance:
# 启动隔离，用于激活配置类EnvironmentIsolationConfig
# 启动后会自动在topic上拼接激活的配置文件，达到自动隔离的效果
enabledIsolation: true
# 隔离环境名称，拼接到topic后，topic_dev，默认空字符串
environment: dev@RestController
@RequestMapping("enhance")
@Slf4j
public class EnhanceProduceController {
//注入增强后的模板，可以自动实现环境隔离，日志记录
@Setter(onMethod_ = @Autowired)
private RocketMQEnhanceTemplate rocketMQEnhanceTemplate;
private static final String topic = "rocket_enhance";
private static final String tag = "member";
/**
* 发送实体消息
*/
@GetMapping("/member")
public SendResult member() {
String key = UUID.randomUUID().toString();
MemberMessage message = new MemberMessage();
// 设置业务key
message.setKey(key);
// 设置消息来源，便于查询
message.setSource("MEMBER");
// 业务消息内容
message.setUserName("Java日知录");
message.setBirthday(LocalDate.now());
return rocketMQEnhanceTemplate.send(topic, tag, message);
}
}注意这里使用的是封装后的模板工具类，一旦在配置文件中启动环境隔离，则生产者的消息也自动发送到隔离后的topic中。
@Slf4j
@Component
@RocketMQMessageListener(
consumerGroup = "enhance_consumer_group",
topic = "rocket_enhance",
selectorExpression = "*",
consumeThreadMax = 5 //默认是64个线程并发消息，配置 consumeThreadMax 参数指定并发消费线程数，避免太大导致资源不够
)
public class EnhanceMemberMessageListener extends EnhanceMessageHandler<MemberMessage> implements RocketMQListener<MemberMessage> {
@Override
protected void handleMessage(MemberMessage message) throws Exception {
// 此时这里才是最终的业务处理，代码只需要处理资源类关闭异常，其他的可以交给父类重试
System.out.println("业务消息处理:"+message.getUserName());
}
@Override
protected void handleMaxRetriesExceeded(MemberMessage message) {
// 当超过指定重试次数消息时此处方法会被调用
// 生产中可以进行回退或其他业务操作
log.error("消息消费失败，请执行后续处理");
}
/**
* 是否执行重试机制
*/
@Override
protected boolean isRetry() {
return true;
}
@Override
protected boolean throwException() {
// 是否抛出异常，false搭配retry自行处理异常
return false;
}
@Override
protected boolean filter() {
// 消息过滤
return false;
}
/**
* 监听消费消息，不需要执行业务处理，委派给父类做基础操作，父类做完基础操作后会调用子类的实际处理类型
*/
@Override
public void onMessage(MemberMessage memberMessage) {
super.dispatchMessage(memberMessage);
}
}为了方便消费者对RocketMQ中的消息进行处理，我们可以使用EnhanceMessageHandler来进行消息的处理和逻辑的处理。
Pendant que les consommateurs implémentent RocketMQListener, ils peuvent hériter d'EnhanceMessageHandler pour traiter la logique publique, tandis que la logique métier de base doit implémenter la méthode handleMessage par elle-même. Si vous devez filtrer ou supprimer les messages en double, vous pouvez remplacer la méthode de filtrage de la classe parent pour l'implémenter. Cela facilite le traitement des messages et réduit la charge de travail des développeurs.
Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!
Étapes de création du projet SpringBoot
Quelle est la différence entre j2ee et springboot
Qu'est-ce que le métaverse
marque PC
Comment sauvegarder des programmes écrits en pycharm
Que faire si le code d'écran bleu 0x0000007e apparaît
Quels sont les types de trafic ?
Comment définir les numéros de page dans Word
![[Web front-end] Démarrage rapide de Node.js](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
