消息生产者端使用 ScheduledExecutorService 来定期重置计数器，以实现限流

本文主要是介绍消息生产者端使用 ScheduledExecutorService 来定期重置计数器，以实现限流，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

使用 ScheduledExecutorService 可以很容易地实现定时任务，比如每秒重置计数器来实现限流。下面是一个改进的示例，展示了如何在消息生产者端使用 ScheduledExecutorService 来定期重置计数器，以实现限流：

java
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

public class LimitedRateProducer {
private static final String TOPIC = “test-topic”;
private static final String TAG = “*”;
private static final long MAX_RATE = 100L; // 每秒最大发送速率
private static final long RESET_INTERVAL = 1000; // 重置计数器的时间间隔，单位毫秒
private final DefaultMQProducer producer;
private final AtomicLong counter = new AtomicLong(0);
private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

public LimitedRateProducer() throws Exception {  producer = new DefaultMQProducer("limited-rate-producer-group");  producer.setNamesrvAddr("localhost:9876");  producer.start();  // 安排一个任务来每秒重置计数器  scheduler.scheduleAtFixedRate(() -> counter.set(0), RESET_INTERVAL, RESET_INTERVAL, TimeUnit.MILLISECONDS);  
}  public void sendMessage(String content) throws Exception {  long currentRate = counter.incrementAndGet();  if (currentRate > MAX_RATE) {  // 超出限流速率，丢弃消息或执行其他策略  counter.decrementAndGet(); // 减少计数器，因为这条消息没有被发送  System.out.println("Message discarded due to rate limit.");  return;  }  Message msg = new Message(TOPIC, TAG, content.getBytes());  producer.send(msg);  
}  public void shutdown() throws Exception {  producer.shutdown();  scheduler.shutdown(); // 关闭调度器，停止所有计划任务  
}  public static void main(String[] args) throws Exception {  LimitedRateProducer producer = new LimitedRateProducer();  // 模拟发送消息  for (int i = 0; i < 1000; i++) {  producer.sendMessage("Hello RocketMQ " + i);  // 休眠一段时间来模拟发送间隔  Thread.sleep(10);  }  producer.shutdown(); // 关闭生产者和调度器  
}

}
在这个示例中，我们创建了一个 ScheduledExecutorService 来定期重置 counter。scheduleAtFixedRate 方法用于安排一个固定频率执行的任务，这里我们设置每 RESET_INTERVAL 毫秒（即每秒）执行一次任务，任务的内容是简单地将 counter 设置为0。

代码注意事项：

异常处理：
- 在 sendMessage 方法中，当调用 producer.send(msg) 时，应该捕获并处理可能抛出的异常。
- 在 shutdown 方法中，也需要处理 producer.shutdown() 和 scheduler.shutdown() 可能抛出的异常。
资源关闭：
- 在 shutdown 方法中，不仅要关闭 scheduler，还要确保 producer 也被正确关闭，并等待关闭操作完成。
- 考虑使用 try-finally 块或 try-with-resources 语句来确保资源被释放。
并发安全：
- 虽然 AtomicLong 提供了线程安全的计数器操作，但如果限流逻辑变得更复杂，可能需要进一步考虑并发控制。
限流精度：
- 使用 ScheduledExecutorService 的定时任务进行限流可能不够精确，特别是在高并发场景下。如果精度要求较高，可能需要考虑使用其他限流算法或工具。
日志记录：
- 在限流逻辑中加入日志记录，有助于监控和调试。当消息被丢弃或限流逻辑被触发时，应该记录相关信息。