学习rabbitmq

Posted 2020-12-03 gqymy

基本概念

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：自动删除

 

工作过程：

生产者客户端：

1. 客户端连接到RabbitMQ服务器上，打开一个消息通道（channel）；
2. 客户端声明一个消息交换机（exchange），并设置相关属性。
3. 客户端声明一个消息队列（queue），并设置相关属性。
4. 客户端使用routing key在消息交换机（exchange）和消息队列（queue）中建立好绑定关系。
5. 客户端投递消息都消息交换机（exchange）上
6. 客户端关闭消息通道（channel）以及和服务器的连接。

服务器端：

exchange接收到消息后，根据消息的key（这个key的产生规则暂时没研究，有知道的小伙伴可以留言告诉我）和以及设置的binding，进行消息路由，将消息投递到一个或多个消息队列中。

 

消息持久化：

RabbitMQ支持数据持久化，也就是把数据写在磁盘上，可以增加数据的安全性。消息队列持久化包括三个部分：

1. 消息交换机（exchange）持久化，在声明时指定durable为1
2. 消息队列（queue）持久化，在声明时指定durable为1
3. 消息持久化，在投递时指定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);

}

}