乐趣区

关于rocketmq:RocketMq商用RocketMq和开源RocketMq的兼容问题解决方案

作者：

jiezi

在

引言

在阿里云的官方网站提供了RocketMq的商用版本，然而集体在我的项目利用上发现和SpirngBoot以及Spring Cloud（Alibaba）等开源的RocketMQ依赖尽管能够失常兼容，然而仍然呈现了注解生效、启动报错，商用和开源版本的不兼容导致局部代码要反复编写的蛋疼问题。

这样的兼容问题不是简略加个SDK依赖，切换到商用配置就能够间接应用的（因为集体起初真就是这么想），为了防止前面再遇到这种奇葩的开发测试用开源RocketMq，生产环境须要应用商用集群的RocketMq的混合配置的业务场景，集体花了小半天工夫熟读阿里云的接入文档，加上各种尝试和测试，总结出一套能够疾速应用的兼容模板计划。

如果不理解阿里云商用RocketMq，能够看最初一个小节的【阿里云商用RocketMq介绍】介绍。集体的兼容计划灵感来自于官网提供的这个DEMO我的项目：springboot/java-springboot-demo。

留神本计划是基于SpringBoot2.X和 Spring cloud Alibaba 的两个我的项目环境构建我的项目根底，在SpringBoot上只做了生产者的配置，而在Spring Cloud Alibaba的Nacos上进行了生产者和消费者的残缺兼容计划。

最初留神兼容集成的版本为商用RocketMq应用4.x版本，最近新出的5.X 的版本并未进行测试，不保障失常应用。

兼容关键点

  1. 在沿用SpringBoot的YML根底配置根底上实现商用和开源模式的兼容。
  2. 商用RocketMq须要应用官网提供的依赖包，依赖包能够失常兼容SpringBoot等依赖。
  3. 集成之后便于开发和扩大，并且易于其余开发人员了解。
  4. 两种模式之间相互不会产生烦扰。

SpringBoot2.X 兼容

上面先介绍SpringBoot我的项目的兼容。

SpringBoot我的项目兼容

开源版本

首先咱们察看YAML文件，对于开源版本的RocketMq设置，单机版本能够间接配置一个ip和端口即可，如果是集群则用分号隔开多个NameServer的连贯IP地址（NameServ独立部署，外部进行主动同步 ）。

#音讯队列
rocketmq:
  # 自定义属性，作用下文将会进行解释
  use-aliyun-rocketSever: false
  name-server: 192.168.58.128:9876 # 192.168.244.128:9876;192.168.244.129:9876;
  producer:
    group: testGroup
    # 本地开发不应用商业版，能够不配置
    secret-key: NONE
    # 本地开发不应用商业版，能够不配置
    access-key: NONE
    # 商用版本申请超时工夫，开源版本不应用此参数
    timeoutMillis: NONE

SpringBoot中集成开源的RocketMq非常简单，只须要一个依赖就能够主动实现相干筹备：

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot</artifactId>
    <version>2.2.2</version>
    </dependency>

具体的应用通常为封装或者间接应用RocketMqTemplate：


@Autowired
private RocketMQTemplate mqTemplate;

public void sendRocketMqUniqueTextMessage(String topic, String tag, String queueMsg) {
        if (StringUtils.isNotEmpty(topic) && StringUtils.isNotEmpty(tag) && StringUtils.isNotEmpty(queueMsg)) {
            String queueName = topic + ":" + tag;
            //封装音讯，调配惟一id
            MessageData messageData = new MessageData();
            messageData.setMsgId(IdUtil.randomUUID());
            messageData.setMsgContent(queueMsg);
            queueMsg = JSON.toJSONString(messageData);
            log.info("线程：{}，向队列：{}，发送音讯：{}", Thread.currentThread()
                    .getName(), queueName, queueMsg);
            try {
                mqTemplate.syncSend(queueName, queueMsg);
            } catch (Exception e) {
                log.info("向队列：{}，发送音讯出现异常：{}", queueName, queueMsg);
                //出现异常，保留异样信息到数据库
                SaMqMessageFail saMqMessageFail = new SaMqMessageFail();
                // 封装失败音讯，调用生效解决Service将失败发送申请入库，或者通过其余办法重试
                saMqMessageFailService.insert(saMqMessageFail);
            }
        }
    }

开源版本的SpringBoot集成RocketMq就是如此简略。

商用版本

商用版本RocketMq咱们应用商业版TCP协定SDK（举荐），留神这里用的是4.X版本。商用版本的YAML配置和开源版本显式配置是一样的，然而须要留神参数use-aliyun-rocketSever=true，并且secretKey和accessKey以及name-server都须要配置为阿里云提供的配置，最初设置音讯发送超时工夫timeoutMillis设置正当工夫（单位为毫秒），便于排查问题和避免线程长期占用：

#音讯队列
rocketmq:
  use-aliyun-rocketSever: true
  # 应用阿里云提供的endpoint
  name-server: http://xxxxx.aliyuncs.com:8080
  producer:
    group: testGroup
    secret-key: xxx
    access-key: xxxx
    # 商用版本申请超时工夫
    timeoutMillis: 15000
  # 如果须要设置消费者，能够依照同样的形式集成
  consumer:
      group: testGroup
    secret-key: xxx
    access-key: xxxx
    # 商用版本申请超时工夫 15秒
    timeoutMillis: 15000

留神这里仅仅配置了生产者，读者能够按需设置为消费者，设置形式和生产者同理。

设置YAML之后，咱们须要在Maven中引入上面的配置：

<dependency>
    <groupId>com.aliyun.openservices</groupId>
    <artifactId>ons-client</artifactId>
    <!--以下版本号请替换为Java SDK的最新版本号-->
    <version>1.8.8.1.Final</version>
</dependency>

最初应该如何应用呢？这里就是商用RocketMq比拟蛋疼的点了，应用RocketMQTemplate是这种状况下是无奈应用商用RocketMq的，咱们须要 手动注入商用的SDK依赖ProducerBean，具体的操作如下：

  1. 构建配置类，这里仿照了官网提供的demo增减配置：
/**
阿里云服务配置封装，留神和本地部署的rocketmq配置辨别
*/
@Data
@Configuration
@ConfigurationProperties(prefix = "rocketmq")
public class AliyunRocketMqConfig {

    /**
     *鉴权须要的AccessKey ID
     */
    @Value("${rocketmq.use-aliyun-rocketSever:null}")
    private String useAliyunRocketMqServerEnable;
    /**
     *鉴权须要的AccessKey ID
     */
    @Value("${rocketmq.producer.access-key:null}")
    private String accessKey;

    /**
     *鉴权须要的AccessKey Secret
     */
    @Value("${rocketmq.producer.secret-key:null}")
    private String secretKey;

    /**
     * 实例TCP 协定公网接入地址（理论我的项目，填写本人阿里云MQ的公网地址）
     */
    @Value("${rocketmq.name-server:null}")
    private String nameSrvAddr;

    /**
     * 延时队列group
     */
    @Value("${rocketmq.producer.group:null}")
    private String groupId;

    /**
     * 音讯发送超时工夫，如果服务端在配置的对应工夫内未ACK，则发送客户端认为该音讯发送失败。
     */
    @Value("${rocketmq.producer.timeoutMillis:null}")
    private String timeoutMillis;


    //获取Properties
    public Properties getRocketMqProperty() {
        Properties properties = new Properties();
        properties.setProperty(PropertyKeyConst.GROUP_ID,this.getGroupId());
        properties.setProperty(PropertyKeyConst.AccessKey, this.accessKey);
        properties.setProperty(PropertyKeyConst.SecretKey, this.secretKey);
        properties.setProperty(PropertyKeyConst.NAMESRV_ADDR, this.nameSrvAddr);
        properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, this.timeoutMillis);
        return properties;
    }

}
  1. 构建商用RocketMq初始化类。这里会遇到比拟蛋疼的事件，因为咱们的依赖是商用RocketMq与开源的SpringBoot依赖共存的，尽管咱们能够商用的RocketMq，然而启动的时候会执行到此类进行初始化，返回NULL会导致SpringBoot我的项目无奈失常启动，这里无奈只能应用一个warn日志进行提醒开源版本有可能呈现Bean被笼罩问题，实际上应用下来没有特地大的影响。

写的比拟俊俏，读者有更优雅的解决形式欢送领导。笔者目前只想到了应用这种“不论”的形式保障我的项目不改任何代码的状况失常运行。

/**
 * 阿里云rocketMq初始化
 **/
@Configuration
@Slf4j
public class AliyunProducerInit {

    @Autowired
    private AliyunRocketMqConfig aliyunRocketMqConfig;

    @Bean(initMethod = "start", destroyMethod = "shutdown")
    public ProducerBean buildProducer() {
        if(!Boolean.valueOf(aliyunRocketMqConfig.getUseAliyunRocketMqServerEnable())){
            log.warn("非商用版本为了兼容仍然须要注入此Bean，然而只读取无关nameServ和group信息");
        }
        ProducerBean producer = new ProducerBean();
        // ProducerBean中的properties只有被笼罩的配置会应用自定义配置，其余配置会应用SDK的默认配置。
        producer.setProperties(aliyunRocketMqConfig.getRocketMqProperty());
        return producer;
    }
}

此外这里必须要吐槽一下商用RocketMq的正文竟然是全中文的！比方com.aliyun.openservices.ons.api.bean.ProducerBean的正文：

这样是坏事还是好事，大家自行领会。。。。。集体第一次看到的时候着实被震惊了。

  1. 做完下面两步之后，咱们就能够实现RocketMqTemplate调用申请，至此实现兼容。

SpringBoot我的项目兼容小结

这里简略小结一下SpringBoot的兼容过程，能够看到整个步骤仅仅是 在商用RocketMq多做了一步bean注入的操作而已，整体应用上非常简略。然而这里只介绍了生产者的集成，那么消费者如何兼容？稍安勿躁，咱们接着看Spring Cloud版本的集成案例。

Spring Cloud Alibaba我的项目兼容

目前国内应用的比拟多的是Spring Cloud Alibaba，留神这些配置都写入到Nacos当中。

开源版本

开源版本的接入形式和SpringBoot是一样的，这里简略回顾：

  1. 开源版本须要设置参数，这里设置了生产者和消费者：
#音讯队列 - 开源版本
rocketmq:
  use-aliyun-rocketSever: false
  name-server: 192.168.0.92:9876
  producer:
    group: testGroup
    secret-key: NONE
    access-key: NONE
    timeoutMillis: 15000
    consumeThreadNums: 20
  consumer:
    group: testGroup
    secret-key: NONE
    access-key: NONE
    timeoutMillis: 15000
    consumeThreadNums: 20
  1. 增加依赖：
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot</artifactId>
    <version>2.2.2</version>
</dependency>
  1. 应用RocketMqTemplate能够进行音讯发送，而消费者则须要应用监听器+注解的形式，疾速注入一个消费者。大体模板如下：
@Slf4j
@Component
@RocketMQMessageListener(topic = "test_topic", consumerGroup = "testGroup", selectorExpression = "test_tag")
public class QueueRemoteRecoListener implements RocketMQListener<MessageExt> {

    @Override
    public void onMessage(MessageExt message) {
        // dosomething
    }

}

商用版本

这节是本文略微简单一点的局部，咱们依照步骤介绍接入过程：

  1. 在Nacos的配置中退出RocketMq商用所需的配置内容，和开源版本的设置相似：
#音讯队列 - 商用版本
rocketmq:
 # 开源RocketMq和商业版RocketMq切换开关
use-aliyun-rocketSever: true
name-server: http://xxxx.mq.aliyuncs.com:80
producer:
# 目前uat借助开发测试应用
  group: testGroup
  secret-key: xxx
  access-key: xxx
  timeoutMillis: 15000
  consumeThreadNums: 20
consumer:
  group: testGroup
  secret-key: xxx
  access-key: xxx
  timeoutMillis: 15000
  consumeThreadNums: 20
  1. 增加商用RocketMq的TCP接入形式须要的依赖包。
<dependency>
    <groupId>com.aliyun.openservices</groupId>
    <artifactId>ons-client</artifactId>
    <!--以下版本号请替换为Java SDK的最新版本号-->
    <version>1.8.8.1.Final</version>
</dependency>
  1. 新增配置类，和SpringBoot的商用版本形式也是相似的：
/**
 * 
 * rocketmq 阿里云服务配置封装，留神和本地部署的rocketmq配置辨别
 **/
@Data
@Configuration
@ConfigurationProperties(prefix = "rocketmq")
public class AliyunCommercialRocketMqConfig {

    /**
     *鉴权须要的AccessKey ID
     */
    @Value("${rocketmq.use-aliyun-rocketSever:null}")
    private String useAliyunRocketMqServerEnable;
    /**
     *鉴权须要的AccessKey ID
     */
    @Value("${rocketmq.consumer.access-key:null}")
    private String accessKey;

    /**
     *
     */
    @Value("${rocketmq.use-aliyun-rocketsever:null}")
    private String useAliyunRocketServer;

    /**
     *鉴权须要的AccessKey Secret
     */
    @Value("${rocketmq.consumer.secret-key:null}")
    private String secretKey;

    /**
     * 实例TCP 协定公网接入地址（理论我的项目，填写本人阿里云MQ的公网地址）
     */
    @Value("${rocketmq.name-server:null}")
    private String nameSrvAddr;

    /**
     * 延时队列group
     */
    @Value("${rocketmq.consumer.group:null}")
    private String groupId;

    /**
     * 音讯发送超时工夫，如果服务端在配置的对应工夫内未ACK，则发送客户端认为该音讯发送失败。
     */
    @Value("${rocketmq.consumer.timeoutMillis:null}")
    private String timeoutMillis;

    /**
     * 将消费者线程数固定为20个 20为默认值
     */
    @Value("${rocketmq.consumer.consumeThreadNums:null}")
    private String consumeThreadNums;



    public Properties getRocketMqProperty() {
        Properties properties = new Properties();
        properties.setProperty(PropertyKeyConst.GROUP_ID,this.getGroupId());
        properties.setProperty(PropertyKeyConst.AccessKey, this.accessKey);
        properties.setProperty(PropertyKeyConst.SecretKey, this.secretKey);
        properties.setProperty(PropertyKeyConst.NAMESRV_ADDR, this.nameSrvAddr);
        properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, this.timeoutMillis);
        return properties;
    }

}
  1. 下一步是扩大官网音讯订阅类，为啥要这样做？次要是接入试验中发现官网的音讯订阅类没有 全量参数的结构器，难以对于音讯订阅类动态参数化常量，结构一个订阅音讯应用默认形式还须要static代码块设置。所以这里对于官网的音讯订阅类进行扩大，子类代码是对于父类拷贝了一遍，没有做其余改变：
/**
对于指定类进行扩大，更容易的初始化
@see com.aliyun.openservices.ons.api.bean.Subscription 被扩大类
 **/
@Data
@ToString
@AllArgsConstructor
@NoArgsConstructor
public class AliyunCommercialRocketMqSubscriptionExt extends Subscription {
    /**
     * 主题
     */
    private String topic;
    /**
     * 条件表达式，具体参考rocketmq
     */
    private String expression;

    /**
     * TAG or SQL92
     * <br>if null, equals to TAG
     *
     * @see com.aliyun.openservices.ons.api.ExpressionType#TAG
     * @see com.aliyun.openservices.ons.api.ExpressionType#SQL92
     */
    private String type;


    @Override
    public int hashCode() {
        final int prime = 31;
        int result = 1;
        result = prime * result + ((topic == null) ? 0 : topic.hashCode());
        return result;
    }

    @Override
    public boolean equals(Object obj) {
        if (this == obj) {
            return true;
        }
        if (obj == null) {
            return false;
        }
        if (getClass() != obj.getClass()) {
            return false;
        }
        AliyunCommercialRocketMqSubscriptionExt other = (AliyunCommercialRocketMqSubscriptionExt) obj;
        if (topic == null) {
            if (other.topic != null) {
                return false;
            }
        } else if (!topic.equals(other.topic)) {
            return false;
        }
        return true;
    }

}
  1. 接着咱们构建动态化的音讯订阅类，这个常量类会封装零碎须要应用到的音讯订阅对象，在后续注册监听器须要应用，这里先提前定义。
/**
 * 队列惯例配置，用于启动时候初始化配置，
 * 所有配置均须要搁置到此类中对外扩大应用
 **/
public final class AliyunCommercialRocketMqConstants {
    
    public static final String TEST_TOPIC = "test_topic";
    public static final String TEST_TAG = "test_tag";
    
    /**
     * 参考案例，仅仅作为开发验证应用
     */
    public static final AliyunCommercialRocketMqSubscriptionExt QUEUE_TEST = new AliyunCommercialRocketMqSubscriptionExt(RocketMqKey.TEST_TOPIC, RocketMqKey.TEST_TAG, null);
}

为了方便管理，咱们能够把这些要害Key在独自放到一个类：

public class RocketMqKey {
    public static final String TEST_TOPIC = "test_topic";
    public static final String TEST_TAG = "test_tag";   
}

/**
 * 队列惯例配置，用于启动时候初始化配置，
 * 所有配置均须要搁置到此类中对外扩大应用
 **/
public final class AliyunCommercialRocketMqConstants {
    /**
     * 参考案例，仅仅作为开发验证应用
     */
    public static final AliyunCommercialRocketMqSubscriptionExt QUEUE_TEST = new AliyunCommercialRocketMqSubscriptionExt(RocketMqKey.TEST_TOPIC, RocketMqKey.TEST_TAG, null);
}
  1. 做好一系列筹备之后，咱们开始商用版本生产者兼容，商用版本的发送者能够像是上面这样封装官网提供的demo，也能够应用注入ProducerBean的形式集成：
/**
 * 商用aliyun Rocketmq 工具类封装
 **/
@Component
@Slf4j
public class AliyunCommercialRocketMqSendUtils {

    private static final long KEEP_ALIVE_TIME = 60L;

    private final ThreadPoolExecutor THREAD_POOL_EXECUTOR = new ThreadPoolExecutor(0, Integer.MAX_VALUE,
            KEEP_ALIVE_TIME, TimeUnit.SECONDS,
            new SynchronousQueue<>());

    @Autowired
    private Producer producer;

    @Autowired
    private SaMqMessageFailService saMqMessageFailService;

    @Autowired
    private AliyunRocketMqConfig aliyunRocketMqConfig;

    /**
     * 异步推送，须要调用方手动指定callback
     
     */
    public void singleAsyncSend(String topic, String tag, String msgBody, SendCallback sendCallback) {
        singleAsyncSend(topic, tag, msgBody, null, sendCallback);
    }

    /**
     * 异步推送，须要调用方手动指定callback
     * 如果须要应用callback告诉，能够应用上面的代码进行解决
     <pre>
         producer.sendAsync(msg, new SendCallback() {
        @Override
        public void onSuccess(final SendResult sendResult) {
            assert sendResult != null;
            System.out.println(sendResult);
        }

        @Override
        public void onException(final OnExceptionContext context) {
            ONSClientException exception = context.getException();
            //                    //出现异常意味着发送失败，为了防止音讯失落，倡议缓存该音讯而后进行重试。
            log.error("【RocketMq-Commercial】发送失败，音讯存储到失败记录表，发送 topic => {}, 发送 tag => {}， msgBody => {}， key => {}（如果设置key，能够应用key查找音讯）,商用版须要在RocketMq云服务重试，失败起因为: {}"
            , topic, tag, msgBody, key, exception.getMessage());
            buildMqErrorInfoAndInsertDb(msgBody, exception);
        }
    });
     </pre>
     * @param topic 主题
     * @return void
     * @description
     * @param: tag  标签
     * @param: msgBody 推送音讯
     * @param: key 不便音讯查找的key
     * @param: sendCallback 手动回调
     */
    public void singleAsyncSend(String topic, String tag, String msgBody, String key, SendCallback sendCallback) {
        if(Boolean.FALSE.equals(Boolean.valueOf(aliyunRocketMqConfig.getUseAliyunRocketMqServerEnable()))){
            throw new UnsupportedOperationException("请设置 UseAliyunRocketServer=true 切换到商用rocketMq模式");
        }
        if (StringUtils.isAnyBlank(topic, tag, msgBody)) {
            throw new IllegalArgumentException("topic, tag and msgBody must be not blank");
        }
        //对于应用异步接口，倡议设置独自的回调解决线程池，领有更灵便的配置和监控能力。
        //如下结构线程的形式申请队列为无界仅用作示例，有OOM的危险。
        //更正当的结构形式请参考阿里巴巴Java开发手册： https://github.com/alibaba/p3c
        producer.setCallbackExecutor(THREAD_POOL_EXECUTOR);

        //循环发送音讯
        Message msg = new Message( //
                // Message所属的Topic
                topic,
                // Message Tag 可了解为Gmail中的标签，对音讯进行再归类，不便Consumer指定过滤条件在MQ服务器过滤
                tag,
                // Message Body 能够是任何二进制模式的数据， MQ不做任何干涉
                // 须要Producer与Consumer协商好统一的序列化和反序列化形式
                msgBody.getBytes());
        // 设置代表音讯的业务要害属性，请尽可能全局惟一
        // 以不便您在无奈失常收到音讯状况下，可通过MQ 控制台查问音讯并补发
        // 留神：不设置也不会影响音讯失常收发
        if (CharSequenceUtil.isNotBlank(key)) {
            msg.setKey(key);
        }
        // 发送音讯，只有不抛异样就是胜利
        try {
            producer.sendAsync(msg, sendCallback);
        } catch (ONSClientException exception) {
            //出现异常意味着发送失败，为了防止音讯失落，倡议缓存该音讯而后进行重试。
            log.error("【RocketMq-Commercial】发送失败，音讯存储到失败记录表，发送 topic => {}, 发送 tag => {}， msgBody => {}， key => {}（如果设置key，能够应用key查找音讯）,商用版须要在RocketMq云服务重试，失败起因为: {}"
                    , topic, tag, msgBody, key, exception.getMessage());
            buildMqErrorInfoAndInsertDb(msgBody, exception);
        }


    }

    /**
     * 阻塞独自推送队列，应用零碎默认的Key生成规定
     *
     * @param topic 主题
     * @return void
     * @description 阻塞独自推送队列
     * @param: tag 标签
     * @param: msgBody msgBody内容
     */
    public void singleSyncSend(String topic, String tag, String msgBody) {
        singleSyncSend(topic, tag, msgBody, null);
    }

    /**
     * 阻塞独自推送队列，应用零碎默认的Key生成规定
     *
     * @param topic 主题
     * @return void
     * @description 阻塞独自推送队列
     * @param: tag 标签
     * @param: msgBody msgBody内容
     */
    public void singleSyncSend(String topic, String tag, String msgBody, String key) {
        if(!Boolean.valueOf(aliyunRocketMqConfig.getUseAliyunRocketMqServerEnable())){
            throw new UnsupportedOperationException("请设置 UseAliyunRocketServer=true 切换到商用rocketMq模式");
        }
        if (StringUtils.isAnyBlank(topic, tag, msgBody)) {
            throw new IllegalArgumentException("topic, tag and msgBody must be not blank");
        }
        Message msg = new Message( //
                // Message所属的Topic
                topic,
                // Message Tag 可了解为Gmail中的标签，对音讯进行再归类，不便Consumer指定过滤条件在MQ服务器过滤
                tag,
                // Message Body 能够是任何二进制模式的数据， MQ不做任何干涉
                // 须要Producer与Consumer协商好统一的序列化和反序列化形式
                msgBody.getBytes());
        // 设置代表音讯的业务要害属性，请尽可能全局惟一
        // 以不便您在无奈失常收到音讯状况下，可通过MQ 控制台查问音讯并补发
        // 留神：不设置也不会影响音讯失常收发
        if (CharSequenceUtil.isNotBlank(key)) {
            msg.setKey(key);
        }
        // 发送音讯，只有不抛异样就是胜利
        try {
            SendResult sendResult = producer.send(msg);
            log.info("【RocketMq-Commercial】推送胜利，singleSyncSend => {}", JSON.toJSONString(sendResult));
        } catch (ONSClientException e) {
            log.error("【RocketMq-Commercial】发送失败，音讯存储到失败记录表，发送 topic => {}, 发送 tag => {}， msgBody => {}， key => {}（如果设置key，能够应用key查找音讯）,商用版须要在RocketMq云服务重试，失败起因为: {}"
                    , topic, tag, msgBody, key, e.getMessage());
            buildMqErrorInfoAndInsertDb(msgBody, e);
        }

    }

    /**
     * 构建mq错误信息，并且插入到异样推送表
     *
     * @param msgBody
     * @return void
     * @description 构建mq错误信息，并且插入到异样推送表
     * @param: e
     */
    private void buildMqErrorInfoAndInsertDb(String msgBody, ONSClientException e) {
        //出现异常意味着发送失败，为了防止音讯失落，倡议缓存该音讯而后进行重试。
        //出现异常，保留异样信息到数据库
        SaMqMessageFail saMqMessageFail = new SaMqMessageFail();
        saMqMessageFail.setMqId(IdUtil.randomUUID());
        saMqMessageFail.setQueueName("RocketMq-Commercial");
        saMqMessageFail.setQueueMessage(msgBody);
        // 2 代表失败
        saMqMessageFail.setStatus(2);
        // 1 代表重试次数
        saMqMessageFail.setProcCount(1);
        saMqMessageFail.setFailReason(e.toString());
        saMqMessageFail.setCreateDate(new Date());
        saMqMessageFailService.insert(saMqMessageFail);
    }

}

注入ProductBean的形式能够从下面提到的SpringBoot集成形式解决。

  1. 当初咱们解决之前遗留的问题，如果是消费者，应该如何更优雅的接管音讯？首先咱们须要明确，商用RocketMq注入是须要依附手动构建监听器，然而咱们下面提到SpringBoot提供了注解+ @Component的形式实现队列监听的消费者。此外为了防止和开源版本抵触，咱们应用之前参数配置自定义的开关，在遇到开源版本的时候咱们返回null（这里返回null Spring会扫描获取SpringBoot的RocketMq依赖，不会呈现报错和无奈启动的问题，和后面的状况略有不同）避免主动注入商用RocketMq的监听器。

上面是商用版本注入ConsumerBean的代码，订阅商用RocketMq的生产监听器：

/**
 *  阿里云商用Rocketmq 队列接管
 *  商用rocketmqConsumer的所有申请会进入一个入口，须要散发到不同的具体业务解决。
 **/
@Component
@Slf4j
public class AliyunCommercialRocketMqQueueConsumer {

    @Autowired
    private AliyunCommercialRocketMqConfig aliyunCommercialRocketMqConfig;

    // 自定义的监听器
    @Autowired
    private AliyunCommerciaRocketMqTestListener aliyunCommerciaRocketMqTestListener;

    @Bean(initMethod = "start", destroyMethod = "shutdown")
    public ConsumerBean buildConsumer() {
        // 如果开关为false, 则不能注入此对象，否则商用的API会顶替掉 框架诸如的Bean出现异常
        if(!Boolean.valueOf(aliyunCommercialRocketMqConfig.getUseAliyunRocketServer())){
            log.warn("非商用版本不注入商用版本监听器");
            return null;
        }
        ConsumerBean consumerBean = new ConsumerBean();
        //配置文件
        Properties properties = aliyunCommercialRocketMqConfig.getRocketMqProperty();
        properties.setProperty(PropertyKeyConst.GROUP_ID, aliyunCommercialRocketMqConfig.getGroupId());
        //将消费者线程数固定为20个 20为默认值
        properties.setProperty(PropertyKeyConst.ConsumeThreadNums, aliyunCommercialRocketMqConfig.getConsumeThreadNums());
        consumerBean.setProperties(properties);
        //订阅关系
        Map<Subscription, MessageListener> subscriptionTable = new HashMap<>(2);
        // 应用了之前定义的测试用的监听器
        subscriptionTable.put(AliyunCommercialRocketMqConstants.QUEUE_TEST, aliyunCommerciaRocketMqTestListener);
        //订阅多个topic如下面设置

        consumerBean.setSubscriptionTable(subscriptionTable);
        return consumerBean;
    }


}

AliyunCommerciaRocketMqTestListener 自定义监听器为了兼容商用和开源版本做了上面的扭转，为了不便了解这里退出了相干注解：

/**
 * 监听器开发模板
 **/
@Slf4j
@Component
// 开源版本能够兼容此注解
@RocketMQMessageListener(topic = RocketMqKey.TEST_TOPIC, consumerGroup = RocketMqKey.TEST_GROUP, selectorExpression = RocketMqKey.TEST_TAG)
// RocketMQListener<MessageExt> 属于SpringBoot RocketMq的依赖
// MessageListener 属于商用RocketMq的SDK
public class QueueRemoteReconListener implements RocketMQListener<MessageExt>, MessageListener {


    // 开源版本的应用
    @Override
    public void onMessage(MessageExt message) {
        // 开源版本的SpringBoot形式，和上文介绍雷同
    }

    /** 
     * @description  
     * @param message 音讯类. 一条音讯由主题, 音讯体以及可选的音讯标签, 自定义从属键值对形成.
     * 留神: 咱们对每条音讯的自定义键值对的长度没有限度, 但所有的自定义键值对, 零碎键值对序列化后, 所占空间不能超过32767字节.
     * @param: context 每次生产音讯的上下文，供未来扩大应用
     * @return com.aliyun.openservices.ons.api.Action 
     */ 
    @Override
    public Action consume(Message message, ConsumeContext context) {
        // 商用版本要求实现这个办法,Action如下，默认失败重试16次，返回ReconsumeLater会触发重试机制，所以会存在反复生产的问题
        //public enum Action {
            /**
             * 生产胜利，持续生产下一条音讯
             */
            //CommitMessage,
            /**
             * 生产失败，告知服务器稍后再投递这条音讯，持续生产其余音讯
             */
            //ReconsumeLater,
        //}

    }
}

以上就是消费者的兼容解决，既能够满足开源版本的注解开发要求，也能够不收缩类的状况下沿用扩大。

Spring Cloud Alibaba我的项目兼容小结

从集体的角度来看，在生产者的兼容上比拟容易实现，依照官网的demo构建带商用RocketMq的ProducerBean即可，而消费者的集成则要简单一些，这里为了思考不反复的设置或者写反复代码，把要害的配置动态化，同时为了官网写的毛糙的音讯订阅类做了继承笼罩的操作兼容。此外为了避免开源版本的消费者Bean被商用的监听器笼罩导致生效，应用了简略的开关来进行Bean的注入管制，写的比拟毛糙，读者有更好的写法欢送探讨。

总体来看来说商用版本的RocketMq集成起来稍微麻烦，然而还是能够承受。此外也能够看到SDK的兼容性实际上是比拟简陋的，很多中央的很像然而又齐全是本人从新设计的，感觉就是一个新的团队给老团队做进去的货色做兼容的感觉，不过不就是桥接嘛，我也会，所以最初集体凑合消费者兼容就用来多接口实现的兼容写法了。

最终的集成成果是开源版本敞开注入bean的开关，配置为了关照Properties对于不须要的配置进行占位解决，而商用版本则关上开关，通过自定义注入监听器的形式顶替掉SpringBoot的依赖。

最初的成果是只须要批改RocketMq的配置，启动之后主动连贯到相干的RocketMq。

开源版本：

#音讯队列 - 开源版本
rocketmq:
  use-aliyun-rocketSever: false
  name-server: 192.168.0.92:9876
  producer:
    group: testGroup
    secret-key: NONE
    access-key: NONE
    timeoutMillis: 15000
    consumeThreadNums: 20
  consumer:
    group: testGroup
    secret-key: NONE
    access-key: NONE
    timeoutMillis: 15000
    consumeThreadNums: 20

商用版本

#音讯队列 - 商用版本
rocketmq:
 # 开源RocketMq和商业版RocketMq切换开关
  use-aliyun-rocketSever: true
  name-server: http://xxxx.mq.aliyuncs.com:80
  producer:
# 目前uat借助开发测试应用
    group: testGroup
      secret-key: xxx
      access-key: xxx
      timeoutMillis: 15000
      consumeThreadNums: 20
    consumer:
      group: testGroup
      secret-key: xxx
      access-key: xxx
      timeoutMillis: 15000
      consumeThreadNums: 20

阿里云商用RocketMq简略介绍（4.X版本）

官网介绍：https://help.aliyun.com/product/29530.html

无关RocketMq自身的介绍局部能够浏览：[[【RocketMq】RocketMq 扫盲]]，这里挑了商用版本集体认为须要关注的几个点进行介绍。留神这里应用的商用RocketMq版本为4.X的版本。

计费模式

商用RocketMq次要分为上面几个局部：

  • 主系列：标准版、专业版、铂金版
  • 子系列：单节点版、集群版

集体最初是白嫖了公司的商用集群版RocketMq进行试验，计费的形式分为包年包月和按量付费，前者实用于流量比拟大并且固定的状况，应用套餐比拟划算，而后者按需付费则实用于RocketMq应用较少（或者不稳固）或者调用量较少的状况，集体学习也比拟举荐按量付费模式，比包年包月划算很多。

如果呈现欠费，在实例停服7天后两种付费模式均会革除所有的RocketMq数据，而日常欠费则会被主动停用，嘛，和电话卡欠费差不多的情理，这里就不过多拓展了。如果要退订商用RocketMq的服务。按量付费能够随时退订，包年包月也会依据天数进行退订。

PS：看完这一整套计费模式下来，发现还是挺良心的。让我没想到的是包年包月竟然能够计算未应用的天数返还，这一点挺不错的，用起来不必放心被套进去。

接入形式

商用RocketMq的接入步骤如下：

官网在疾速入门中提供了创立资源的解释视频：https://help.aliyun.com/document_detail/441914.html。有两个点须要留神，第一个点是RAM用户必须要进行账户受权能力失常应用，第二个点是对外拜访形式设置，音讯队列RocketMQ版反对VPC拜访和公网拜访。资源创立实现之后就能够应用官网提供的SDK收发音讯，在应用之前须要留神上面这些前提：

  • 步骤二：创立资源

  • 装置IDEA

    您能够应用IntelliJ IDEA或者Eclipse，本文以IntelliJ IDEA Ultimate为例。

  • 装置1.8或以上版本JDK
  • 装置2.5或以上版本Maven

之后便是在我的项目中引入JAVA依赖和复制粘贴模板代码验证。

音讯队列模型

商用RocketMq的根本应用逻辑如下：

依照音讯类型，商用RocketMq提供了上面的音讯类型：

  • 一般音讯
  • 程序音讯
  • 定时/延时音讯
  • 事务音讯

SDK接入参考

SDK接入的主页链接如下：

https://help.aliyun.com/document_detail/69111.html

次要介绍了上面几个点，理论参考第二个举荐的接入形式即可。

  • SDK参考概述
  • 商业版TCP协定SDK（举荐）
  • 商业版HTTP协定SDK（多语言举荐）
  • 社区版TCP协定SDK（仅供开源用户上云应用）
  • 订阅音讯常见问题

对于JAVA应用程序倡议应用TCP协定SDK，集成的形式也比较简单，为了不便了解，也能够通过以下链接获取相干Demo，外面都有具体的正文解释：

  • spring/java-spring-demo
  • springboot/java-springboot-demo

如果是HTTP协定的SDK，则应用上面的依赖包：

<dependency>
    <groupId>com.aliyun.mq</groupId>
    <artifactId>mq-http-sdk</artifactId>
    <!--以下版本号请替换为Java SDK的最新版本号-->
    <version>1.0.3.2</version> 
    <classifier>jar-with-dependencies</classifier>
</dependency>

如果是TCP协定的接入形式，则应用上面的依赖包：

<dependency>
    <groupId>com.aliyun.openservices</groupId>
    <artifactId>ons-client</artifactId>
    <!--以下版本号请替换为Java SDK的最新版本号-->
    <version>1.8.8.1.Final</version>
</dependency>

接入细节阐明

集体在试验的时候发现HTTP协定的版本集成SDK更像是给了一套API工具包，益处是能够不须要依赖Spring框架等独自应用，然而最大的毛病也是这里，很难用于框架当中，如果要实现主动接管音讯并且解决，须要依附手动实现定时工作拉取音讯实现主动生产，这一点也比拟蛋疼。

这里列举 HTTP的SDK的发送和承受代码，首先是生产者的模板代码，这些代码和开源的RocketMq应用相似：

// 省略大量代码。。。
// 循环发送4条音讯。
            for (int i = 0; i < 4; i++) {
                TopicMessage pubMsg;        // 一般音讯。
                pubMsg = new TopicMessage(
                // 音讯内容。
                "hello mq!".getBytes(),
                // 音讯标签。
                "A"
                );
                // 设置音讯的自定义属性。
                pubMsg.getProperties().put("a", String.valueOf(i));
                // 设置音讯的Key。
                pubMsg.setMessageKey("MessageKey");
               
            // 同步发送音讯，只有不抛异样就是胜利。
            TopicMessage pubResultMsg = producer.publishMessage(pubMsg);

            // 同步发送音讯，只有不抛异样就是胜利。
            System.out.println(new Date() + " Send mq message success. Topic is:" + topic + ", msgId is: " + pubResultMsg.getMessageId()
                        + ", bodyMD5 is: " + pubResultMsg.getMessageBodyMD5());
            }
        } catch (Throwable e) {
            // 音讯发送失败，须要进行重试解决，可从新发送这条音讯或长久化这条数据进行弥补解决。
            System.out.println(new Date() + " Send mq message failed. Topic is:" + topic);
            e.printStackTrace();
        }
// 省略大量代码。。。

而在接管方略微麻烦一些，官网的案例是应用 死循环一直查看Broker是否有积压音讯，如果有则通过被动拉取音讯的模式去拉取音讯。

实现多线程期待的成果能够写入到Thread继承类的Run办法中。

 // 在以后线程循环生产音讯，倡议多开个几个线程并发生产音讯。
        do {
            
            // 省略大量代码

// 解决业务逻辑。
            for (Message message : messages) {
                System.out.println("Receive message: " + message);
            }

            // 音讯重试工夫达到前若不确认音讯生产胜利，则音讯会被反复生产。
            // 音讯句柄有工夫戳，同一条音讯每次生产的工夫戳都不一样。
            {
                List<String> handles = new ArrayList<String>();
                for (Message message : messages) {
                    handles.add(message.getReceiptHandle());
                }

                try {
                    consumer.ackMessage(handles);
                } catch (Throwable e) {
                    // 某些音讯的句柄可能超时，会导致音讯生产状态确认不胜利。
                    if (e instanceof AckMessageException) {
                        AckMessageException errors = (AckMessageException) e;
                        System.out.println("Ack message fail, requestId is:" + errors.getRequestId() + ", fail handles:");
                        if (errors.getErrorMessages() != null) {
                            for (String errorHandle :errors.getErrorMessages().keySet()) {
                                System.out.println("Handle:" + errorHandle + ", ErrorCode:" + errors.getErrorMessages().get(errorHandle).getErrorCode()
                                        + ", ErrorMsg:" + errors.getErrorMessages().get(errorHandle).getErrorMessage());
                            }
                        }
                        continue;
                    }
                    e.printStackTrace();
                }
            }
       } while (true);
     // 省略大量代码

写在最初

算是一次集体兼容的笔记。还是要吐槽商用的RocketMq集成真的挺无语的，不过阿里的货色懂的都懂。

评论

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

这个站点使用 Akismet 来减少垃圾评论。了解你的评论数据如何被处理。

更多文章