业务背景：
项目中涉及的消息队列既有RocketMQ，又有MQTT，均为阿里云提供（阿里云有专门的“微消息队列 MQTT 版”模块，但博主公司消息队列的实例都在“消息队列 RocketMQ 版”模块下，只是实例不同，猜测是做了适配，有清楚的大佬欢迎指点）。其中MQTT的消息是由硬件设备上报而来，由java服务进行消费，使用一套内部框架连接。现因框架存在适配问题，经讨论决定放弃使用，需改造原消费代码。
技术方案：
查询资料与询问同事得知两种消息队列在底层逻辑上高度相似，可以用RocketMQ方式连接MQTT的topic并消费。在实际改造过程中发现二者均可通过com.aliyun.openservices.ons.api.ONSFactory#createConsumer这一阿里云官方接口进行连接。而项目中已有一套配置用于连接RocketMQ，两类消息队列的地址、密钥等实例信息并不相同。因此改造的重点其实就是实现RocketMQ多实例配置。
代码实现：
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;/*** 自定义注解*/
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
public @interface OnsMqttMessageListener {String topic();String consumerGroup();}
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.stereotype.Component;/*** 参数类*/
@Data
@Component
@EnableConfigurationProperties
@ConfigurationProperties(prefix = "aliyun.rocketmq")
public class OnsMqttMessageProperties {private String onsAddr;private String accessKey;private String secretKey;private String groupId;}
import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.geelytech.bms.annotation.OnsMqttMessageListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.stereotype.Component;import java.util.Map;
import java.util.Properties;/*** 配置类*/
@Slf4j
@Component
public class OnsMqttMessageConsumerConfig implements ApplicationListener<ApplicationReadyEvent> {@Autowiredprivate OnsMqttMessageProperties onsMqttMessageProperties;/*** 注入所有MessageListener实例*/@Autowiredprivate Map<String, MessageListener> messageListeners;@Overridepublic void onApplicationEvent(ApplicationReadyEvent event) {String onsAddr = onsMqttMessageProperties.getOnsAddr();String accessKey = onsMqttMessageProperties.getAccessKey();String secretKey = onsMqttMessageProperties.getSecretKey();log.info("ConsumerSubscriptionConfig onApplicationEvent messageListeners={}", messageListeners);// 订阅每个MessageListenerfor (Map.Entry<String, MessageListener> entry : messageListeners.entrySet()) {MessageListener listener = entry.getValue();OnsMqttMessageListener annotation = AnnotationUtils.findAnnotation(listener.getClass(), OnsMqttMessageListener.class);if (annotation != null) {String topic = annotation.topic();String consumerGroup = annotation.consumerGroup();Properties properties = new Properties();properties.setProperty(PropertyKeyConst.GROUP_ID, consumerGroup);properties.setProperty(PropertyKeyConst.NAMESRV_ADDR, onsAddr);properties.setProperty(PropertyKeyConst.AccessKey, accessKey);properties.setProperty(PropertyKeyConst.SecretKey, secretKey);Consumer consumer = ONSFactory.createConsumer(properties);consumer.subscribe(topic, "*", listener);consumer.start();}}}}import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.geelytech.bms.annotation.OnsMqttMessageListener;
import com.geelytech.satellite.constant.TopicConstant;
import com.geelytech.satellite.dto.eventOriginalData.BizSwapDone;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;import static com.aliyun.openservices.ons.api.Action.CommitMessage;@Slf4j
@Component
@OnsMqttMessageListener(topic = "", consumerGroup = "")
public class BmsBizSwapDoneConsumer implements MessageListener {@Overridepublic Action consume(Message message, ConsumeContext consumeContext) {// 业务逻辑return CommitMessage;}}
