RocketMQ是一个分布式消息中间件，可以用于发送和接收消息。以下是一个使用RocketMQ的简单示例，展示如何在Spring项目中配置和使用RocketMQ。
- 在Spring项目中添加RocketMQ依赖，比如使用Maven:
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.9.0</version>
</dependency>
- 在Spring配置文件中配置RocketMQ的Producer和Consumer:
@Configuration
public class RocketMQConfig {
@Value("${rocketmq.namesrvAddr}")
private String namesrvAddr;
@Value("${rocketmq.producer.group}")
private String producerGroup;
@Value("${rocketmq.consumer.group}")
private String consumerGroup;
@Bean(initMethod = "start", destroyMethod = "shutdown")
public DefaultMQProducer producer() {
DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
producer.setNamesrvAddr(namesrvAddr);
producer.start();
return producer;
}
@Bean(initMethod = "start", destroyMethod = "shutdown")
public DefaultMQPushConsumer consumer() {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
consumer.setNamesrvAddr(namesrvAddr);
consumer.subscribe("TopicTest", "*");
consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
for (MessageExt msg : msgs) {
System.out.println(new String(msg.getBody()));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
consumer.start();
return consumer;
}
}
- 使用Producer发送消息:
@Autowired
private DefaultMQProducer producer;
public void sendMessage(String topic, String tags, String message) throws Exception {
Message msg = new Message(topic, tags, message.getBytes(RemotingHelper.DEFAULT_CHARSET));
SendResult sendResult = producer.send(msg);
System.out.println(sendResult);
}
- 使用Consumer接收消息。
以上代码展示了如何在Spring项目中配置和启动RocketMQ的Producer和Consumer。Producer用于发送消息，Consumer用于接收并处理消息。
注意：在实际应用中，你需要根据自己的RocketMQ服务器地址、生产者和消费者的组名以及主题（Topic）等配置信息来调整配置。同时，消息的发送和接收应该根据实际业务逻辑来进行异常处理和资源管理。