RabbitMQ

朱雀 2024-04-18 15:25 8阅读 0赞

**一、交换机类型**  
fanout/direct/topic/header  
**二、常用的方法详解**  
1、exchangeDeclare 交换机的声明  
exchange：交换机名称；  
type：交换机类型；  
durable：是否为持久型交换机，true为是/false为否；持久型交换机：rabbimq服务关闭后不会被自动删除；非持久型交换机：rabbitmq服务关闭后，就会被自动删除；  
autoDelete：是否自动删除，true为自动删除；如果为true，与该交换机绑定的所有队列都解绑后，交换机会自动删除；  
internal：是否为内置交换机，该交换机只能收到其他交换机发送的消息，提供者不能直接给内置交换机发送消息；  
argument：设置交换机的额外参数；  
2、queueDeclare 队列声明  
queue：队列的名称；  
durable：是否为吃持久化队列；  
exclusive：是否为排他队列，排他队列只对创建队列的连接可见；  
autoDelete：是否自动删除，一旦消费者断开和队列的绑定，就会自动删除该队列；  
arguments：设置队列的额外参数；  
3、queueBind：队列绑定  
queue：队列名称，该队列必须存在；  
exchange：交换机名称，必须存在；  
routingKey：路由键(当交换机为direct和topic的时侯有效)  
**三、RabbitMQ的消息过期时间(TTL)**  
1、通过队列给消息设置过期时间；  
2、通过消息本身独立设置过期时间；

1、rabbitmq只会删除队头的过期时间；
    2、如果通过队列设置消息的过期时间，过期的消息一定会出现在队头；
    3、如果给消息独立设置过期时间，则过期消息有可能出现在对中；

**四、死信队列**  
普通队列中，如果产生了死信消息，如果该队列绑定了一个死信交换机，则死信消息会自动发送给死信交换机，如果这个死信交换机还绑定了一个其他队列，则消息最终会发送到这个队列中，则这个队列就是所谓的死信队列；  
产生死信消息的时机：  
1、消息过期并且出现在队头；  
2、队列已满，继续添加新的元素(基础队头的元素)；  
3、消费者拒绝该消息，并且requeue设置为false；  
五、延迟队列  
延迟队列就是消费者会延后一定的时间收到提供者发送的消息，RabbitMQ没有提供延迟队列，但是开发者可以通过使用TTL+死信队列模拟出一个延迟队列；  
延迟队列的运用场景：  
1、订单下单后，支付时间超时后订单需要关闭  
2、延迟队列可以设置定时消息  
**六、消息的持久化**

1、队列的持久化不等于消息的持久化；
    2、如果队列不设置成持久化，则消息持久化毫无意义；
    3、持久化的消息最终会写入到硬盘中，当rabbitmq服务启动的时候，会从硬盘中读取消息恢复到队列；
    4、如果所有的消息都设置成持久化的消息，则会严重的降低RabbitMQ服务的吞吐量，所以在实际开发中，应该只给哪那些真正需要数据安全的消息进行持久化设置

**七、消息确认机制**  
发送方：  
1、事务机制  
2、publish confirm机制  
2.1同步确认机制  
2.2异步确认机制

注意：
    通过发送方消息确认机制+消费方消息确认机制，有可能导致一个问题：一个消息可能被多次消费；
    解决：将消费方的消费方法设置成幂等方法：基于数据库的唯一值限制；基于Redis判定；

**八、RabbitMQ的运用场景**  
消息的异步化 eg：异步发送邮件，异步发送短信验证码  
消息的延迟处理 eg：订单定时支付的场景  
消息的广播 eg：发送消息到Netty集群  
请求削峰 eg：秒杀的业务场景

## 实际操作中的运用 ##

**一、发送方的消息确认**  
**同步回调确认(publisher confirm)**  
1.设置channel为confirm模式：

channel.confirmSelect();

2.发送消息：

channel.basicPublish(exchange,routingKey,props,发送的内容);

3.同步等待rabbitMQ的返回结果：

boolean flag = channel.waitForConfirms();

**异步回调确认**  
1.设置channel的confirm模式

channel.confirmSelect();

2.设置confirm的异步回调方法

channel.addConfirmListener(new ConfirmListener(){
    	//deliveryTag - 表示失败消息的id号；
    	//multiple - 表示是否为批量操作；
    	@Override
    	public void handleAck(long deliveryTag, boolean multiple){
    		//消息发送的回调方法
    		if(!multiple){
    			//非批量，只有一个成功
    			treemap.remove(deliveryTag);
    			//TODO 重新发送失败的消息；
    		}else{
    			//批量操作，表示id为deliveryTag之前的消息全部成功；
    			treemap = (TreeMap<Long,Object>) treemap.tailMap(deliveryTag);
    			//TODO 重新批量发送失败的消息；
    		}
    	}
    });

3.发送消息

//获得下一个消息的id号
    long id  = channel.getNextPublishSeqNo();
    //发送消息
    channel.basicPublish(......);
    //消息失败与否，都存入缓存中
    treemap.put(id,发送的内容);

**二、消费方的消息确认**

//手动确认消息
    channel.basicConsume(queue, autoAck(false为关闭自动确认/true开启自动确认), new DefaultConsumer(channel){
    	@Override
    	public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, ... ...){
    		//TODO 根据接收的消息执行业务；
    		//执行完业务，手动确认消息
    		channel.basicAck(envelope.getDeliveryTag(), (multiple:)false);
    	}
    });

//手动拒绝消息
    channel.basicConsume(queue, autoAck(false为关闭自动确认/true开启自动确认), new DefaultConsumer(channel){
    	@Override
    	public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, ... ...){
    		//TODO 根据接收的消息执行业务；
    		//执行完业务，手动确认消息
    		//channel.basicAck(envelope.getDeliveryTag(), (multiple:)false);
    		
    		//手动拒绝消息
    		//requeue设置为false，拒绝的消息为死信消息；
    		//热queue设置为true，则拒绝的消息重新设置为队列，并且设置到队列头部，重新进行消费者的投递；
    		channel.basicAck(envelope.getDeliveryTag(), (multiple:)false, (requeue:)true);
    	}
    });

**三、消费方的消息限制**

//接收消息
    channel.basicQos(100);//限制为最多100条；
    channel.basicConsume(queue, autoAck(false为关闭自动确认/true开启自动确认), new DefaultConsumer(channel){
    	@Override
    	public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, ... ...){
    		//TODO 根据接收的消息执行业务；
    		//执行完业务，手动确认消息
    		//channel.basicAck(envelope.getDeliveryTag(), (multiple:)false);//每确认一条消息RabbitMQ就会推送一条消息到消费方内存
    		
    		//手动拒绝消息
    		//requeue设置为false，拒绝的消息为死信消息；
    		//热queue设置为true，则拒绝的消息重新设置为队列，并且设置到队列头部，重新进行消费者的投递；
    		//channel.basicAck(envelope.getDeliveryTag(), (multiple:)false, (requeue:)true);
    	}
    });

**注意：**  
发送方的确认机制和消费方的确认机制，都只能保证消息最少会被发送一次，有可能造成一个消息多次发送。所以为了避免消息的重复消费，消费方的接口应该设置成幂等模式。  
**幂等：**  
多次调用一个方法（接口），得到的结果永远一样。