消息可靠性投递

confirm 确认模式

return 返回模式

rabbitmq整个消息投递的路径为

producer--->rabbitmq broker--->exchange--->queue--->consumer

消息从 producer 到 exchange则会返一个 confirmCallback·

消息从 exchange 到 queue投递失败则会返回个 returnCallback。

➢    设置ConnectionFactory的publisher-confirms="true" 开启确认模式。
➢    使用rabbitTemplate.setConfirmCallback设置回调函数。当消息发送到exchange后回调confirm方法。在方法中判断ack，如果为true，则发送成功，如果为false，则发送失败，需要处理。
➢    设置ConnectionFactory的publisher-returns="true" 开启退回模式。
➢    使用rabbitTemplate.setReturnCallback设置退回函数，当消息从exchange路由到queue失败后，如果设置了rabbitTemplate.setMandatory(true)参数，则会将消息退回给producer。并执行回调函数returnedMessage。
➢    在RabbitMQ中也提供了事务机制，但是性能较差

Consumer Ack

ack指Acknowledge，确认。 表示消费端收到消息后的确认方式。

有三种确认方式：

•    自动确认：acknowledge="none"
•    手动确认：acknowledge="manual"
•    根据异常情况确认：acknowledge="auto"

其中自动确认是指，当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从 RabbitMQ 的消息缓存中移除。但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么该消息就会丢失。如果设置了手动确认方式，则需要在业务处理成功后，调用channel.basicAck()，手动签收，如果出现异常，则调用channel.basicNack()方法，让其自动重新发送消息。

        如果消费端没有出现异常,则调用channel.basicAck(deliveryTag,false);确认签收消息

        如果出现异常,则在catch中调用 basicNack或 basicReject，拒绝消息，让MQ重新发送

消息可靠性总结

1、持久化

        •    exchange要持久化
        •    queue要持久化
        •    message要持久化

2、生产方确认Confirm
3、消费方确认Ack
4.、Broker高可用

消费端限流

在 中配置 prefetch属性设置消费端一次拉取多少消息

消费端的确认模式一定为手动确认。acknowledge="manual"

 

TTL（Time To Live）

1、设置队列过期时间使用参数:x- message-ttl,单位毫秒,会对整个队列消息统一过期

2、设置消息过期时使用参数: expiration。单位毫秒，当该消息在队列头部时(消费时),才会单独判断这一消息是否过期

3、如果两者都进行了设置,以时间短的为准。

死信队列

死信交换机和死信队列和普通的没有区别

需要设置给队列参数:

        x-dead-letter- exchange

        x-dead-letter- routing-key

当消息成为死信后,如果该队列绑定了死信交换机,则消息会被死信交换机重新路由到死信队

消息成为死信的三种情况

        1、队列消息长度到达限制

        2、消费者拒接消费消息,并且不重回队列

        3、原队列存在消息过期设置,消息到达超时时间末被消费

延迟队列

RabbitMQ中没用提供延迟队列功能

但是可以使用  TTL+死信队列  来实现该功能

应用场景：

        下单后,30分钟未支付,取消订单,回滚库存

        新用户注册成功7天后,发送短信问候

消息可靠性保障--消息补偿

  消息幂等性保障--乐观锁机制

幂等性指一次和多次请求某一个资源，对于资源本身应该具有同样的结果。

也就是说，任意多次执行对资源本身所产生的影响均与一次执行的影响相同。

 

spring配置rabbitMQ

rabbitmq.properties

rabbitmq.host=192.168.xxx.xxx
rabbitmq.port=5672
rabbitmq.username=guest
rabbitmq.password=guest
rabbitmq.virtual-host=/

produce.xml

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/context
       https://www.springframework.org/schema/context/spring-context.xsd
       http://www.springframework.org/schema/rabbit
       http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
 
    
    <context:property-placeholder location="classpath:rabbitmq.properties"/>
 
    
    <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
                               port="${rabbitmq.port}"
                               username="${rabbitmq.username}"
                               password="${rabbitmq.password}"
                               virtual-host="${rabbitmq.virtual-host}"
                               publisher-confirms="true"
                               publisher-returns="true"
    />
    
    <rabbit:admin connection-factory="connectionFactory"/>
 
    
    <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>
 
 
 
    
    <rabbit:queue id="test_queue_confirm" name="test_queue_confirm">rabbit:queue>
    <rabbit:direct-exchange name="test_exchange_confirm">
        <rabbit:bindings>
            <rabbit:binding queue="test_queue_confirm" key="confirm">rabbit:binding>
        rabbit:bindings>
    rabbit:direct-exchange>
 
 
    
    <rabbit:queue name="test_queue_ttl" id="test_queue_ttl">
        
        <rabbit:queue-arguments>
            
            <entry key="x-message-ttl" value="100000" value-type="java.lang.Integer"/>
        rabbit:queue-arguments>
    rabbit:queue>
 
    <rabbit:topic-exchange name="test_exchange_ttl" >
        <rabbit:bindings>
            <rabbit:binding pattern="ttl.#" queue="test_queue_ttl">rabbit:binding>
        rabbit:bindings>
    rabbit:topic-exchange>
 
 
    
 
    
 
    <rabbit:queue name="test_queue_dlx" id="test_queue_dlx">
        
        <rabbit:queue-arguments>
            
            <entry key="x-dead-letter-exchange" value="exchange_dlx" />
 
            
            <entry key="x-dead-letter-routing-key" value="dlx.hehe" />
 
            
            <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer" />
            
            <entry key="x-max-length" value="10" value-type="java.lang.Integer" />
        rabbit:queue-arguments>
    rabbit:queue>
    <rabbit:topic-exchange name="test_exchange_dlx">
        <rabbit:bindings>
            <rabbit:binding pattern="test.dlx.#" queue="test_queue_dlx">rabbit:binding>
        rabbit:bindings>
    rabbit:topic-exchange>
 
 
    
 
    <rabbit:queue name="queue_dlx" id="queue_dlx">rabbit:queue>
    <rabbit:topic-exchange name="exchange_dlx">
        <rabbit:bindings>
            <rabbit:binding pattern="dlx.#" queue="queue_dlx">rabbit:binding>
        rabbit:bindings>
    rabbit:topic-exchange>
 
 
 
    
    
    <rabbit:queue id="order_queue" name="order_queue">
        
        <rabbit:queue-arguments>
            <entry key="x-dead-letter-exchange" value="order_exchange_dlx" />
            <entry key="x-dead-letter-routing-key" value="dlx.order.cancel" />
            <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer" />
 
        rabbit:queue-arguments>
 
    rabbit:queue>
    <rabbit:topic-exchange name="order_exchange">
        <rabbit:bindings>
            <rabbit:binding pattern="order.#" queue="order_queue">rabbit:binding>
        rabbit:bindings>
    rabbit:topic-exchange>
 
    
    <rabbit:queue id="order_queue_dlx" name="order_queue_dlx">rabbit:queue>
    <rabbit:topic-exchange name="order_exchange_dlx">
        <rabbit:bindings>
            <rabbit:binding pattern="dlx.order.#" queue="order_queue_dlx">rabbit:binding>
        rabbit:bindings>
    rabbit:topic-exchange>
 
beans>

consumer.xml

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/context
       https://www.springframework.org/schema/context/spring-context.xsd
       http://www.springframework.org/schema/rabbit
       http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
    
    <context:property-placeholder location="classpath:rabbitmq.properties"/>
 
    
    <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
                               port="${rabbitmq.port}"
                               username="${rabbitmq.username}"
                               password="${rabbitmq.password}"
                               virtual-host="${rabbitmq.virtual-host}"/>
 
 
    <context:component-scan base-package="com.itheima.listener" />
 
    
    <rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual" prefetch="1" >
       
       
        
        
 
        
        <rabbit:listener ref="orderListener" queue-names="order_queue_dlx">rabbit:listener>
    rabbit:listener-container>
 
beans>