SpringBoot集成RabbitMQ(五)
2019-04-06 本文已影响59人
3d0829501918
一、RabbitMQ
RabbitMQ 即一个消息队列，主要是用来实现应用程序的异步和解耦，同时也能起到消息缓冲，消息分发的作用。
消息中间件最主要的作用是解耦，中间件最标准的用法是生产者生产消息传送到队列，消费者从队列中拿取消息并处理，生产者不用关心是谁来消费，消费者不用关心谁在生产消息，从而达到解耦的目的。在分布式的系统中，消息队列也会被用在很多其它的方面，比如：分布式事务的支持，RPC的调用等等。
对于消息队列，我们一般知道有三个概念：发消息者、队列、收消息者，RabbitMQ 在这个基本概念之上，多做了一层抽象，在发消息者和队列之间,加入了交换器(Exchange)这样发消息者和队列就没有直接联系，转而变成发消息者把消息给交换器,交换器根据调度策略再把消息再给队列。
左侧P代表生产者，也就是往RabbitMQ发消息的程序。
中间即是RabbitMQ，其中绿色的X代表交换机，红色的通道代表队列。
右侧C代表消费者，也就是往RabbitMQ拿消息的程序。
交换机的功能主要是接收消息并且转发到绑定的队列。
交换机类型: Direct类型、Topic类型、Headers类型和Fanout类型。
Direct是RabbitMQ默认的交换机模式,也是最简单的模式.即创建消息队列的时候,指定一个BindingKey.当发送者发送消息的时候,指定对应的Key.当Key和消息队列的BindingKey一致的时候,消息将会被发送到该消息队列中。
Topic 转发信息主要是依据通配符,队列和交换机的绑定主要是依据一种模式(通配符+字符串),而当发送消息的时候,只有指定的Key和该模式相匹配的时候,消息才会被发送到该消息队列中。
Headers是根据一个规则进行匹配,在消息队列和交换机绑定的时候会指定一组键值对规则,而发送消息的时候也会指定一组键值对规则,当两组键值对规则相匹配的时候,消息会被发送到匹配的消息队列中。
Fanout 是路由广播的形式,将会把消息发给绑定它的全部队列,即便设置了key,也会被忽略。
RabbitMQ 即一个消息队列，主要是用来实现应用程序的异步和解耦，同时也能起到消息缓冲，消息分发的作用。
二、Direct类型
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
#配置rabbitmq的安装地址、端口以及账户信息
#注意port 15672是管理端的端口
spring.application.name=spirng-boot-rabbitmq-sender
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
@Configuration
public class RabbitConfig {
@Bean
public Queue helloQueue() {
return new Queue("hello");
}
}
@Component
public class HelloSender {
@Autowired
private AmqpTemplate rabbitTemplate;
public void send() {
String context = "hello " + new Date();
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("hello", context);
}
}
@Component
@RabbitListener(queues = "hello")
public class HelloReceiver {
@RabbitHandler
public void process(String hello) {
System.out.println("Receiver : " + hello);
}
}
@RunWith(SpringRunner.class)
@SpringBootTest
public class Test {
@Autowired
private HelloSender helloSender;
// 发送单条消息
@org.junit.Test
public void contextLoads() {
helloSender.send();
}
}
Sender : hello Sun Jan 06 10:58:18 CST 2019
2019-01-06 10:58:18.530 INFO 24784 --- [ Thread-2]
o.s.a.r.l.SimpleMessageListenerContainer : Waiting for workers to finish.
Receiver : hello Sun Jan 06 10:58:18 CST 2019
发送者和接收者的queue name必须一致，不然接收不到消息。Direct模式相当于一对一模式，一个消息被发送者发送后,会被转发到一个绑定的消息队列中,然后被一个接收者接收。
三、Direct发送对象
public void send(User user) {
System.out.println("Sender object: " + user.toString());
this.rabbitTemplate.convertAndSend("object", user);
}
@RabbitHandler
public void process(User user) {
System.out.println("Receiver object : " + user);
System.out.println("username:"+user.getUsername());
System.out.println("password:"+user.getPassword());
}
// 发送对象
@org.junit.Test
public void sendObj() {
User user = new User();
user.setId(1);
user.setUsername("admin");
user.setPassword("1234");
objectSender.send(user);
}
Sender object: com.example.bean.User@fa5f81c
2019-01-06 11:13:40.175 INFO 26456 --- [ Thread-2]
o.s.a.r.l.SimpleMessageListenerContainer : Waiting for workers to finish.
Receiver object : com.example.bean.User@7b0f425
username:admin
password:1234
四、Topic类型
Topic 是 RabbitMQ中最灵活的一种方式，可以根据routing_key自由的绑定不同的队列。
@Configuration
public class TopicRabbitConfig {
final static String message = "topic.message";
final static String messages = "topic.messages";
// 创建队列
@Bean
public Queue queueMessage() {
return new Queue(TopicRabbitConfig.message);
}
// 创建队列
@Bean
public Queue queueMessages() {
return new Queue(TopicRabbitConfig.messages);
}
// 将对列绑定到Topic交换器
@Bean
TopicExchange exchange() {
return new TopicExchange("topicExchange");
}
// 将对列绑定到Topic交换器
@Bean
Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
}
// 将对列绑定到Topic交换器 采用#的方式
@Bean
Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
}
}
public void send1() {
String context = "hi, i am message 1";
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("topicExchange", "topic.message", context);
}
public void send2() {
String context = "hi, i am messages 2";
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("topicExchange", "topic.messages", context);
}
@Component
@RabbitListener(queues = "topic.message")
public class TopicReceiver {
@RabbitHandler
public void process(String message) {
System.out.println("Topic Receiver1 : " + message);
}
}
发送send1会匹配到topic.#和topic.message 两个Receiver都可以收到消息，发送send2只有topic.#可以匹配所有只有Receiver2监听到消息
@RunWith(SpringRunner.class)
@SpringBootTest
public class Test {
@Autowired
private TopicSender topicSender;
@org.junit.Test
public void send1() {
topicSender.send1();
}
@org.junit.Test
public void send2() {
topicSender.send2();
}
}
发送send1会匹配到topic.#和topic.message 两个Receiver都可以收到消息，发送send2只有topic.#可以匹配所有只有Receiver2监听到消息。
五、Headers类型
配置一个routingKey为credit.bank的消息队列并绑定在creditBankExchange交换机上
配置一个routingKey为credit.finance的消息队列并绑定在creditFinanceExchange交换机上
@Configuration
public class HeadersConfig {
@Bean
public Queue creditBankQueue() {
return new Queue("credit.bank");
}
@Bean
public Queue creditFinanceQueue() {
return new Queue("credit.finance");
}
@Bean
public HeadersExchange creditBankExchange() {
return new HeadersExchange("creditBankExchange");
}
@Bean
public HeadersExchange creditFinanceExchange() {
return new HeadersExchange("creditFinanceExchange");
}
@Bean
public Binding bindingCreditAExchange(Queue creditBankQueue, HeadersExchange creditBankExchange) {
Map<String,Object> headerValues = new HashMap<>();
headerValues.put("type", "cash");
headerValues.put("aging", "fast");
return BindingBuilder.bind(creditBankQueue).to(creditBankExchange).whereAll(headerValues).match();
}
@Bean
public Binding bindingCreditBExchange(Queue creditFinanceQueue, HeadersExchange creditFinanceExchange) {
Map<String,Object> headerValues = new HashMap<>();
headerValues.put("type", "cash");
headerValues.put("aging", "fast");
return BindingBuilder.bind(creditFinanceQueue).to(creditFinanceExchange).whereAny(headerValues).match();
}
}
@Component
public class ApiCreditSender {
@Autowired
private AmqpTemplate rabbitTemplate;
public void creditBank(Map<String, Object> head, String msg){
System.out.println("credit.bank send message: "+msg);
rabbitTemplate.convertAndSend("creditBankExchange", "credit.bank", getMessage(head, msg));
}
public void creditFinance(Map<String, Object> head, String msg){
System.out.println("credit.finance send message: "+msg);
rabbitTemplate.convertAndSend("creditFinanceExchange", "credit.finance", getMessage(head, msg));
}
private Message getMessage(Map<String, Object> head, Object msg){
MessageProperties messageProperties = new MessageProperties();
for (Map.Entry<String, Object> entry : head.entrySet()) {
messageProperties.setHeader(entry.getKey(), entry.getValue());
}
MessageConverter messageConverter = new SimpleMessageConverter();
return messageConverter.toMessage(msg, messageProperties);
}
}
@Component
public class ApiCreditReceive {
@RabbitHandler
@RabbitListener(queues = "credit.bank")
public void creditBank(String msg) {
System.out.println("credit.bank receive message: "+msg);
}
@RabbitHandler
@RabbitListener(queues = "credit.finance")
public void creditFinance(String msg) {
System.out.println("credit.bank receive message: "+msg);
}
}
@RunWith(SpringRunner.class)
@SpringBootTest
public class Test {
@Autowired
private ApiCreditSender sender;
@org.junit.Test
public void test_creditBank_type() {
Map<String,Object> head = new HashMap<>();
head.put("type", "cash");
sender.creditBank(head, "银行授信(部分匹配)");
}
@org.junit.Test
public void test_creditBank_all() {
Map<String,Object> head = new HashMap<>();
head.put("type", "cash");
head.put("aging", "fast");
sender.creditBank(head, "银行授信(全部匹配)");
}
@org.junit.Test
public void test_creditFinance_type() {
Map<String,Object> head = new HashMap<>();
head.put("type", "cash");
sender.creditFinance(head, "金融公司授信(部分匹配)");
}
@org.junit.Test
public void test_creditFinance_all() {
Map<String,Object> head = new HashMap<>();
head.put("type", "cash");
head.put("aging", "fast");
sender.creditFinance(head, "金融公司授信(全部匹配)");
}
}
// 发送者日志
credit.bank send message: 银行授信(部分匹配)
credit.finance send message: 金融公司授信(全部匹配)
credit.bank send message: 银行授信(全部匹配)
credit.finance send message: 金融公司授信(部分匹配)
// 接收者日志
credit.bank receive message: 银行授信(全部匹配)
credit.bank receive message: 金融公司授信(全部匹配)
credit.bank receive message: 金融公司授信(部分匹配)
分析日志：
通过发送者日志中可以看出4个测试方法均已成功发送消息。
通过接收者日志可以看出credit.bank监听的队列有一条消息没有接收到。
ApiCreditSenderTests.test_creditBank_type()为什么发送的消息，没有被处理？
答：因为在HeadersConfig配置类中，creditBankExchange交换机的匹配规则是完全匹配，即header attribute参数必须完成一致。
六、Fanout类型
FanoutExchange交换机是转发消息到所有绑定队列（和routingKey没有关系）,即我们熟悉的广播模式或者订阅模式，给Fanout交换机发送消息，绑定了这个交换机的所有队列都收到这个消息。
@Configuration
public class FanoutRabbitConfig {
// 创建队列
@Bean
public Queue AMessage() {
return new Queue("fanout.A");
}
// 创建队列
@Bean
public Queue BMessage() {
return new Queue("fanout.B");
}
// 创建队列
@Bean
public Queue CMessage() {
return new Queue("fanout.C");
}
// 创建Fanout交换器
@Bean
FanoutExchange fanoutExchange() {
return new FanoutExchange("fanoutExchange");
}
// 将对列绑定到Fanout交换器
@Bean
Binding bindingExchangeA(Queue AMessage, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(AMessage).to(fanoutExchange);
}
// 将对列绑定到Fanout交换器
@Bean
Binding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(BMessage).to(fanoutExchange);
}
// 将对列绑定到Fanout交换器
@Bean
Binding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(CMessage).to(fanoutExchange);
}
}
@Component
public class FanoutSender {
@Autowired
private AmqpTemplate rabbitTemplate;
public void send() {
String context = "hi, fanout msg ";
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("fanoutExchange","", context);
}
}
这里使用了A、B、C三个队列绑定到Fanout交换机上面，发送端的routing_key写任何字符都会被忽略。
@RabbitHandler
public void process(String message) {
System.out.println("fanout Receiver A : " + message);
}
@RunWith(SpringRunner.class)
@SpringBootTest
public class Test {
@Autowired
private FanoutSender fanoutSender;
@org.junit.Test
public void sendFanout () {
fanoutSender.send();
}
}
fanout Receiver B: hi, fanout msg
fanout Receiver A : hi, fanout msg
fanout Receiver C: hi, fanout msg
本文中的代码在这里：
https://github.com/xiaonongOne/springboot-rabbitmq
