springboot集成rabbitmq
安装步骤请百度，安装完rabbitMq后，输入http://ip:15672/ ，是可以看到一个简单后台管理界面的。
在这个界面里面我们可以做些什么？
可以手动创建虚拟host，创建用户，分配权限，创建交换机，创建队列等等，还有查看队列消息，消费效率，推送效率等等。
以上这些管理界面的操作在这篇暂时不做扩展描述，首先先介绍一个简单的一个消息推送到接收的流程，提供一个简单的图：
黄色的圈圈就是我们的消息推送服务，将消息推送到 中间方框里面也就是 rabbitMq的服务器，然后经过服务器里面的交换机、队列等各种关系（后面会详细讲）将数据处理入列后，最终右边的蓝色圈圈消费者获取对应监听的消息。
常用的交换机有以下三种:
Direct Exchange
直连型交换机，根据消息携带的路由键将消息投递给对应队列。
大致流程，有一个队列绑定到一个直连交换机上，同时赋予一个路由键 routing key 。然后当一个消息携带着路由值为X，这个消息通过生产者发送给交换机时，交换机就会根据这个路由值X去寻找绑定值也是X的队列。
Fanout Exchange
扇型交换机，这个交换机没有路由键概念，就算你绑了路由键也是无视的。 这个交换机在接收到消息后，会直接转发到绑定到它上面的所有队列。
Topic Exchange
主题交换机，这个交换机其实跟直连交换机流程差不多，但是它的特点就是在它的路由键和绑定键之间是有规则的。
简单地介绍下规则：
* (星号) 用来表示一个单词 (必须出现的)
# (井号) 用来表示任意数量（零个或多个）单词
主题交换机是非常强大的，为啥这么膨胀？
当一个队列的绑定键为 “#”（井号） 的时候，这个队列将会无视消息的路由键，接收所有的消息。
当 * (星号) 和 # (井号) 这两个特殊字符都未在绑定键中出现的时候，此时主题交换机就拥有的直连交换机的行为。
所以主题交换机也就实现了扇形交换机的功能，和直连交换机的功能。
代码实战：
本次实例教程需要创建2个springboot项目，一个 rabbitmq-provider （生产者），一个rabbitmq-consumer（消费者）。
首先创建 rabbitmq-provider，pom.xml里用到的jar依赖：
1
2
3
4
5
6
7
8
9
| <!--rabbitmq-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
|
然后application.yml：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| server:
port: 8021
spring:
#给项目来个名字
application:
name: rabbitmq-provider
#配置rabbitMq 服务器
rabbitmq:
host: 127.0.0.1
port: 5672
username: root
password: root
#虚拟host 可以不设置,使用server默认host
virtual-host: /xinke
|
ps：里面的虚拟host配置项不是必须的，我自己在rabbitmq服务上创建了自己的虚拟host，所以我配置了；你们不创建，就不用加这个配置项。
Direct Exchange 直连型交换机
rabbitmq-provider项目：创建DirectRabbitConfig.java：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
| import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @Author: zhangxiaozhou
* @Date: 2021/5/27
* @Time: 17:21
* @Description:
*/
@Configuration
public class DirectRabbitConfig {
//队列 起名：TestDirectQueue
@Bean
public Queue zxzDirectQueue() {
// durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
// exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
// autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
// return new Queue("TestDirectQueue",true,true,false);
//一般设置一下队列的持久化就好,其余两个就是默认false
return new Queue("zxzDirectQueue",true);
}
//Direct交换机 起名：zxzDirectExchange
@Bean
DirectExchange zxzDirectExchange() {
// return new DirectExchange("TestDirectExchange",true,true);
return new DirectExchange("zxzDirectExchange",true,false);
}
//绑定 将队列和交换机绑定, 并设置用于匹配键：zxzDirectRouting
@Bean
Binding bindingDirect() {
return BindingBuilder.bind(zxzDirectQueue()).to(zxzDirectExchange()).with("zxzDirectRouting");
}
@Bean
DirectExchange lonelyDirectExchange() {
return new DirectExchange("lonelyDirectExchange");
}
}
|
然后写个简单的接口进行消息推送（根据需求也可以改为定时任务等等，具体看需求），SendMessageController.java：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
| /**
* @Author: zhangxiaozhou
* @Date: 2021/5/27
* @Time: 17:32
* @Description:
*/
@RestController
public class SendMessageController {
@Autowired
RabbitTemplate rabbitTemplate; //使用RabbitTemplate,这提供了接收/发送等等方法
/**
* 直连型交换机，根据消息携带的路由键将消息投递给对应队列。
* 大致流程，有一个队列绑定到一个直连交换机上，同时赋予一个路由键 routing key 。
* 然后当一个消息携带着路由值为X，这个消息通过生产者发送给交换机时，交换机就会根据这个路由值X去寻找绑定值也是X的队列。
* @param messageData
* @return
*/
@GetMapping("/sendDirectMessage")
public String sendDirectMessage(@Param("messageData") String messageData) {
String messageId = String.valueOf(UUID.randomUUID());
String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
Map<String,Object> map=new HashMap<>();
map.put("messageId",messageId);
map.put("messageData",messageData);
map.put("createTime",createTime);
//将消息携带绑定键值：zxzDirectRouting 发送到交换机zxzDirectExchange
rabbitTemplate.convertAndSend("zxzDirectExchange", "zxzDirectRouting", map);
return "ok";
}
}
rabbitmq-consumer项目： 创建消费者FirstDirectReceiver.java：
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
/**
* @Author: zhangxiaozhou
* @Date: 2021/5/27
* @Time: 17:52
* @Description: 直连型交换机消费者
*/
@Component
@RabbitListener(queues = "zxzDirectQueue")//监听的队列名称 zxzDirectQueue
public class FirstDirectReceiver {
@RabbitHandler
public void process(Map testMessage) {
System.out.println("第一个DirectReceiver消费者收到消息 : " + testMessage.toString());
}
}
|
这样就完成了直连型交换机的生产者和消费者，可以将两个项目都启动，调用[localhost:xxxx/sendDirectMessage] 进行测试，测试结果如下：（如果多个消费者，就会轮询消费）
Topic Exchange 主题交换机
rabbitmq-provider项目： 创建TopicRabbitConfig.java：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
| 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.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @Author: zhangxiaozhou
* @Date: 2021/5/27
* @Time: 18:55
* @Description:
*/
@Configuration
public class TopicRabbitConfig {
//绑定键
public final static String first = "topic.first";
public final static String second = "topic.second";
@Bean
public Queue firstQueue() {
return new Queue(TopicRabbitConfig.first);
}
@Bean
public Queue secondQueue() {
return new Queue(TopicRabbitConfig.second);
}
@Bean
TopicExchange exchange() {
return new TopicExchange("topicExchange");
}
//将firstQueue和topicExchange绑定,而且绑定的键值为topic.man
//这样只要是消息携带的路由键是topic.man,才会分发到该队列
@Bean
Binding bindingExchangeMessage() {
return BindingBuilder.bind(firstQueue()).to(exchange()).with(first);
}
//将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#
// 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列
@Bean
Binding bindingExchangeMessage2() {
return BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");
}
}
|
多添加2个接口，用于推送消息到主题交换机：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
| /**
* 主题交换机，这个交换机其实跟直连交换机流程差不多，但是它的特点就是在它的路由键和绑定键之间是有规则的。
*
* 简单地介绍下规则：
* * (星号) 用来表示一个单词 (必须出现的)
* # (井号) 用来表示任意数量（零个或多个）单词
*
* @param messageData
* @return
*/
@GetMapping("/sendTopicMessage1")
public String sendTopicMessage1(@Param("messageData") String messageData) {
String messageId = String.valueOf(UUID.randomUUID());
String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
Map<String, Object> manMap = new HashMap<>();
manMap.put("messageId", messageId);
manMap.put("messageData", messageData);
manMap.put("createTime", createTime);
rabbitTemplate.convertAndSend("topicExchange", "topic.first", manMap);
return "ok";
}
@GetMapping("/sendTopicMessage2")
public String sendTopicMessage2(@Param("messageData") String messageData) {
String messageId = String.valueOf(UUID.randomUUID());
String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
Map<String, Object> womanMap = new HashMap<>();
womanMap.put("messageId", messageId);
womanMap.put("messageData", messageData);
womanMap.put("createTime", createTime);
rabbitTemplate.convertAndSend("topicExchange", "topic.second", womanMap);
return "ok";
}
rabbitmq-consumer项目：创建消费者1，FirstTopicReceiver.java：
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
/**
* @Author: zhangxiaozhou
* @Date: 2021/5/27
* @Time: 18:58
* @Description: 主题型交换机
*/
@Component
@RabbitListener(queues = "topic.first")
public class FirstTopicReceiver {
@RabbitHandler
public void process(Map testMessage) {
System.out.println("FirstTopicReceiver消费者收到消息 : " + testMessage.toString());
}
}
消费者2，SecondTopicReceiver.java：
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
/**
* @Author: zhangxiaozhou
* @Date: 2021/5/27
* @Time: 19:00
* @Description: 主题型交换机
*/
@Component
@RabbitListener(queues = "topic.second")
public class SecondTopicReceiver {
@RabbitHandler
public void process(Map testMessage) {
System.out.println("SecondTopicReceiver消费者收到消息 : " + testMessage.toString());
}
}
|
主题型交换机的消息配置和消费者已完成，下面调用sendTopicMessage1接口，进行测试，测试结果如下：
由于 firstQueue队列 绑定的路由键是 topic.first ,所以可以收到该消息；
由于 secondQueue队列 绑定的路由键是 topic.# ,所以只要消息携带的路由键是以topic.开头,都会分发到该队列
如果调用sendTopicMessage2接口，进行测试，则只会有SecondTopicReceiver消费者收到消息。
消息确认
application.yml文件：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| server:
port: 8021
spring:
#给项目来个名字
application:
name: rabbitmq-provider
#配置rabbitMq 服务器
rabbitmq:
host: 127.0.0.1
port: 5672
username: root
password: root
#虚拟host 可以不设置,使用server默认host
virtual-host: JCcccHost
#消息确认配置项
#确认消息已发送到交换机(Exchange)
publisher-confirms: true
#确认消息已发送到队列(Queue)
publisher-returns: true
|
然后是配置相关的消息确认回调函数，RabbitConfig.java：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
| import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @Author : JCccc
* @CreateTime : 2019/9/3
* @Description :
**/
@Configuration
public class RabbitConfig {
@Bean
public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){
RabbitTemplate rabbitTemplate = new RabbitTemplate();
rabbitTemplate.setConnectionFactory(connectionFactory);
//设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数
rabbitTemplate.setMandatory(true);
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
System.out.println("ConfirmCallback: "+"相关数据："+correlationData);
System.out.println("ConfirmCallback: "+"确认情况："+ack);
System.out.println("ConfirmCallback: "+"原因："+cause);
}
});
rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
System.out.println("ReturnCallback: "+"消息："+message);
System.out.println("ReturnCallback: "+"回应码："+replyCode);
System.out.println("ReturnCallback: "+"回应信息："+replyText);
System.out.println("ReturnCallback: "+"交换机："+exchange);
System.out.println("ReturnCallback: "+"路由键："+routingKey);
}
});
return rabbitTemplate;
}
}
|
可以看到上面写了两个回调函数，一个叫 ConfirmCallback ，一个叫 RetrunCallback；
那么以上这两种回调函数都是在什么情况会触发呢？
先从总体的情况分析，推送消息存在四种情况：
①消息推送到server，但是在server里找不到交换机
②消息推送到server，找到交换机了，但是没找到队列
③消息推送到sever，交换机和队列啥都没找到
④消息推送成功
四种情况的结论：
①这种情况触发的是 ConfirmCallback 回调函数
②这种情况触发的是 ConfirmCallback和RetrunCallback两个回调函数。
③这种情况触发的是 ConfirmCallback 回调函数。
④这种情况触发的是 ConfirmCallback 回调函数。
消费者接收到消息的消息确认机制
和生产者的消息确认机制不同，因为消息接收本来就是在监听消息，符合条件的消息就会消费下来。
所以，消息接收的确认机制主要存在三种模式：
①自动确认， 这也是默认的消息确认情况。 AcknowledgeMode.NONE
RabbitMQ成功将消息发出（即将消息成功写入TCP Socket）中立即认为本次投递已经被正确处理，不管消费者端是否成功处理本次投递。
所以这种情况如果消费端消费逻辑抛出异常，也就是消费端没有处理成功这条消息，那么就相当于丢失了消息。
一般这种情况我们都是使用try catch捕捉异常后，打印日志用于追踪数据，这样找出对应数据再做后续处理。
② 根据情况确认， 这个不做介绍
③ 手动确认 ， 这个比较关键，也是我们配置接收消息确认机制时，多数选择的模式。
消费者收到消息后，手动调用basic.ack/basic.nack/basic.reject后，RabbitMQ收到这些消息后，才认为本次投递成功。
basic.ack用于肯定确认
basic.nack用于否定确认（注意：这是AMQP 0-9-1的RabbitMQ扩展）
basic.reject用于否定确认，但与basic.nack相比有一个限制:一次只能拒绝单条消息
消费者端以上的3个方法都表示消息已经被正确投递，但是basic.ack表示消息已经被正确处理。
而basic.nack,basic.reject表示没有被正确处理：
着重讲下reject，因为有时候一些场景是需要重新入列的。
channel.basicReject(deliveryTag, true); 拒绝消费当前消息，如果第二参数传入true，就是将数据重新丢回队列里，那么下次还会消费这消息。设置false，就是告诉服务器，我已经知道这条消息数据了，因为一些原因拒绝它，而且服务器也把这个消息丢掉就行。 下次不想再消费这条消息了。
使用拒绝后重新入列这个确认模式要谨慎，因为一般都是出现异常的时候，catch异常再拒绝入列，选择是否重入列。
但是如果使用不当会导致一些每次都被你重入列的消息一直消费-入列-消费-入列这样循环，会导致消息积压。
顺便也简单讲讲 nack，这个也是相当于设置不消费某条消息。
channel.basicNack(deliveryTag, false, true);
第一个参数依然是当前消息到的数据的唯一id;
第二个参数是指是否针对多条消息；如果是true，也就是说一次性针对当前通道的消息的tagID小于当前这条消息的，都拒绝确认。
第三个参数是指是否重新入列，也就是指不确认的消息是否重新丢回到队列里面去。
同样使用不确认后重新入列这个确认模式要谨慎，因为这里也可能因为考虑不周出现消息一直被重新丢回去的情况，导致积压。
原文链接：https://blog.csdn.net/qq_35387940/article/details/100514134
