listener示例

  2. import lombok.extern.slf4j.Slf4j;
  3. import org.apache.rocketmq.common.message.MessageExt;
  4. import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
  5. import org.springframework.stereotype.Service;
  7. /**
  8.  * 示例消息
  9.  *
  10.  * @author 微风
  11.  * @date 2020-04-10
  12.  */
  13. @Service
  14. @Slf4j
  15. @RocketMQMessageListener(topic = "topic-demo",
  16.         selectorExpression = "tag-demo",
  17.         consumerGroup = "GID-demo_demo-demo",
  18.         consumeThreadMax = 10,
  19.         consumeTimeout = 10000L)
  20. public class DemoTopicTag2MessageListener extends AbstractMessageListener<MessageExt> {
  22.     @Override
  23.     public void consumerMessage(MessageExt messageExt) throws ServerException {
  24.         if (messageExt == null) {
  25.             return;
  26.         }
  27.         if (messageExt.getTags() == null) {
  28.             throw new ServerException(CommonResponseCode.SYS_E_SERVICE_EXCEPTION, "参数缺失");
  29.         }
  30.     }
  31. }
  2. import com.alibaba.dubbo.config.annotation.Reference;
  3. import com.google.common.cache.CacheBuilder;
  4. import com.google.common.cache.CacheLoader;
  5. import com.google.common.cache.LoadingCache;
  6. import lombok.extern.slf4j.Slf4j;
  7. import org.apache.rocketmq.common.message.MessageExt;
  8. import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
  9. import org.apache.rocketmq.spring.core.RocketMQListener;
  10. import org.springframework.beans.factory.annotation.Autowired;
  11. import org.springframework.core.env.StandardEnvironment;
  13. import javax.annotation.PostConstruct;
  14. import java.lang.reflect.ParameterizedType;
  15. import java.lang.reflect.Type;
  16. import java.nio.charset.Charset;
  17. import java.util.Objects;
  18. import java.util.concurrent.TimeUnit;
  19. import java.util.concurrent.atomic.AtomicInteger;
  21. /**
  22.  * @param <T> 消息体类型
  23.  * @description: 基础消息处理类
  24.  * @author: 微风
  25.  * @create: 2020/04/11 10:32
  26.  */
  27. @Slf4j
  28. public abstract class AbstractMessageListener<T> implements RocketMQListener<MessageExt> {
  30.     @Autowired
  31.     private StandardEnvironment environment;
  33.     private String charset = "UTF-8";
  34.     private String destination;
  35.     private AtomicInteger count = new AtomicInteger(0);
  36.     private static final int MAX_CONCURRENT_CONSUME_COUNT = 100;
  37.     private static final int CONSUME_TIMES = 10;
  38.     private final LoadingCache<String, Integer> localCache = CacheBuilder.newBuilder()
  39.             .expireAfterWrite(30, TimeUnit.SECONDS)
  40.             .build(new CacheLoader<String, Integer>() {
  41.                 @Override
  42.                 public Integer load(String key) {
  43.                     try {
  44.                         Object object = redisService.get(key);
  45.                         if (object == null) {
  46.                             return MAX_CONCURRENT_CONSUME_COUNT;
  47.                         }
  48.                         return Integer.valueOf(object.toString());
  49.                     } catch (Exception e) {
  50.                         return MAX_CONCURRENT_CONSUME_COUNT;
  51.                     }
  52.                 }
  53.             });
  56.     @Override
  57.     public void onMessage(MessageExt msg) {
  58.         if (msg == null) {
  59.             return;
  60.         }
  61.         if (!destination.equals(msg.getTopic() + msg.getTags())) {
  62.             log.warn("消息跟listener不匹配");
  63.             return;
  64.         }
  65.         long startTime = System.currentTimeMillis();
  66.         String messageKey = msg.getTopic() + ":" + msg.getTags() + ":" + msg.getKeys();
  67.         try {
  68.             count.incrementAndGet();
  69.             log.info("接收消息, messageKey:{} body:{} ", messageKey, new String(msg.getBody(), charset));
  70.             if (isDuplicate(msg)) {
  71.                 log.warn("重复消息 messageKey = {}, id={}", messageKey, msg.getMsgId());
  72.                 return;
  73.             }
  74.             // 消费次数以及限流检查
  75.             boolean checkMessage = checkMessage(msg, count);
  76.             if (!checkMessage) {
  77.                 throw new ServerException("预检查失败 messageKey:" + messageKey);
  78.             }
  79.             consumerMessage(doConvertMessage(msg));
  80.             long cost = System.currentTimeMillis() - startTime;
  81.             if (cost > 1000) {
  82.                 log.warn("message consume {} cost:{}", messageKey, startTime);
  83.             }
  84.             if (cost > 3000) {
  85.                 sendDingDing(messageKey + ",消费超过3秒");
  86.             }
  87.         } catch (Exception e) {
  88.             log.error("消费失败 {}", messageKey, e);
  89.             throw new RuntimeException(e.getMessage(), e);
  90.         } finally {
  91.             count.decrementAndGet();
  92.         }
  93.     }
  95.     /**
  96.      * 消费消息
  97.      *
  98.      * @param msg 消息对象
  99.      * @throws ServerException 异常
  100.      */
  101.     public abstract void consumerMessage(T msg) throws ServerException;
  103.     /**
  104.      * 消息转换
  105.      *
  106.      * @param messageExt 原消息体
  107.      * @return T 需要接收的消息对象
  108.      */
  109.     private T doConvertMessage(MessageExt messageExt) {
  110.         Type messageType = getMessageType();
  111.         if (Objects.equals(messageType, MessageExt.class)) {
  112.             return (T) messageExt;
  113.         } else {
  114.             String str = new String(messageExt.getBody(), Charset.forName(charset));
  115.             if (Objects.equals(messageType, String.class)) {
  116.                 return (T) str;
  117.             } else {
  118.                 try {
  119.                     return JsonUtils.parseJson(str, messageType);
  120.                 } catch (Exception e) {
  121.                     log.error("convert failed. str:{}, msgType:{}", str, messageType, e);
  122.                     throw new RuntimeException("cannot convert message to " + messageType, e);
  123.                 }
  124.             }
  125.         }
  126.     }
  128.     /**
  129.      * 获取需要接收的消息对象类型
  130.      *
  131.      * @return Type 消息体类型
  132.      */
  133.     private Type getMessageType() {
  134.         Type targetClass = this.getClass().getGenericSuperclass();
  136.         if (Objects.isNull(targetClass)) {
  137.             return Object.class;
  138.         }
  140.         Type[] actualTypeArguments = ((ParameterizedType) targetClass).getActualTypeArguments();
  141.         if (Objects.nonNull(actualTypeArguments) && actualTypeArguments.length > 0) {
  142.             return actualTypeArguments[0];
  143.         }
  144.         return Object.class;
  145.     }
  147.     /**
  148.      * bean初始化调用
  149.      */
  150.     @PostConstruct
  151.     public void init() {
  153.         RocketMQMessageListener annotation = this.getClass().getAnnotation(RocketMQMessageListener.class);
  154.         String topic = this.environment.resolvePlaceholders(annotation.topic());
  155.         String tag = this.environment.resolvePlaceholders(annotation.selectorExpression());
  157.         destination = topic + tag;
  158.     }
  160.     /**
  161.      * 限流检查
  162.      * <p>
  163.      * 其他listener有自己的消息检查需要可以考虑重写此方法
  164.      * </p>
  165.      *
  166.      * @param msg   消息对象
  167.      * @param count 限流数
  168.      * @return boolean 是否可以消费
  169.      */
  170.     public boolean checkMessage(MessageExt msg, AtomicInteger count) {
  171.         if (msg == null) {
  172.             return true;
  173.         }
  174.         String messageKey = msg.getTopic() + ":" + msg.getTags() + ":" + msg.getKeys();
  175.         try {
  176.             // 消息重试次数超过10次报警
  177.             checkConsumeTimes(msg);
  178.             // 消费限流
  179.             int limitCount = localCache.get(RedisKeyEnum.MQ_FLOW_LIMIT_COUNT.getValue() + "_" + messageKey);
  180.             if (count.get() > limitCount) {
  181.                 log.info("messageKey:{} msg count is more than limit count, count:{}", messageKey, count.get());
  182.                 return false;
  183.             }
  184.             log.info("msg messageKey:{}, count:{}", messageKey, count.get());
  185.         } catch (Exception e) {
  186.             log.error("check error destination: {}", messageKey, e);
  187.             return false;
  188.         }
  190.         return true;
  191.     }
  193.     /**
  194.      * 消息消费幂等通用判断逻辑
  195.      * <p>
  196.      * 有自己的幂等判断需求的重写此实现
  197.      * </p>
  198.      *
  199.      * @param msg 消息体
  200.      * @return 是重复消息返回true，否则返回false
  201.      */
  202.     public boolean isDuplicate(MessageExt msg) {
  203.         return false;
  204.     }
  206.     /**
  207.      * 消息重试次数超过10次报警
  208.      *
  209.      * @param msg 消息对象
  210.      */
  211.     private void checkConsumeTimes(MessageExt msg) {
  212.         if (msg.getReconsumeTimes() >= CONSUME_TIMES) {
  213.             try {
  214.                 String message = msg.getTopic() + ":" + msg.getTags() + "_" + msg.getKeys();
  215.                 sendDingDing("消息" + message + "已重试" + msg.getReconsumeTimes() + "次");
  216.             } catch (Exception e) {
  217.                 log.error("error to sendDingDingMsg, e:{}", e.getMessage(), e);
  218.             }
  219.         }
  220.     }
  222.     /**
  223.      * 消息重试次数超过10次报警
  224.      *
  225.      * @param tip 提示内容
  226.      */
  227.     public void sendDingDing(String tip) {
  228.         //发送提示信息
  229.     }
  230. }

1、消息堆积问题

同一个进程中 ons和rocketmq原生客户端同时订阅一个topic消息的话，会出现阻塞问题，原因是因为这里subversion是用户的时间戳，这样同时存在两个类执行同样的代码，会导致服务端判断出现异常
image.png
服务端判断逻辑
image.png

2、上报轨迹信息问题

使用开源客户端后ons日志出现这种错误信息，跟踪错误堆栈发现，是在上报消息轨迹信息时，发送消息的topic不存在引起
RMQ_SYS_TRACE_TOPIC
image.png

发送轨迹消息时会有一个accessChannel的判断，只有是CLOUD方式才能正确获取topic
image.png