工作过程：
生产者客户端：
-
客户端连接到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);
}
}
