消费限流 在分布式消息服务RocketMQ中,消费者消费消息时,可能会出现消费过快导致下游业务来不及处理的情况,进而影响系统的稳定性。本章节介绍在消费端进行限流的示例代码,以保障系统的稳定。 import java.util.List; import java.util.concurrent.TimeUnit; import com.google.common.util.concurrent.RateLimiter; import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; 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.client.exception.MQClientException; import org.apache.rocketmq.common.consumer.ConsumeFromWhere; import org.apache.rocketmq.common.message.MessageExt; public class ConsumerLimitExample { public static void main(String[] args) throws InterruptedException, MQClientException { DefaultMQPushConsumer consumer new DefaultMQPushConsumer("GroupTest"); consumer.subscribe("TopicTest", ""); consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUMEFROMFIRSTOFFSET); RateLimiter rateLimiter RateLimiter.create(200); consumer.registerMessageListener(new MessageListenerConcurrently() { @Override public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext context) { if (!rateLimiter.tryAcquire(msgs.size(),3, TimeUnit.SECONDS)) { return ConsumeConcurrentlyStatus.RECONSUMELATER; } System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs); return ConsumeConcurrentlyStatus.CONSUMESUCCESS; } }); consumer.start(); System.out.printf("Consumer Started.%n"); } }