Spring Boot 整合 RabbitMQ
Spring Boot 整合 RabbitMQ 是非常容易，只需要两个步骤。
首先，在 pom.xml 中增加 RabbitMQ 依赖。
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
第二步，在 src/main/resources/application.properties 中配置信息。
#rabbitmq
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
实战演练
一个简单的实战开始
我们来实现一个简单的发送、接收消息。
Configuration
在 Spring Boot 中使用 @Bean 注册一个队列。
@Configuration
public class RabbitMQConfig {
public static final String QUEUE_NAME = "spring-boot-simple";
@Bean
public Queue queue() {
return new Queue(QUEUE_NAME);
}
}
消息生产者
创建消息生产者 Sender。通过注入 AmqpTemplate 接口的实例来实现消息的发送。
@Service
public class Sender {
@Autowired
private AmqpTemplate rabbitTemplate;
public void send() {
System.out.println("梁桂钊 发送消息...");
rabbitTemplate.convertAndSend(RabbitMQConfig.QUEUE_NAME, "你好， 梁桂钊!");
}
}
消息消费者
创建消息消费者 Receiver。通过 @RabbitListener 注解定义对队列的监听。
@Service
public class Receiver {
@Autowired
private AmqpTemplate rabbitTemplate;
@RabbitListener(queues = "spring-boot-simple")
public void receiveMessage(String message) {
System.out.println("Received <" + message + ">");
}
}
运行
@SpringBootApplication
@EnableAutoConfiguration
@ComponentScan(basePackages = { "com.lianggzone.springboot" })
public class RunMain {
public static void main(String[] args) {
SpringApplication.run(RunMain.class, args);
}
}
单元测试
创建单元测试用例
public class RabbitMQTest {
@Autowired
private Sender sender;
@Test
public void send() throws Exception {
sender.send();
}
}
路由的实战演练
经过上面的实战案例，我们对 Spring Boot 整合 RabbitMQ 有了一定的了解。现在，我们再来看下 RabbitMQ 路由场景。
Configuration
在 RabbitMQConfig 中，我们注册 队列，转发器，监听等。
@Configuration
public class RabbitMQConfig2 {
public static final String QUEUE_NAME = "spring-boot";
public static final String QUEUE_EXCHANGE_NAME = "spring-boot-exchange";
@Bean
public Queue queue() {
// 是否持久化
boolean durable = true;
// 仅创建者可以使用的私有队列，断开后自动删除
boolean exclusive = false;
// 当所有消费客户端连接断开后，是否自动删除队列
boolean autoDelete = false;
return new Queue(QUEUE_NAME, durable, exclusive, autoDelete);
}
@Bean
public TopicExchange exchange() {
// 是否持久化
boolean durable = true;
// 当所有消费客户端连接断开后，是否自动删除队列
boolean autoDelete = false;
return new TopicExchange(QUEUE_EXCHANGE_NAME, durable, autoDelete);
}
@Bean
public Binding binding(Queue queue, TopicExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(QUEUE_NAME);
}
@Bean
SimpleMessageListenerContainer container(ConnectionFactory connectionFactory,
MessageListenerAdapter listenerAdapter) {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
container.setQueueNames(QUEUE_NAME);
container.setMessageListener(listenerAdapter);
return container;
}
@Bean
MessageListenerAdapter listenerAdapter(Receiver receiver) {
return new MessageListenerAdapter(receiver, "receiveMessage");
}
}
消息生产者
创建消息生产者 Sender。通过注入 AmqpTemplate 接口的实例来实现消息的发送。
@Service
public class Sender {
@Autowired
private AmqpTemplate rabbitTemplate;
public void send() {
System.out.println("梁桂钊 发送消息...");
rabbitTemplate.convertAndSend(RabbitMQConfig2.QUEUE_NAME, "你好， 梁桂钊!");
}
}
消息消费者
创建消息消费者 Receiver。通过 @RabbitListener 注解定义对队列的监听。
@Service
public class Receiver {
public void receiveMessage(String message) {
System.out.println("Received <" + message + ">");
}
}
运行
@SpringBootApplication
@EnableAutoConfiguration
@ComponentScan(basePackages = { "com.lianggzone.springboot" })
public class RunMain {
public static void main(String[] args) {
SpringApplication.run(RunMain.class, args);
}
}
单元测试
创建单元测试用例
public class RabbitMQTest {
@Autowired
private Sender sender;
@Test
public void send() throws Exception {
sender.send();
}
}
写在最后：欢迎留言讨论，加关注，持续更新！！！