Spring整合RocketMQ(Version 1.3)

ゝ一世哀愁。 2022-05-27 00:44 167阅读 0赞

相对于Version1.2版本，主要是增加了一个**消费失败处理器**，当消费失败时可执行指定的处理方法。当消费次数超过最大消费次数时，可做相应处理。RocketMQ在  
Version 3.4 .9 后可指定消息的最大消费次数，若未指定时，默认最大消费次数为16，当超过最大消费次数时，放入死信队列。16的次数有点多了，可以在配置Consumer时设置下，减小其值，这样当多次消费失败时，能及时感知到，对消息进行及时处理。

先看整合框架结构图：

![这里写图片描述][SouthEast]

ConsumeFailureHandler：消费失败处理器

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
    import org.apache.rocketmq.common.message.MessageClientExt;
    
    /** * 消费失败处理器 * * @author Administrator * @create 2018-04-08 18:53 */
    public interface ConsumeFailureHandler { 
    
        /** * 处理有序消费失败情况 * * @param msg * @param context * @param maxReconsumeTimes 消费最大消费次数 * @return 处理后的结果 */
        ConsumeOrderlyStatus handlerOrderlyFailure(MessageClientExt msg, ConsumeOrderlyContext context, int maxReconsumeTimes);
    
        /** * 处理有序消费失败情况 * * @param msg * @param context * @param maxReconsumeTimes 消费最大消费次数 * @param ex * @return 处理后的结果 */
        ConsumeOrderlyStatus handlerOrderlyFailure(MessageClientExt msg, ConsumeOrderlyContext context, int maxReconsumeTimes, Exception ex);
    
        /** * 处理并发消费失败情况 * * @param msg * @param context * @param maxReconsumeTimes 消费最大消费次数 * @return 处理后的结果 */
        ConsumeConcurrentlyStatus handlerConcurrentlyFailure(MessageClientExt msg, ConsumeConcurrentlyContext context, int maxReconsumeTimes);
    
        /** * 处理并发消费失败情况 * * @param msg * @param context * @param maxReconsumeTimes 消费最大消费次数 * @param ex * @return 处理后的结果 */
        ConsumeConcurrentlyStatus handlerConcurrentlyFailure(MessageClientExt msg, ConsumeConcurrentlyContext context, int maxReconsumeTimes, Exception ex);
    
    }

当消费失败时，消息监听器会执行相应的处理方法，将其返回结果作为最终的消息处理结果。

import java.lang.reflect.Method;
    import java.util.List;
    
    import com.bob.common.utils.rocket.ann.RocketListener;
    import com.bob.common.utils.rocket.handler.ConsumeFailureHandler;
    import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
    import org.apache.rocketmq.common.message.MessageClientExt;
    import org.apache.rocketmq.common.message.MessageExt;
    import org.springframework.util.ReflectionUtils;
    
    /** * 封装{@link RocketListener}方法,生成有序的消息监听器 * * @author wb-jjb318191 * @create 2018-04-02 16:24 */
    public class OrderlyMessageListener extends AbstractMessageListener implements MessageListenerOrderly { 
    
        public OrderlyMessageListener(Object consumeBean, Method consumeMethod, ConsumeFailureHandler failureHandler) {
            super(consumeBean, consumeMethod, failureHandler);
        }
    
        @Override
        public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
            Object[] args = buildConsumeArguments(msgs, context);
            boolean result;
            Exception ex = null;
            try {
                result = (boolean)ReflectionUtils.invokeMethod(consumeMethod, consumeBean, args);
                if (result) {
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            } catch (Exception e) {
                ex = e;
            }
            MessageClientExt msg = (MessageClientExt)msgs.get(0);
            warnWhenConsumeFailed(msg, ex);
            if (ex == null) {
                return failureHandler.handlerOrderlyFailure(msg, context, getMaxReconsumeTimes());
            } else {
                return failureHandler.handlerOrderlyFailure(msg, context, getMaxReconsumeTimes(), ex);
            }
        }
    }
    
    import java.lang.reflect.Method;
    import java.util.List;
    
    import com.bob.common.utils.rocket.ann.RocketListener;
    import com.bob.common.utils.rocket.handler.ConsumeFailureHandler;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import org.apache.rocketmq.common.message.MessageClientExt;
    import org.apache.rocketmq.common.message.MessageExt;
    import org.springframework.util.ReflectionUtils;
    
    /** * 封装{@link RocketListener}标识的方法成一个消息并发消费器 * * @author wb-jjb318191 * @create 2018-03-20 10:03 */
    public class ConcurrentlyMessageListener extends AbstractMessageListener implements MessageListenerConcurrently { 
    
        public ConcurrentlyMessageListener(Object consumeBean, Method consumeMethod, ConsumeFailureHandler failureHandler) {
            super(consumeBean, consumeMethod, failureHandler);
        }
    
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            Object[] args = buildConsumeArguments(msgs, context);
            boolean result;
            Exception ex = null;
            try {
                result = (boolean)ReflectionUtils.invokeMethod(consumeMethod, consumeBean, args);
                if (result) {
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            } catch (Exception e) {
                ex = e;
            }
            MessageClientExt msg = (MessageClientExt)msgs.get(0);
            warnWhenConsumeFailed(msg, ex);
            if (ex == null) {
                return failureHandler.handlerConcurrentlyFailure(msg, context, getMaxReconsumeTimes());
            } else {
                return failureHandler.handlerConcurrentlyFailure(msg, context, getMaxReconsumeTimes(), ex);
            }
        }
    
    }

在定义@RocketListener时，可以指定失败处理器，默认情况下使用ConsumeFailureHandlerAdapter。

import java.lang.annotation.Documented;
    import java.lang.annotation.ElementType;
    import java.lang.annotation.Retention;
    import java.lang.annotation.RetentionPolicy;
    import java.lang.annotation.Target;
    
    import com.bob.common.utils.rocket.handler.ConsumeFailureHandler;
    import com.bob.common.utils.rocket.handler.ConsumeFailureHandlerAdapter;
    
    /** * RocketMQ消费者监听 * * @author wb-jjb318191 * @create 2018-03-20 9:19 */
    @Documented
    @Retention(RetentionPolicy.RUNTIME)
    @Target({ElementType.METHOD})
    public @interface RocketListener { 
    
        /** * 消费者组 * * @return */
        String consumerGroup() default "";
    
        /** * @return */
        String topic() default "";
    
        /** * @return */
        String tag() default "";
    
        /** * @return */
        String namesrvAddr() default "";
    
        /** * 是否有序 * * @return */
        boolean orderly() default false;
    
        /** * 配置Properties文件名称 * * @return */
        String configProperties() default "";
    
        /** * 消费失败处理器 BeanName * * @return */
        String faliureHandler() default "";
    
    }
    
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
    import org.apache.rocketmq.common.message.MessageClientExt;
    
    /** * 消费失败处理器适配器 * * @author Administrator * @create 2018-04-08 18:59 */
    public class ConsumeFailureHandlerAdapter implements ConsumeFailureHandler { 
    
        @Override
        public ConsumeOrderlyStatus handlerOrderlyFailure(MessageClientExt msg, ConsumeOrderlyContext context, int maxReconsumeTimes) {
            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
        }
    
        @Override
        public ConsumeOrderlyStatus handlerOrderlyFailure(MessageClientExt msg, ConsumeOrderlyContext context, int maxReconsumeTimes, Exception ex) {
            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
        }
    
        @Override
        public ConsumeConcurrentlyStatus handlerConcurrentlyFailure(MessageClientExt msg, ConsumeConcurrentlyContext context, int maxReconsumeTimes) {
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    
        @Override
        public ConsumeConcurrentlyStatus handlerConcurrentlyFailure(MessageClientExt msg, ConsumeConcurrentlyContext context, int maxReconsumeTimes, Exception ex) {
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }

其他类的内部结构和上个版本差异不大，就不继续列举了。

下面说明下使用方法，在定义消费者时，可以指定其消息的最大消费次数：

rocket-concurrently-config.properties:
    
        namesrvAddr=127.0.0.1:9876
        instanceName=concurrently
        vipChannelEnabled=false
        consumerGroup=my-group
        topic=my-topic
        tag=*
        consumeFromWhere=CONSUME_FROM_LAST_OFFSET
        consumeThreadMin=4
        consumeThreadMax=4
        pullBatchSize=3
        #自定义消息最大重消费次数,超过此数值时消息被加入死信队列
        maxReconsumeTimes=1

使用@RocketListener标识消费方法时，可以指定消费失败处理器：

@Configuration
    public class RocketConsumerConfiguration { 
    
    
        @Bean
        public DefaultConsumeFailureHandler defaultConsumeFailureHandler(){
            return new DefaultConsumeFailureHandler();
        }
    
        /** * 定义RocketMQ消费器 * * @param msg * @param context * @return true:消费成功; false:消费失败,发回给Broker,一段时间后重试 */
        @RocketListener(configProperties = "rocket-concurrently-config.properties", faliureHandler = "defaultConsumeFailureHandler")
        public boolean concurrently(MessageClientExt msg, ConsumeConcurrentlyContext context) {
            String delay = msg.getProperty(PROPERTY_DELAY_TIME_LEVEL);
            int reconsumeTimes = msg.getReconsumeTimes();
            System.out.println(String.format("延迟级别: [%s], 重消费次数:[%d], 消费内容为[%s]", delay, reconsumeTimes, new String(msg.getBody())));
            throw new IllegalArgumentException("测试消费时抛出异常");
        }
    
    }
    
    
    import com.bob.common.utils.rocket.handler.ConsumeFailureHandlerAdapter;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import org.apache.rocketmq.common.message.MessageClientExt;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    /** * 消费失败处理器 * * @author Administrator * @create 2018-04-08 19:40 */
    public class DefaultConsumeFailureHandler extends ConsumeFailureHandlerAdapter { 
    
        private static final Logger LOGGER = LoggerFactory.getLogger(DefaultConsumeFailureHandler.class);
    
        @Override
        public ConsumeConcurrentlyStatus handlerConcurrentlyFailure(MessageClientExt msg, ConsumeConcurrentlyContext context, int maxReconsumeTimes) {
            if (msg.getReconsumeTimes() == maxReconsumeTimes) {
                LOGGER.info("当前消息即将被纳入死信队列");
            }
            LOGGER.info("自定义消费失败处理器");
            return super.handlerConcurrentlyFailure(msg, context, maxReconsumeTimes);
        }
    
        @Override
        public ConsumeConcurrentlyStatus handlerConcurrentlyFailure(MessageClientExt msg, ConsumeConcurrentlyContext context, int maxReconsumeTimes, Exception ex) {
            LOGGER.info("自定义消费失败处理器");
            return super.handlerConcurrentlyFailure(msg, context, maxReconsumeTimes, ex);
        }
    }

最后源码地址奉上：[https://pan.baidu.com/s/1kdAENIrpB2d1s5hRtVW15A][https_pan.baidu.com_s_1kdAENIrpB2d1s5hRtVW15A]

[SouthEast]: /images/20220527/a230c10d01744000ac8c71915babf9d4.png
[https_pan.baidu.com_s_1kdAENIrpB2d1s5hRtVW15A]: https://pan.baidu.com/s/1kdAENIrpB2d1s5hRtVW15A