|
public class DefaultMessageListenerConcurrently implements MessageListenerConcurrently { |
|
|
|
@SuppressWarnings("unchecked") |
|
@Override |
|
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { |
|
for (MessageExt messageExt : msgs) { |
|
log.debug("received msg: {}", messageExt); |
|
try { |
|
long now = System.currentTimeMillis(); |
|
handleMessage(messageExt); |
|
long costTime = System.currentTimeMillis() - now; |
|
log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime); |
|
} catch (Exception e) { |
|
log.warn("consume message failed. messageId:{}, topic:{}, reconsumeTimes:{}", messageExt.getMsgId(), messageExt.getTopic(), messageExt.getReconsumeTimes(), e); |
|
context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume); |
|
return ConsumeConcurrentlyStatus.RECONSUME_LATER; |
|
} |
|
} |
|
|
|
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; |
|
} |
|
} |
|
public class DefaultMessageListenerOrderly implements MessageListenerOrderly { |
|
|
|
@SuppressWarnings("unchecked") |
|
@Override |
|
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) { |
|
for (MessageExt messageExt : msgs) { |
|
log.debug("received msg: {}", messageExt); |
|
try { |
|
long now = System.currentTimeMillis(); |
|
handleMessage(messageExt); |
|
long costTime = System.currentTimeMillis() - now; |
|
log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime); |
|
} catch (Exception e) { |
|
log.warn("consume message failed. messageId:{}, topic:{}, reconsumeTimes:{}", messageExt.getMsgId(), messageExt.getTopic(), messageExt.getReconsumeTimes(), e); |
|
context.setSuspendCurrentQueueTimeMillis(suspendCurrentQueueTimeMillis); |
|
return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT; |
|
} |
|
} |
|
|
|
return ConsumeOrderlyStatus.SUCCESS; |
|
} |
|
} |
在更新了#617 pr后，我尝试aop切面handleMessage方法，但是没有作用，内部类调用外部类好像并不会走bean调用，导致aop失效，修改内部类调用handleMessage为bean调用，使aop切面handleMessage方法可以正常使用。
After updating PR #617, I tried to use AOP to intercept the handleMessage method, but it didn't work. It seems that when an inner class calls a method of the outer class, it doesn't go through the bean invocation, causing the AOP to fail. Need to modify the inner class to call handleMessage as a bean invocation, so that the AOP interception of the handleMessage method can work properly.