Spring Boot 集成RabbitMQ

RabbitMQ is an open source multi-protocol messaging broker.

前言

参照官方Messaging with RabbitMQ，记录在实战中的一些坑。

搭建RabbitMQ服务

本文使用Docker搭建MQ服务。Docker部署服务，快捷、方便。

安装镜像

参照docker 安装ubuntu安装镜像

启动镜像

docker run -d -p 15672:15672 -p 5672:5672 rabbitmq:3-management

这里要映射2个端口：15672是Web管理界面的端口；5672是MQ访问的端口。

Web管理界面

http://192.168.99.100:15672/

guest/guest

RabbitMQ服务部署好了。。。

集成

环境

IntelliJ IDEA 2016.3.4

<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.5.1.RELEASE</version>
</parent>

1
2
3
4
5

<java.version>1.8</java.version>

1
2

maven依赖

<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
</dependencies>

1
2
3
4
5
6
7

代码实现

配置

AmqpInitConfig

@Configuration
@ConditionalOnProperty(prefix = "spring.rabbitmq",name = "enable", matchIfMissing = false)
public class AmqpInitConfig {

final static String queueName="spring.boot";

@Bean
public Queue queue(){
return new Queue(queueName,false);
}

@Bean
public TopicExchange exchange(){
return new TopicExchange("spring.boot.exchange");
}

@Bean
public Binding binding(TopicExchange exchange,Queue queue){
return BindingBuilder.bind(queue).to(exchange).with(queueName+".key");
}

@Bean
public SimpleMessageListenerContainer container(ConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter){
SimpleMessageListenerContainer container=new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
container.setMessageListener(listenerAdapter);
container.addQueueNames(queueName);
return container;
}

@Bean
public MessageListenerAdapter listenerAdapter(Receiver receiver){
return new MessageListenerAdapter(receiver,"receiveMessage");
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

接受消息

Receiver

@Component
public class Receiver {
CountDownLatch latch = new CountDownLatch(1);

public void receiveMessage(String message) {
System.out.println("Received <" + message + ">");
latch.countDown();
}

public CountDownLatch getLatch() {
return latch;
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13

发送消息

Runner

@Component
public class Runner implements CommandLineRunner {
private final RabbitTemplate rabbitTemplate;
private final Receiver receiver;
private final ConfigurableApplicationContext context;

public Runner(Receiver receiver, RabbitTemplate rabbitTemplate,
ConfigurableApplicationContext context) {
this.receiver = receiver;
this.rabbitTemplate = rabbitTemplate;
this.context = context;
}

@Override
public void run(String... args) throws Exception {
System.out.println("Sending message...");
rabbitTemplate.convertAndSend("spring.boot.exchange","spring.boot.key", "Hello from RabbitMQ!");
receiver.getLatch().await(  10000, TimeUnit.MILLISECONDS);
context.close();
}

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

启动

AmqpApplication

@SpringBootApplication
@ComponentScan(basePackages = "com.wxs.amqp")
public class AmqpApplication {
public static void main(String[] args) {
SpringApplication.run(AmqpApplication.class,args);
}
}

1
2
3
4
5
6
7

踩过的坑

坑一

发送消息需要制定exchange，如果不指定，不会发送消息。

参考

Messaging with RabbitMQ 
How to use this image Running the daemon