当前位置：首页 > news >正文

RocketMq方便测试，提供一个controller的接口，支持拉取消息，查看消息内容

news 2025/7/19 19:55:50

通过一个REST API接口动态地启动RocketMQ的消费者，并基于传入的参数（topicName，filterExpression，consumerGroupId）决定要监听哪些消息。在Spring Boot项目中，这通常不是推荐的做法，因为消息消费者通常在应用启动时就配置好，并且持续运行，而不是被动态地创建和销毁。

不过，如果确实需要这样做，您可以考虑以下的设计思路：

方案概述

创建一个服务，该服务能够根据传入的参数创建并管理RocketMQ消费者的实例。
设计一个Controller，通过这个Controller接收到的参数来调用上述服务，动态启动消费者。
由于这种设计涉及到动态管理和维护消费者实例，需要注意资源的释放和异常处理。

实现动态消费者管理服务

这个服务将负责根据参数创建和管理RocketMQ消费者实例。

import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;@Service
public class DynamicConsumerService {private final Map<String, PushConsumer> consumerMap = new ConcurrentHashMap<>();private final RocketMQConsumerService rocketMQConsumerService;@Autowiredpublic DynamicConsumerService(RocketMQConsumerService rocketMQConsumerService) {this.rocketMQConsumerService = rocketMQConsumerService;}public void startConsumer(String topicName, String filterExpression, String consumerGroupId) throws ClientException {if (consumerMap.containsKey(consumerGroupId)) {// 可能需要考虑停止或重置已存在的消费者return;}PushConsumer consumer = rocketMQConsumerService.createConsumer(topicName, filterExpression, consumerGroupId);consumer.start();consumerMap.put(consumerGroupId, consumer);}// 停止并移除消费者public void stopConsumer(String consumerGroupId) {PushConsumer consumer = consumerMap.remove(consumerGroupId);if (consumer != null) {consumer.shutdown();}}
}

这里createConsumer方法需要在RocketMQConsumerService中实现，返回一个配置好的PushConsumer实例，这个方法的实现与之前的startConsumer方法类似，但不会自动启动消费者。

实现Controller

然后，实现一个Controller来处理REST API请求，根据请求参数动态启动和停止消费者。

import org.apache.rocketmq.client.apis.ClientException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;@RestController
@RequestMapping("/consumer")
public class DynamicConsumerController {private final DynamicConsumerService dynamicConsumerService;@Autowiredpublic DynamicConsumerController(DynamicConsumerService dynamicConsumerService) {this.dynamicConsumerService = dynamicConsumerService;}@PostMapping("/start")public String startConsumer(@RequestParam String topicName,@RequestParam String filterExpression,@RequestParam String consumerGroupId) {try {dynamicConsumerService.startConsumer(topicName, filterExpression, consumerGroupId);return "Consumer started for group: " + consumerGroupId;} catch (ClientException e) {e.printStackTrace();return "Failed to start consumer: " + e.getMessage();}}@PostMapping("/stop")public String stopConsumer(@RequestParam String consumerGroupId) {dynamicConsumerService.stopConsumer(consumerGroupId);return "Consumer stopped for group: " + consumerGroupId;}
}