基本概念
channel：消息通道 ，在客户端的每个连接里，可建立多个channel，每个channel代表一个 会话任务。
exchage：消息路由规则，四种模式（topic、direct、fanout、header）
direct：默认，根据routingKey完全匹配，好处是先匹配再发送
topic：根据绑定关键字通配符规则匹配、比较灵活
fanout：不需要指定routingkey，相当于群发
header：不太常用，可以自定义匹配规则
queue：消息存储
binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来。
routerKey：消息路由关键字（发送的时候成为bindingkey,接收成为routingKey）
队列的概念（生产者消费者启动报错大多数都是这几个不匹配导致）
durable：持久化到硬盘
broker：消息队列服务器实体
vhost：虚拟主机，一个broker里可以开设多个vhost，用做不同用户的权限分离
exclusive：唯一性
autoDelete：自动删除
工作过程：
生产者客户端：
- 客户端连接到RabbitMQ服务器上，打开一个消息通道（channel）；
- 客户端声明一个消息交换机（exchange），并设置相关属性。
- 客户端声明一个消息队列（queue），并设置相关属性。
- 客户端使用routing key在消息交换机（exchange）和消息队列（queue）中建立好绑定关系。
- 客户端投递消息都消息交换机（exchange）上
- 客户端关闭消息通道（channel）以及和服务器的连接。
服务器端：
exchange接收到消息后，根据消息的key（这个key的产生规则暂时没研究，有知道的小伙伴可以留言告诉我）和以及设置的binding，进行消息路由，将消息投递到一个或多个消息队列中。
消息持久化：
RabbitMQ支持数据持久化，也就是把数据写在磁盘上，可以增加数据的安全性。消息队列持久化包括三个部分：
- 消息交换机（exchange）持久化，在声明时指定durable为1
- 消息队列（queue）持久化，在声明时指定durable为1
- 消息持久化，在投递时指定delivery_mode为2（1是非持久化）
如果消息交换机（exchange）和消息队列（queue）都是持久化的话，那么他们之间的绑定（Binding）也是持久化的。如果消息交换机和消息队列之间一个持久化、一个非持久化，那么就不允许绑定。
springboot如何集成
pom引入：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
@Configuration
public class RabbitConfig {
// 发送消息的格式转换器
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate template = new RabbitTemplate(connectionFactory);
template.setMessageConverter(new Jackson2JsonMessageConverter());
return template;
}
// 接收消息的格式转换器
@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
factory.setMessageConverter(new Jackson2JsonMessageConverter());
return factory;
}
// 信道配置，此地使用direct模式
@Bean
public DirectExchange defaultExchange() {
return new DirectExchange(MQConstant.EXCHANGE, true, false);
}
// 配置队列规则属性 例如保活时间 持久化 是否自动删除等
@Bean
public Queue queue() {
Map<String, Object> arguments = new HashMap<>();
arguments.put("x-message-ttl", 25000);//25秒自动删除
Queue queue = new Queue(MQConstant.QUEUE_NAME, true, false, true, arguments);
return queue;
}
// 绑定队列和exchange
@Bean
public Binding binding() {
return BindingBuilder.bind(queue()).to(defaultExchange()).with(MQConstant.QUEUE_NAME);
}
}
如何发送
@Autowired
private RabbitTemplate template;
//convertAndSend和send的区别是这个convert更方便使用，可以传object进去
template.convertAndSend(MQConstant.EXCHANGE, bindingKey, msg);
//如何接收，注意队列名称、exchange名称、routingKey的指定。
//注意：队列的消息只要被一个消费者匹配消费后就不存在了
@Component
@RabbitListener(containerFactory = "rabbitListenerContainerFactory", bindings = @QueueBinding(value = @Queue(value = "default_queue", durable = "true"), exchange = @Exchange(value = "default_exchange", type = ExchangeTypes.TOPIC), key = "meeting"))
@Log4j
public class RabbitMqListener {
@RabbitHandler
public void processMessage(MqMsg message) {
log.error( message);
}
}