基于AOP思想构造RocketMQ组件

原创张海威转转技术

1.背景

1.1 RocketMQ集群简介
1.2 业务开发对于Consumer、Producer客户端的一般用法
1.3 设计新组件的基本思路

2.HunterConsumer的工作原理

2.1 读取Spring容器中的全部配置
2.2 加载Spring容器中的全部HunterConsumer注解
2.3 启动前检查
2.4 根据配置创建并启动Consmer客户端
2.5 销毁

3.HunterProducer的工作原理
4.总结

1.背景

1.1 RocketMQ集群简介

如图所示，RocketMQ集群由4部分组成：Producer会根据业务需要发送消息；Broker负责接收、存储和分发消息；Consumer负责按需消费消息；Name Server负责通过长连接、Topic路由、心跳检测等手段保证集群的高可用。

其中的Broker和Name Server都是由运维和架构部同学负责管理，业务开发接触较少。业务开发同学接触比较多的就是Producer、Consumer部分，这两部分都有RocketMQ提供的Java客户端，但需要嵌入到具体的业务代码里，我们的组件就是针对Java客户端易用性的扩展。

1.2 业务开发对于Consumer、Producer客户端的一般用法

Consumer一般用法如下：

public class TestListener implements MessageListenerConcurrently {

    DefaultMQPushConsumer consumer = null;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        Iterator<MessageExt> it = msgs.iterator();
        while (it.hasNext()) {
            MessageExt msgExt = it.next();
            //todo 业务处理
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
    }

    public void startConsumer() throws Exception {
        //创建&初始化&启动
        //创建Listener
        TestListener testListener = new TestListener();

        //创建Consumer
        consumer = new DefaultMQPushConsumer();

        //初始化方式1：通过配置文件指定参数
        consumer.init("配置文件路径");

        //初始化方式2：手动控制参数
        consumer.setSubscribeTopic("testTopic");
        consumer.subscribe("testTopic", "testTags");
        consumer.setConsumerGroup("testConsumerGroup");
        consumer.setNamesrvAddr("name server address");
        //todo 设置其他的一些客户端参数

        //注册Listener
        consumer.registerMessageListener(testListener);
        
        //启动
        consumer.start();
    }

    public void shutdown() {
        //销毁
        if (consumer != null) {
            consumer.shutdown();
        }
    }
}

Producer的一般用法：

public class TestProducer {

    private static DefaultMQProducer producer = null;

    public static void main(String[] args) throws Exception {
        //启动
        startProducer();

        //todo 业务处理
        Message msg = new Message();
        msg.setTopic("testTopic");
        msg.setTags("testTag");
        msg.setBody("消息体".getBytes(StandardCharsets.UTF_8));
        producer.send(msg);
    }

    public static void startProducer() throws Exception {
        //创建&初始化&启动
        //创建producer
        producer = new DefaultMQProducer();
        //初始化方式1：通过配置文件指定参数
        producer.init("配置文件路径");
        //初始化方式2：手动控制参数
        producer.setNamesrvAddr("name server address");
        producer.setProducerGroup("testProducerGroup");
        //启动
        producer.start();
    }

    public void shutdown() {
        //销毁
        if (producer != null) {
            producer.shutdown();
        }
    }
}

从代码中我们看到两个组件的生命周期都是3个阶段：创建&初始化&启动、业务处理、销毁，且在这3个阶段里其实只有业务处理阶段是与具体的业务开发紧密相关的。当我们要定义多个Topic、Group、Tag去发送或者消费消息的时候，发现其他2个阶段都属于重复代码，且在初始化阶段中的参数有些许不同，但还是要在业务开发中进行重复开发或配置。

1.3 设计新组件的基本思路

由于现有RcoketMQ客户端存在上述问题，我希望有一种新组件，能够在业务开发中，将这两个客户端代码中重复出现的2个阶段抽离出来，并允许业务侧保留个性化定制的部分。

于是想到可以基于AOP的思想，使用注解来定义Consumer和Producer实例的创建和初始化参数，利用Spring容器加载和管理实例化后的bean对象的生命周期，来设计和使用该组件，这样即满足了业务定制化的诉求，又达到了减少重复开发的目的。

1，新的Consumer组件（HunterConsumer）的一般用法：

    @HunterConsumer(topic = "testTopic", tags = {"testTags"}, group = "testConsumerTopic")
    public boolean testConsumerXXX(String tag, String msgId, String body) {
        //todo 业务处理。返回true表示消费成功、false表示消费失败，消费异常表示消费失败
        return true;
    }

即，将@HunterConsumer用在被注解的方法上，将注解上的topic、tags、group等作为个性化的初始化参数，在消费消息时将消息内容通过参数名称和类型确定，作为参数传递给消费者客户端，再根据返回值true/false决定消费成功或者失败。

2，新的Producer组件（HunterProducer）的一般用法：

    @HunterProducer(topic = "testTopic", group = "testProducerGroup")
    private IHunterProducer testTopicProducer;
    
    //todo 业务处理
    testTopicProducer.send("testTag", "消息体字符串");

即，将@HunterProducer用在被注解的字段上，将注解上的topic、tags、group等作为个性化的初始化参数，创建IHunterProducer客户端实例，提供发送消息的接口。

如上述代码所示，对于Consumer、Producer客户端生命周期中的创建&初始化&启动、销毁2个阶段，由工程在首次引入组件时，一次性简单配置即可，整个业务开发过程中只需要使用@HunterConsumer和@HunterProducer两个注解就行了。

2.HunterConsumer的工作原理

HunterConsumer组件依托于Spring容器来管理bean的生命周期，所以在业务侧在引入该组件时要使用Spring的ApplicationContext来进行初始化，具体如下：

@Component
public class ProjectInit implements ApplicationListener<ContextRefreshedEvent> {

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        try {
            HunterConsumerStarter.startConsumers(event.getApplicationContext());
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(1);
        }
    }
}

HunterConsumerStarter.startConsumers内部在初始化时总体会进行如下4步处理：读取Spring容器中的全部配置、加载Spring容器中的全部HunterConsumer注解、启动前检查、根据配置创建并启动Consmer客户端

2.1 读取Spring容器中的全部配置

从Spring容器中加载组件所需要的基本配置信息，如：是否启动HunterConsumer组件、name server地址、默认是否打印详细日志、默认核心线程池大小核心代码如下：

//构造Spring配置读取工具
@Slf4j
@Component("com.zhuanzhuan.hunter.mq.consumer.util.PropertyTool")
public class PropertyTool implements EnvironmentAware, EmbeddedValueResolverAware {

    private static Environment environment;
    private static StringValueResolver stringValueResolver;

    /**
     * 动态获取配置文件中的值
     *
     * @param key 配置key
     * @return
     */
    public static String getFromResover(String key) {
        if (PropertyTool.stringValueResolver == null) {
            return null;
        }
        String name = "${" + key + "}";
        try {
            return PropertyTool.stringValueResolver.resolveStringValue(name);
        } catch (Exception e) {
            log.info("key={} name={} error={}", key, name, e.getMessage());
            return null;
        }
    }

    /**
     * 从环境变量中获取值
     *
     * @param key 配置key
     * @return
     */
    public static String getFromEnvironment(String key) {
        if (PropertyTool.environment != null) {
            return PropertyTool.environment.getProperty(key);
        }
        return null;
    }

    public static String getString(String key) {
        String value = getFromEnvironment(key);
        if (StringUtils.isNotBlank(value)) {
            return value;
        }
        return getFromResover(key);
    }

    public static String getString(String key, String defaultValue) {
        String value = getString(key);
        if (StringUtils.isNotBlank(value)) {
            return value;
        }
        return defaultValue;
    }

    public static Integer getInt(String key) {
        String value = getString(key);
        if (value == null) {
            return null;
        }
        return NumberUtils.toInt(value);
    }

    public static int getInt(String key, Integer defaultValue) {
        Integer value = getInt(key);
        if (value != null) {
            return value;
        }
        return defaultValue;
    }

    public static Long getLong(String key) {
        String value = getString(key);
        if (value == null) {
            return null;
        }
        return NumberUtils.toLong(value);
    }

    public static long getLong(String key, long defaultValue) {
        Long value = getLong(key);
        if (value != null) {
            return value;
        }
        return defaultValue;
    }

    public static Double getDouble(String key) {
        String value = getString(key);
        if (value == null) {
            return null;
        }
        return NumberUtils.toDouble(value);
    }

    public static double getDouble(String key, double defaultValue) {
        Double value = getDouble(key);
        if (value != null) {
            return value;
        }
        return defaultValue;
    }

    public static Float getFloat(String key) {
        String value = getString(key);
        if (value == null) {
            return null;
        }
        return NumberUtils.toFloat(value);
    }

    public static float getFloat(String key, float defaultValue) {
        Float value = getFloat(key);
        if (value != null) {
            return value;
        }
        return defaultValue;
    }

    public static Date getDate(String key) {
        return DateTool.str2Date(getString(key));
    }

    public static Date getDate(String key, Date defaultValue) {
        Date value = getDate(key);
        if (value != null) {
            return value;
        }
        return defaultValue;
    }

    public static Boolean getBoolean(String key) {
        String value = getString(key);
        if (value == null) {
            return null;
        }
        return "true".equalsIgnoreCase(value.trim());
    }

    public static boolean getBoolean(String key, boolean defaultValue) {
        Boolean value = getBoolean(key);
        if (value != null) {
            return value;
        }
        return defaultValue;
    }

    @Override
    public void setEnvironment(Environment environment) {
        PropertyTool.environment = environment;
    }

    @Override
    public void setEmbeddedValueResolver(StringValueResolver stringValueResolver) {
        PropertyTool.stringValueResolver = stringValueResolver;
    }

}

读取Spring中的配置信息

@Slf4j
@Configuration
public class HunterConsumerConfig {

    /**
     * 运行环境cup核数
     **/
    private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();

    @Bean(HunterConsumerConsts.HUNTER_CONSUMER_CONFIG_BEAN)
    public HunterMqEnvConfig getEnvConfig() {
        log.info("HunterConsumer initEnvConfig start ...");
        HunterMqEnvConfig config = new HunterMqEnvConfig();
        config.setShouldStart(PropertyTool.getBoolean(HunterConsumerConsts.ENV_SHOULD_START, false));
        config.setLogAllDetail(PropertyTool.getBoolean(HunterConsumerConsts.ENV_LOG_ALL_INFO, true));
        config.setBodyCharset(PropertyTool.getString(HunterConsumerConsts.ENV_BODY_CHARSET, HunterConsumerConsts.UTF8));
        config.setNamesrvAddr(PropertyTool.getString(HunterConsumerConsts.ENV_NAME_SRV_ADDR));
        config.setConsumeFromWhere(PropertyTool.getString(HunterConsumerConsts.ENV_CONSUME_FROM_WHERE));
        config.setConsumeThreadMin(PropertyTool.getInt(HunterConsumerConsts.ENV_CONSUME_THREAD_MIN, CPU_COUNT + 1));
        config.setConsumeThreadMax(PropertyTool.getInt(HunterConsumerConsts.ENV_CONSUME_THREAD_MAX, CPU_COUNT * 3));
        config.setPullBatchSize(PropertyTool.getInt(HunterConsumerConsts.ENV_PULL_BATCH_SIZE, 32));
        config.setTopicAutoCreate(PropertyTool.getBoolean(HunterConsumerConsts.ENV_TOPIC_AUTO_CREATE, true));
        log.info("HunterConsumer initEnvConfig end envConfig={}", GsonUtil.toJson(config));
        return config;
    }
}

2.2 加载Spring容器中的全部HunterConsumer注解

扫描Spring容器中的全部bean获取各个方法上的@HunterConsmuer注解，并检查注解使用方式是否正确

    private static void initExecUnits(ApplicationContext applicationContext) {
        String[] beanNames = applicationContext.getBeanDefinitionNames();
        if (beanNames == null || beanNames.length == 0) {
            return;
        }
        for (String beanName : beanNames) {
            Object bean = applicationContext.getBean(beanName);
            if (AopUtils.isAopProxy(bean)) {
                try {
                    initExecUnit(HunterConsumerAopTargetUtils.getTarget(bean), beanName);
                } catch (Exception e) {
                    throw new HunterInfoException("HunterConsumer获取被代理对象失败 beanName=" + beanName, e);
                }
            } else {
                initExecUnit(bean, beanName);
            }
        }
    }

    private static void initExecUnit(Object bean, String beanName) throws BeansException {
        log.info("initBean registUnitCount={} beanName={}", HunterConsumerHelper.getRegistUnitList().size(), beanName);
        Method[] methods = ReflectionUtils.getAllDeclaredMethods(bean.getClass());
        if (methods != null) {
            for (Method method : methods) {

                HunterConsumer anno = AnnotationUtils.findAnnotation(method, HunterConsumer.class);
                if (null != anno) {
                    log.info("initBean start beanName={} methodName={} topic={} tags={} group={} namesrvAddr={}",
                            beanName, method.getName(), anno.topic(), Arrays.toString(anno.tags()),
                            anno.group(), anno.namesrvAddr());

                    //构建mq执行单元
                    HunterExecUnit execUnit =
                            new HunterExecUnitBuilder()
                                    .bean(bean).beanName(beanName).anno(anno).method(method).build();
                    checkExecUnit(execUnit);
                    HunterConsumerHelper.addToRegistUnit(execUnit);
                    log.info("initBean end execUnit={}", GUtil.toJson(execUnit));
                }
            }
        }
    }

    private static void checkExecUnit(HunterExecUnit execUnit) throws BeansException {
        if (StringUtils.isBlank(execUnit.getTopic())) {
            throw new HunterConsumerStartException(
                    String.format("topic不能为空 beanName=%s method=%s",
                            execUnit.getBeanName(), execUnit.getMethod().getName()));
        }

        if (StringUtils.isBlank(execUnit.getNamesrvAddr())) {
            throw new HunterConsumerStartException(
                    String.format("nameSrvAddr不能为空 beanName=%s method=%s",
                            execUnit.getBeanName(), execUnit.getMethod().getName()));
        }

        if (StringUtils.isBlank(execUnit.getGroup())) {
            throw new HunterConsumerStartException(
                    String.format("group不能为空 beanName=%s method=%s",
                            execUnit.getBeanName(), execUnit.getMethod().getName()));
        }
    }

2.3 启动前检查

组件为防止一些错误用法，将做限定检查（具体检查代码不再展示）：

1，同一个group下不能有两个不同的topic

2，同一个group和topic下不能出现在多个@HunterConsumer注解中（防止在本地直连线上mq服务器时修改线上group的订阅tag，进而导致mq消息丢失）

3，广播消费模式下，tags必须是空或者{}或者{*}

4，无法获取到被@HunterConsumer注解的方法参数名称

第四点需要重点说明下，这里@HunterConsumer使用到了Java8开始有的一个特性，即可以获取到源码中方法的参数名，所以引入@HunterConsumer组件的工程需要保证在编译是使用-parameters参数。以Maven为例，需要引入配置：

<plugins>
    <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-compiler-plugin</artifactId>
        <version>3.1</version>
        <configuration>
            <source>1.8</source>
            <target>1.8</target>
            <encoding>UTF-8</encoding>
            <compilerArgument>-parameters</compilerArgument>
        </configuration>
    </plugin>
</plugins>

2.4 根据配置创建并启动Consmer客户端

创建并启动客户端：

new HunterRegistUnitCompletor().registUnit(registUnit).complete();
registUnit.getConsumer().start();

创建并启动客户端的具体过程：


    public DefaultMQPushConsumer createConsumer(HunterRegistUnit registUnit) throws MQClientException {
        HunterExecUnit execUnit = registUnit.getExecUnitList().toArray(new HunterExecUnit[0])[0];
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
        consumer.setSubscribeTopic(registUnit.getTopic());
        consumer.subscribe(registUnit.getTopic(), getSubExpression(registUnit.getTags()));
        consumer.setConsumerGroup(registUnit.getGroup());
        consumer.setNamesrvAddr(registUnit.getNamesrvAddr());
        //设置消费模式，默认集群模式
        if (execUnit.getAnno().messageModel() != null) {
            consumer.setMessageModel(execUnit.getAnno().messageModel());
        }
        return consumer;
    }
    
    public HunterRegistUnit complete() throws MQClientException {
        registUnit.setListener(new HunterMqListener(registUnit));
        registUnit.setConsumer(createConsumer(registUnit));
        registUnit.getConsumer().registerMessageListener(registUnit.getListener());

        ConsumeFromWhere consumeFromWhere = getConsumeFromWhere(registUnit.getExecUnitList());
        if (consumeFromWhere != null) {
            registUnit.setConsumeFromWhere(consumeFromWhere);
            registUnit.getConsumer().setConsumeFromWhere(consumeFromWhere);
        }

        int max = getConsumeThreadMax(registUnit.getExecUnitList());
        if (max > 0) {
            registUnit.setConsumeThreadMax(max);
            registUnit.getConsumer().setConsumeThreadMax(max);
        }

        int min = getConsumeThreadMin(registUnit.getExecUnitList());
        if (min > 0) {
            registUnit.setConsumeThreadMin(max);
            registUnit.getConsumer().setConsumeThreadMin(max);
        }

        int size = getPullBatchSize(registUnit.getExecUnitList());
        if (size > 0) {
            registUnit.setPullBatchSize(size);
            registUnit.getConsumer().setPullBatchSize(size);
        }
        return registUnit;
    }

消费并监控失败信息：

public class HunterMqListener implements MessageListenerConcurrently {
    public ConsumeConcurrentlyStatus consumeMessage(
            List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        Iterator<MessageExt> it = msgs.iterator();
        while (it.hasNext()) {
            MessageExt msgExt = it.next();
            try {
                initLogContext(msgExt);

                //消费失败则立即返回失败
                HunterExecHelper.printBeforeStartProcess(registUnit, msgExt);
                if (!HunterExecHelper.processOneMq(msgExt, context, registUnit)) {
                    HunterExecHelper.printFailDetail(registUnit, msgExt);
                    upToPrometheus("hunter_consume_fail", registUnit, msgExt);
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                } else {
                    HunterExecHelper.printSuccessLog(registUnit, msgExt);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            } catch (Throwable e) {
                HunterExecHelper.printErrorDetail(registUnit, msgExt, e);
                //异常后上传到普罗米休斯监控平台
                upToPrometheus("hunter_consume_fail", registUnit, msgExt);
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            } finally {
                destroyLogContext();
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}

2.5 销毁

RocketMQ客户端要在Spring容器关闭时销毁，业务侧无感知。

@Component("com.zhuanzhuan.hunter.mq.consumer.HunterConsumerStarter")
@Slf4j
public class HunterConsumerStarter implements DisposableBean {
    @Override
    public void destroy() throws Exception {
        List<HunterRegistUnit> registUnitList = HunterConsumerHelper.getRegistUnitList();
        for (HunterRegistUnit registUnit : registUnitList) {
            try {
                if (registUnit.getConsumer() != null) {
                    registUnit.getConsumer().shutdown();
                }
            } catch (Exception e) {
                log.error("destroy shutdown mq consumer error", e);
            }
        }
    }

3.HunterProducer的工作原理

HunterProducer组件也是依托于Spring容器来管理bean的生命周期，所以在业务侧在引入该组件时也需要使用Spring的ApplicationContext来进行初始化，具体如下：

@Component
public class ProjectInit  implements ApplicationListener<ContextRefreshedEvent> {

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        try {
            HunterProducerStarter.startProducers(event.getApplicationContext());
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(1);
        }
    }
}

HunterProducerStarter.startProducers内部在初始化时总体步骤与HunterConsumer组件相同：读取Spring容器中的全部配置、加载Spring容器中的全部HunterProducer注解、启动前检查、根据配置创建并启动Producer客户端。以下四个阶段不再单独介绍：

1，读取Spring容器中的全部配置阶段与HunterConsumer完全相同

2，加载Spring容器中的全部HunterProducer注解阶段是通过扫描Spring容器中bean的全部字段上是否有@HunterProducer注解来实现的，可以参照@HunterConsmuer扫描bean的全部方法

3，启动前检查阶段是检查注解参数是否为空、是否重复等，较为简单

4，销毁阶段与HunterConsumer完全相同

HunterProducer与HunterConsmuer的最大的不同点在于根据配置创建并启动Producer客户端阶段，这一阶段要给调用方提供操作接口，而发送延迟消息与普通消息在操作接口上又有略微区别，所以这里提取出来IProducer接口作为公共的父接口来定义公共方法，IHunterProduer和IHunterDelayProducer作为子接口，来提供个性化方法，具体如下：

//公共发送方法
public interface IProducer {
    SendResult send(Message msg);
    SendResult send(String tag, Object body);
    void sendCallback(Message msg, HunterSendCallback callback) throws HunterProducerException;
    void sendCallback(String tag, Object body, HunterSendCallback callback) throws HunterProducerException;
}

//普通消息发送
public interface IHunterProducer extends IProducer {

}

//延迟消息发送
public interface IHunterDelayProducer extends IProducer {

    SendResult send(String tag, Object body, int delay, TimeUnit timeUnit);

    SendResult send(String tag, Object body, Date delayEndDate);

    void sendCallback(String tag, Object body, int delay, TimeUnit timeUnit, HunterSendCallback callback);

    void sendCallback(String tag, Object body, Date delayEndDate, HunterSendCallback callback);
}

当我们要发送不同的消息时可以使用不同字段类型来获取具体的客户端实例：

@HunterProducer(topic = "testTopic", group = "testProducerGroup")
private IHunterProducer testTopicProducer;


@HunterProducer(topic = "testDelayTopic", group = "testDelayProducerGroup")
private IHunterDelayProducer testTopicDelayProducer;

组件内部再通过实现IHunterProduer和IHunterDelayProducer接口的实例，转调RocketMQ原生客户端DefaultMQProducer来做消息发送。

4.总结

HunterConsumer和HunterProducer组件主要是利用AOP思想实现，它使开发人员在编写业务逻辑时可以专心于核心业务，而不用过多的关注于一些非业务的重复代码，这不但提高了开发效率，而且增强了代码的可维护性。

作者简介：张海威，采货侠暗拍业务开发工程师

想了解更多转转公司的业务实践，欢迎点击关注下方公众号：

继续滑动看下一个