在订阅模式消息通讯中,需要配置交换机,交换机有三种模式: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="heima" 7 password="heima" 8 virtual-host="/itcast" 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","610731230@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(); } }
原文:https://www.cnblogs.com/pomelo-lemon/p/11390813.html