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

RabbitMQ 即一个消息队列，主要是用来实现应用程序的异步和解耦，同时也能起到消息缓冲，消息分发的作用。

聊聊RabbitMQ

传统模式

这个流程，全部在主线程完成，注册-》入库-》发送邮件-》发送短信，由于都在主线程，所以要等待每一步完成才能继续执行。由于每一步的操作时间响应时间不固定，所以主线程的请求耗时可能会非常长，如果请求过多，会导致IIS站点巨慢，排队请求，甚至宕机，严重影响用户体验。

 常用方式

这个流程是主线程只做耗时非常短的入库操作，发送邮件和发送短信，会开启2个异步线程，扔进去并行执行，主线程不管，继续执行后续的操作，这种处理方式要远远好过第一种处理方式，极大的增强了请求响应速度，用户体验良好。缺点是，由于异步线程里的操作都是很耗时间的操作，一个请求要开启2个线程，而一台标准配置的ECS服务器支撑的并发线程数大概在800左右，假设一个线程在10秒做完，这个单个服务器最多能支持400个请求的并发，后面的就要排队。出现这种情况，就要考虑增加服务器做负载，尴尬的增加成本。

RabbitMQ

这个流程是，主线程依旧处理耗时低的入库操作，然后把需要处理的消息写进消息队列中，这个写入耗时可以忽略不计，非常快，然后，独立的发邮件子系统，和独立的发短信子系统，同时订阅消息队列，进行单独处理。处理好之后，向队列发送ACK确认，消息队列整条数据删除。这个流程也是现在各大公司都在用的方式，以SOA服务化各个系统，把耗时操作，单独交给独立的业务系统，通过消息队列作为中间件，达到应用解耦的目的，并且消耗的资源很低，单台服务器能承受更大的并发请求。

RabbitMQ简单使用

1、pom文件

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

2、配置文件

 #配置rabbitmq的安装地址、端口以及账户信息#注意port 15672是管理端的端口spring.application.name=spirng-boot-rabbitmq-senderspring.rabbitmq.host=127.0.0.1spring.rabbitmq.port=5672spring.rabbitmq.username=guestspring.rabbitmq.password=guest

3、队列配置

@Configurationpublic class RabbitConfig {@Beanpublic Queue helloQueue() {return new Queue("hello");}}

4、发送者

@Component
public class HelloSender {@Autowiredprivate AmqpTemplate rabbitTemplate;public void send() {String context = "hello " + new Date();System.out.println("Sender : " + context);this.rabbitTemplate.convertAndSend("hello", context);}
}

5、接收者

@Component
@RabbitListener(queues = "hello")
public class HelloReceiver {@RabbitHandlerpublic void process(String hello) {System.out.println("Receiver  : " + hello);}}

6、测试

@RunWith(SpringRunner.class)
@SpringBootTest
public class Test {@Autowiredprivate HelloSender helloSender;// 发送单条消息@org.junit.Testpublic void contextLoads() {helloSender.send();}}

7、测试结果

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模式相当于一对一模式，一个消息被发送者发送后,会被转发到一个绑定的消息队列中,然后被一个接收者接收。

RabbitMQ发送对象

1、发送者

public void send(User user) {System.out.println("Sender object: " + user.toString());this.rabbitTemplate.convertAndSend("object", user);
}

2、接收者

@RabbitHandler
public void process(User user) {System.out.println("Receiver object : " + user);System.out.println("username:"+user.getUsername());System.out.println("password:"+user.getPassword());
}

3、测试方法

// 发送对象
@org.junit.Test
public void sendObj() {User user = new User();user.setId(1);user.setUsername("admin");user.setPassword("1234");objectSender.send(user);
}

4、测试结果

Sender object: com.example.bean.User@fa5f81c2019-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@7b0f425username:adminpassword:1234

Topic

Topic 是 RabbitMQ中最灵活的一种方式，可以根据routing_key自由的绑定不同的队列。

1、Topic配置

@Configuration
public class TopicRabbitConfig {final static String message = "topic.message";final static String messages = "topic.messages";// 创建队列@Beanpublic Queue queueMessage() {return new Queue(TopicRabbitConfig.message);}// 创建队列@Beanpublic Queue queueMessages() {return new Queue(TopicRabbitConfig.messages);}// 将对列绑定到Topic交换器@BeanTopicExchange exchange() {return new TopicExchange("topicExchange");}// 将对列绑定到Topic交换器@BeanBinding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");}// 将对列绑定到Topic交换器 采用#的方式@BeanBinding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");}
}

2、发送者

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);
}

3、接收者

@Component
@RabbitListener(queues = "topic.message")
public class TopicReceiver {@RabbitHandlerpublic void process(String message) {System.out.println("Topic Receiver1  : " + message);}
}

发送send1会匹配到topic.#和topic.message 两个Receiver都可以收到消息，发送send2只有topic.#可以匹配所有只有Receiver2监听到消息

4、测试类

 @RunWith(SpringRunner.class)@SpringBootTestpublic class Test {@Autowiredprivate TopicSender topicSender;@org.junit.Testpublic void send1() {topicSender.send1();}@org.junit.Testpublic void send2() {topicSender.send2();}}

发送send1会匹配到topic.#和topic.message 两个Receiver都可以收到消息，发送send2只有topic.#可以匹配所有只有Receiver2监听到消息。

Headers

1、队列配置

配置一个routingKey为credit.bank的消息队列并绑定在creditBankExchange交换机上

配置一个routingKey为credit.finance的消息队列并绑定在creditFinanceExchange交换机上

@Configuration
public class HeadersConfig {@Beanpublic Queue creditBankQueue() {return new Queue("credit.bank");}@Beanpublic Queue creditFinanceQueue() {return new Queue("credit.finance");}@Beanpublic HeadersExchange creditBankExchange() {return new HeadersExchange("creditBankExchange");}@Beanpublic HeadersExchange creditFinanceExchange() {return new HeadersExchange("creditFinanceExchange");} @Beanpublic 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();}@Beanpublic 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();}
}

2、发送者

@Component
public class ApiCreditSender {@Autowiredprivate 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);}}

3、接收者

@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);}}

4、测试类

@RunWith(SpringRunner.class)
@SpringBootTest
public class Test {@Autowiredprivate ApiCreditSender sender;@org.junit.Testpublic void test_creditBank_type() {Map<String,Object> head = new HashMap<>();head.put("type", "cash");sender.creditBank(head, "银行授信(部分匹配)");}@org.junit.Testpublic void test_creditBank_all() {Map<String,Object> head = new HashMap<>();head.put("type", "cash");head.put("aging", "fast");sender.creditBank(head, "银行授信(全部匹配)");}@org.junit.Testpublic void test_creditFinance_type() {Map<String,Object> head = new HashMap<>();head.put("type", "cash");sender.creditFinance(head, "金融公司授信(部分匹配)");}@org.junit.Testpublic void test_creditFinance_all() {Map<String,Object> head = new HashMap<>();head.put("type", "cash");head.put("aging", "fast");sender.creditFinance(head, "金融公司授信(全部匹配)");}
}

5、测试结果

// 发送者日志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交换机发送消息，绑定了这个交换机的所有队列都收到这个消息。

1、配置队列

 @Configurationpublic class FanoutRabbitConfig {// 创建队列@Beanpublic Queue AMessage() {return new Queue("fanout.A");}// 创建队列@Beanpublic Queue BMessage() {return new Queue("fanout.B");}// 创建队列@Beanpublic Queue CMessage() {return new Queue("fanout.C");}// 创建Fanout交换器@BeanFanoutExchange fanoutExchange() {return new FanoutExchange("fanoutExchange");}// 将对列绑定到Fanout交换器@BeanBinding bindingExchangeA(Queue AMessage, FanoutExchange fanoutExchange) {return BindingBuilder.bind(AMessage).to(fanoutExchange);}// 将对列绑定到Fanout交换器@BeanBinding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {return BindingBuilder.bind(BMessage).to(fanoutExchange);}// 将对列绑定到Fanout交换器@BeanBinding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {return BindingBuilder.bind(CMessage).to(fanoutExchange);}
}

2、发送者

@Component
public class FanoutSender {@Autowiredprivate AmqpTemplate rabbitTemplate;public void send() {String context = "hi, fanout msg ";System.out.println("Sender : " + context);this.rabbitTemplate.convertAndSend("fanoutExchange","", context);}}

3、接收者

这里使用了A、B、C三个队列绑定到Fanout交换机上面，发送端的routing_key写任何字符都会被忽略。

@RabbitHandler
public void process(String message) {System.out.println("fanout Receiver A  : " + message);
}

4、测试类

@RunWith(SpringRunner.class)
@SpringBootTest
public class Test {@Autowiredprivate FanoutSender fanoutSender;@org.junit.Testpublic void sendFanout () {fanoutSender.send();}
}

5、测试结果

fanout Receiver B: hi, fanout msg 
fanout Receiver A  : hi, fanout msg 
fanout Receiver C: hi, fanout msg 

GitHub地址:

https://github.com/xiaonongOne/springboot-rabbitmq