0.前言
发现md代码内部不换行的排版问题（复制的代码），是md的缺陷一直没改，在这样下去要放弃md了使用其他编辑器，很影响阅读，所以这篇博客暂时用默认编辑器。
1.新建工程，添加依赖
|
compile('org.springframework.boot:spring-boot-starter-amqp')
|
2.配置application.yml
|
#配置服务端
spring:
application:
name: rabbit-test
rabbitmq:
host: 10.2.6.33
port: 5672
username: gaojinliang
password: 【自己对应的密码】
|
3.RabbitMq的基本概念
![]()
内部实际上也是AMQP中的基本概念，AMQP中增加了Exchange和Binding的角色。生产者把消息发布到Exchange上，消息最终到达队列并被消费者接收，而
Binding决定交换器的消息应该发送到哪个队列。
Exchange 类型
Exchange分发消息的时候根据类型的不同分发策略有区别，目前有四种类型，direct、fanout、topic、headers 。headers 匹配 AMQP 消息的 header
而不是路由键，此外 headers 交换器和 direct 交换器完全一致，但性能差很多，目前几乎用不到了，所以直接看另外三种类型：
3.1direct
消息中的路由键（routing key）如果和Binding中的binding key一致，交换器就将消息发到对应的队列中。路由键与队列名完全匹配，如果一个队列绑定到交换机要求路由键为“dog”，则只转发routing key标记为“dog”的消息，不会转发“dog.puppy”，也不会转发“dog.guard”等等。他是完全匹配，单播的模式。
![]()
以下是direct的场景，也是最简单，最常用的模式。
|
MQ配置
@Configuration
public class RabbitQueueConfig {
@Bean
public Queue queue1() {
return new Queue("queue1", true); //持久化
}
}
发送者
@Autowired
private AmqpTemplate rabbitTemplate;
@GetMapping("/queue")
public String sendQueueMsg(@RequestParam("msg") String msg) {
rabbitTemplate.convertAndSend("queue1", msg);
return "success";
}
接收者
@Component
public class QueueReceiver {
@RabbitListener(queues = "queue1")
@RabbitHandler
public void process(String strs) {
System.out.println("接收到：" + strs);
}
}
|
3.2fanout
每个发到fanout类型交换器的消息都会分到所有绑定的队列上。fanout交换器不处理路由键，只是简单的将队列绑定到交换器上，每个发送到交换器的消息都会被
转发到与该交换器绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。fanout类型转发消息是最快的。
![]()
以下是fanout场景，单个微服务集群的情况下每个节点都绑定到一个direct交换机上 （例子中是服务端口区分队列，也可以是别的key），然后每个节点都能收到
消息做一些配置更新等操作。
|
MQ配置
@Configuration
public class RabbitFanoutConfig {
@Value("${server.port}")
private String port;
@Bean
public FanoutExchange fanoutExchange() {
return new FanoutExchange ("fanoutExchange");
}
@Bean
public Queue fanoutTp() {
return new Queue("fanout.tp."+port);
}
@Bean
public Binding fanoutBinding() {
return BindingBuilder.bind(fanoutTp()).to(fanoutExchange());
}
}
发送者
@GetMapping("/fanout")
public String sendFanoutMsg(@RequestParam("msg") String msg) {
rabbitTemplate.convertAndSend("fanoutExchange","", msg);
return "success";
}
接收者
@Component
public class FanoutRecerver {
@RabbitListener(queues = "fanout.tp.${server.port}")
@RabbitHandler
public void process(String strs) {
System.out.println("fanout.tp接收到：" + strs);
}
}
|
3.3topic
topic交换器通过模式匹配分配消息的路由键属性，将路由键和单个模式进行匹配，此时队列需要绑定到一个模式上。它将路由键和绑定建的字符串切分成单词，这些单词
之间用点隔开。他同样也会识别两个通配符：符号“#”和符号“*”。#匹配0个或多个单词，*匹配不多不少一个单词。
![]()
以下是topic场景，单个微服务接收到同一个消息后多个功能协调工作，也适用按分类更新、同一问题需要特定人员知晓等场景
|
MQ配置：
@Configuration
public class RabbitTopicConfig {
@Bean
public TopicExchange topicExchange() {
return new TopicExchange ("topicExchange");
}
@Bean
public Queue topicTp1() {
return new Queue("topic.tp2.one");
}
@Bean
public Queue topicTp2() {
return new Queue("topic.tp2.two");
}
@Bean
public Binding binding1() {
return BindingBuilder.bind(topicTp1()).to(topicExchange()).with("topic.tp2.*"); //* 后面匹配单个key
}
@Bean
public Binding binding2() {
return BindingBuilder.bind(topicTp2()).to(topicExchange()).with("topic.tp2.#"); //# 后面匹配多个key
}
}
发送者
@GetMapping("/topic")
public String sendTopicMsg(@RequestParam("msg") String msg) {
rabbitTemplate.convertAndSend("topicExchange","topic.tp2.all", msg);
return "success";
}
接收者
@Component
public class TopicRecerver {
@RabbitListener(queues = "topic.tp2.one")
@RabbitHandler
public void process(String strs) {
System.out.println("topic.tp2.one 接收到：" + strs);
}
@RabbitListener(queues = "topic.tp2.two")
@RabbitHandler
public void process2(String strs) {
System.out.println("topic.tp2.two 接收到：" + strs);
}
}
|
4.接收端手动确认
上面提到的内容都是消息自动确认，当我们程序中收到消息处理失败需要MQ重发时就需要用到手动确认。
application.yml 配置 spring.rabbitmq.listener.simple.acknowledge-mode=MANUAL
spring:
application:
name: rabbit-test
rabbitmq:
host: 10.2.6.33
port: 5672
username: liuyanhui
password: liuyanhui
listener:
simple:
acknowledge-mode: MANUAL
在监听一端
@Component
public class QueueReceiver {
@RabbitListener(queues = "queue2")
@RabbitHandler
public void process(String strs, Message message, Channel channel) throws IOException {
System.out.println("接收到：" + strs);
//channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); //确认成功
channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true); //确认失败
}
}
channel.basicAck 确认成功
channel.basicNack 确认失败 ： 最后一个参数代表是否需要重发，需要根据实际执行情况做选择，否则可能陷入死循环。