要在Spring Boot项目中实现一个通用的消息消费服务，可以将前面的概念整合并利用Spring的依赖注入特性来创建一个更灵活、可配置的服务。下面是如何创建这样的服务，包括通过application.properties来配置连接信息，以及使用@Service注解定义消费服务。

步骤 1: 配置application.properties

首先，在application.properties文件中添加RocketMQ的配置信息。

# RocketMQ 配置
rocketmq.endpoint=你的RocketMQ接入点
rocketmq.username=你的RocketMQ用户名
rocketmq.password=你的RocketMQ密码

步骤 2: 创建RocketMQConsumerService

接下来，定义RocketMQConsumerService服务类。这个类将读取application.properties中的配置，并提供一个方法来启动消息消费者。

package com.aliyun.openservices;import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.apache.rocketmq.client.apis.StaticSessionCredentialsProvider;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;import java.util.Collections;@Service
public class RocketMQConsumerService {@Value("${rocketmq.endpoint}")private String endpoint;@Value("${rocketmq.username}")private String username;@Value("${rocketmq.password}")private String password;public void startConsumer(String topicName, String filterExpression, String consumerGroupId) throws ClientException {ClientServiceProvider provider = ClientServiceProvider.loadService();ClientConfiguration configuration = ClientConfiguration.newBuilder().setEndpoints(endpoint).setCredentialProvider(new StaticSessionCredentialsProvider(username, password)).build();FilterExpression expression = new FilterExpression(filterExpression, FilterExpressionType.TAG);PushConsumer consumer = provider.newPushConsumerBuilder().setClientConfiguration(configuration).setConsumerGroup(consumerGroupId).setSubscriptionExpressions(Collections.singletonMap(topicName, expression)).setMessageListener(messageView -> {// 实现你的消息处理逻辑System.out.println("Received message: " + messageView.toString());return ConsumeResult.SUCCESS;}).build();// 注意: 实际应用中你可能需要更优雅的方式来启动和关闭Consumer}
}

步骤 3: 使用服务

最后，你可以在Spring Boot应用的任何地方注入并使用RocketMQConsumerService服务。例如，在一个配置类或启动监听器中启动消费者：

package com.aliyun.openservices;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;@Component
public class RocketMQConsumerRunner implements CommandLineRunner {@Autowiredprivate RocketMQConsumerService rocketMQConsumerService;@Overridepublic void run(String... args) throws Exception {// 启动消费者rocketMQConsumerService.startConsumer("topicName", "*", "consumerGroupId");}
}

这个CommandLineRunner实现确保了当Spring Boot应用启动时，会自动启动消息消费服务。你需要根据实际情况替换topicName、filterExpression（这里用*表示接收所有消息）和consumerGroupId的值。

通过这种方式，你可以轻松地在Spring Boot应用中集成和使用RocketMQ的消费服务，同时保持高度的灵活性和配置能力。