一、搭建工程
我们需要搭建一个简单的 可以实现与RabbitMq进行数据交互的Spring Boot工程
首先需要在自己的电脑上搭建RabbitMq环境,可以直接去https://www.rabbitmq.com/download.html官网下载,然后安装(这里不作叙述)
1.创建Spring Boot工程(这里使用的Spring Boot版本为2.1.1.RELEASE),命名为:rabbitmq-spring-boot,并用maven管理引用如下依赖关系:
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
2.创建消息接受者的类,命名为:Receiver.java 代码如下:
package com.example.demo;
import java.util.concurrent.CountDownLatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
@Component
public class Receiver {
private static Logger log = LoggerFactory.getLogger(Receiver.class);
//线程计数器
private CountDownLatch latch = new CountDownLatch(1);
public void receiverMessage(String message) {
log.info("接受到的信息:" + message);
latch.countDown();
}
public CountDownLatch getLatch() {
return latch;
}
}
以上代码使用了CountDownLatch 线程计数器的概念,这里作者只简单介绍一下,线程计数器顾名思义是对线程进行计数用的,通过计数控制线程是否可以继续执行。具体功能可以控制线程的执行顺序,也可以判断线程是否全部执行完毕(作者是这么理解的,如果有补充或不对的地方欢迎大家指出)。通常使用以下两个方法:
countDown():当前线程数减一
await():控制线程等待,直到计数器减到0时,才可以继续往下执行。
3.创建消息发送类,命名为Runner.java,代码如下
package com.example.demo;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
@Component
@Order(value=1)
public class Runner implements CommandLineRunner{
private static Logger log = LoggerFactory.getLogger(Runner.class);
private final RabbitTemplate rabbitTemplate;
private final Receiver receiver;
public Runner(RabbitTemplate rabbitTemplate,Receiver receiver) {
this.rabbitTemplate = rabbitTemplate;
this.receiver = receiver;
}
@Override
public void run(String... args) throws Exception {
log.info("开始试着发送消息......");
rabbitTemplate.convertAndSend(RabbitmqSpringBootApplication.topicExchangeName,"foo.bar.baz","99999999");
rabbitTemplate.convertAndSend(RabbitmqSpringBootApplication.topicExchangeName,"foo.bar.baz","123");
rabbitTemplate.convertAndSend(RabbitmqSpringBootApplication.topicExchangeName,"foo.bar.baz","456");
log.info("发送完毕");
receiver.getLatch().await(10000,TimeUnit.MILLISECONDS);
log.info("run 执行完毕!");
}
}
这里简单介绍一下CommandLineRunner接口:实现该接口的类为启动执行任务,会在Spring Boot启动时创建实例,并在启动完成时,自动调用run方法。如果有多个类都实现了该接口,那么会根据@Order(value=1)注解来判断执行顺序,按照从小到大的顺序执行。
4.更改启动类信息,作者这里是RabbitmqSpringBootApplication.java(找到你那个有@SpringBootApplication注解的类,如果没有就手动创建一个),代码如下:
package com.example.demo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
@SpringBootApplication
public class RabbitmqSpringBootApplication {
private static Logger log = LoggerFactory.getLogger(RabbitmqSpringBootApplication.class);
static final String topicExchangeName = "spring-boot-exchange";
static final String queueName = "spring-boot";
@Bean
Queue queue() {
//第二个参数 是否保持队列持久
Queue queue = new Queue(queueName,false);
return queue;
}
@Bean
TopicExchange exchange() {
TopicExchange echange = new TopicExchange(topicExchangeName);
return echange;
}
@Bean
Binding binding(Queue queue,TopicExchange exchange) {
Binding binding = BindingBuilder.bind(queue).to(exchange).with("foo.bar.#");
return binding;
}
@Bean
SimpleMessageListenerContainer container(ConnectionFactory connectionFactory ,MessageListenerAdapter listennerAdapter) {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
container.setQueueNames(queueName);
container.setMessageListener(listennerAdapter);
return container;
}
@Bean
MessageListenerAdapter listenerAdapter(Receiver receiver) {
//接受消息的监听器,第二个参数时接受消息的方法名
MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(receiver,"receiverMessage");
return messageListenerAdapter;
}
public static void main(String[] args) {
SpringApplication.run(RabbitmqSpringBootApplication.class, args).close();
}
}
针对以上Bean 做简单介绍(如果你很熟的话可以直接忽略)
exchange(交换模式,又称交换机)是MQ消息发送后第一个到达的地方,RabbitMq中一共有四种交换模式:fanout,direct,topic,header,其中header不常用,所以经常只说三种。
- Fanout Exchange:不处理RouteKey 所有队列都复制一份消息 所以转发消息最快
- Direct Exchange:精确匹配RouteKey 队列信息和RouteKey完全匹配时才会被队列接受,否则会被舍弃
- Topic Exchange:模糊匹配,拥有两个通配符:#和*
* 匹配当前路由下的单个子路由 例如:foo.bar.* 可以匹配到foo.bar.a 但不能匹配到foo.bar.a.b
# 匹配当前路由下的所有子路由 例如:foo.bar.# 可以匹配到foo.bar.a 也可匹配到foo.bar.a.b
SimpleMessageListenerContainer(AMQP消息队列容器):可以简单理解为Spring Boot通过该方法订阅rabbitMq.这里可以订阅一个也可以订阅多个。
MessageListenerAdapter (JMS消息监听器):将接受到的消息进行类型转换,然后通过反射的机制,将它交给一个java类处理。
5.到此为止,与RabbitMq的数据交互工程已经创建完毕,运行工程,可以看到如下效果:
二、参考来源: