Spring整合RabbitMQ

在订阅模式消息通信中，须要配置交换机，交换机有三种模式：fanout广播、direct定向、topic通配符。咱们这里使用topic通配符模式，这种在企业应用中最为普遍。

（一）消息生产者

 1. 建立消息生产者项目：spring_rabbitmq_producer

 2. 添加依赖　

<dependency>
　　<groupId>org.springframework.amqp</groupId>
　　<artifactId>spring-rabbit</artifactId>
　　<version>2.0.1.RELEASE</version>
</dependency>

 

 

 3.编写spring整合rabbitmq配置文件：spring-rabbitmq-producer.xml 

 1 <!-- 1. 配置链接 -->
 2 <rabbit:connection-factory  3 　　id="connectionFactory"
 4　　 host="127.0.0.1"
 5 　　port="5672"
 6 　　username="pomelo"
 7 　　password="pomelo"
 8 　　virtual-host="/pomelo"
 9 />
10 <!-- 2. 配置队列 -->
11 <rabbit:queue name="myQueue"/>
12 <!-- 3.配置rabbitAdmin -->
13 <rabbit:admin connection-factory="connectionFactory"/>
14 <!-- 4. 配置topic类型exchange；队列绑定到交换机 -->
15 <rabbit:topic-exchange name="myExchange">
16 　　<rabbit:bindings>
17 　　　　<rabbit:binding queue="myQueue" pattern="msg.#"></rabbit:binding>
18 　　</rabbit:bindings>
19 </rabbit:topic-exchange>
20 <!-- 5. 配置消息对象json转换类 -->
21 <bean id="jsonMessageConverter" class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" />
22 <!-- 6. 配置RabbitTemplate -->
23 <rabbit:template 24 　　id="rabbitTemplate"
25 　　connection-factory="connectionFactory"
26 　　exchange="myExchange"
27　　 message-converter="jsonMessageConverter"
28 />

 

 

 4. 发消息测试

 

package com.pomelo.producer; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.context.support.ClassPathXmlApplicationContext; import java.util.HashMap; import java.util.Map; public class Send { 　　public static void main(String[] args) { 　　　　//建立spring容器
　　　　ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("spring-rabbitmq-producer.xml"); 　　　　//从容器中获取对象
　　　　RabbitTemplate template = context.getBean(RabbitTemplate.class); 　　　　// 发送消息
　　　　Map<String,String> map = new HashMap<>(); 　　　　map.put("email","550731230@qq.com"); 　　　　template.convertAndSend("msg.email",map); 　　　　context.close(); 　　} }

 

(二)消息消费者

 1.建立消费者项目：spring_rabbitmq_consumer

 2.添加依赖

 

<dependencies>
　　<dependency>
　　　　<groupId>org.springframework.amqp</groupId>
　　　　<artifactId>spring-rabbit</artifactId>
　　　　<version>2.0.1.RELEASE</version>
　　</dependency>
</dependencies>

 

 

 3.编写消息监听器

 

package com.pomelo.listener; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageListener; import java.io.IOException; @Component public class EmailMessageListener implements MessageListener { 　　private static final ObjectMapper MAPPER = new ObjectMapper(); 　　@Override 　　public void onMessage(Message message) { 　　　　try { 　　　　　　JsonNode jsonNode = MAPPER.readTree(message.getBody()); 　　　　　　String email = jsonNode.get("email").asText(); 　　　　　　System.out.println("获取队列中消息：" + email); 　　　　} catch (IOException e) { 　　　　　　e.printStackTrace(); 　　　　} 　　} }

 

 

 4.编写spring整合rabbitmq配置文件：spring-rabbitmq-consumer.xml

 

<!-- 1. springIOC注解扫描-->
<context:component-scan base-package="com.pomelo.listener"/>
<!-- 2. 配置链接工厂 -->
<rabbit:connection-factory 　　id="connectionFactory" 　　host="127.0.0.1" 　　port="5672" 　　username="pomelo" 　　password="pomelo" 　　virtual-host="/pomelo"
/>
<!-- 3. 配置队列名 -->
<rabbit:queue name="myQueue"/>
<!-- 4.配置rabbitAdmin -->
<rabbit:admin connection-factory="connectionFactory"/>
<!-- 5.配置监听 -->
<rabbit:listener-container connection-factory="connectionFactory">
　　<rabbit:listener ref="emailMessageListener" queue-names="myQueue" />
</rabbit:listener-container>

 

 

 5.运行项目

 

package com.pomelo.consumer; import org.springframework.context.support.ClassPathXmlApplicationContext; import java.io.IOException; public class Consumer { 　　public static void main(String[] args) throws IOException { 　　　　ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("spring-rabbitmq-consumer.xml"); 　　　　System.in.read(); 　　} }