rabbitmq实战 - 消息确认机制：Spring整合实战

曾经终败给现在 2022-12-17 14:50 50阅读 0赞

【服务端确认】

1.确认message成功发送到交换器：exchange

2.确认message成功从exchange发送到了队列：queue

\# 导入依赖

<dependency>
        <groupId>org.springframework.amqp</groupId>
        <artifactId>spring-rabbit</artifactId>
        <version>1.7.14.RELEASE</version>
        <exclusions>
            <exclusion>
                <groupId>com.rabbitmq</groupId>
                <artifactId>amqp-client</artifactId>
            </exclusion>
        </exclusions>
    </dependency>
    
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>4.12.0</version>
    </dependency>

\# 消息提供者配置

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
           xmlns:rabbit="http://www.springframework.org/schema/rabbit"
           xmlns:p="http://www.springframework.org/schema/p"
           xmlns:context="http://www.springframework.org/schema/context"
           xmlns:aop="http://www.springframework.org/schema/aop"
           xmlns:tx="http://www.springframework.org/schema/tx"
           xsi:schemaLocation="http://www.springframework.org/schema/rabbit
                  https://www.springframework.org/schema/rabbit/spring-rabbit.xsd
                  http://www.springframework.org/schema/beans
                 https://www.springframework.org/schema/beans/spring-beans-4.3.xsd
                 http://www.springframework.org/schema/context
                 https://www.springframework.org/schema/context/spring-context-4.3.xsd
                  http://www.springframework.org/schema/aop
                  https://www.springframework.org/schema/aop/spring-aop-4.3.xsd
                  http://www.springframework.org/schema/tx
                  https://www.springframework.org/schema/tx/spring-tx-4.3.xsd">
    
        <context:component-scan base-package="com.learn" />
    
        <bean id="msgConverter" class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" />
    
    
        
        <rabbit:connection-factory id="connectionFactory"
                                   host="192.168.199.237"
                                   port="5672"
                                   username="app_client"
                                   password="client123"
                                   virtual-host="/app"
                                   publisher-confirms="true"
                                   publisher-returns="true" />
    
        
        <bean id="confirmCallback" class="com.learn.config.MsgSendConfirmCallback"></bean>
    
        
        <rabbit:template id="template"
                         message-converter="msgConverter"
                         connection-factory="connectionFactory"
                         exchange="my_direct_exchange" reply-timeout="2000"
                         routing-key="my.binding"
                         confirm-callback="confirmCallback"
                         return-callback="confirmCallback" 
                         mandatory="true" />
    
        <rabbit:admin connection-factory="connectionFactory" auto-startup="true"/>
    
        
        <rabbit:queue id="my_queue"
                      name="my_queue"
                      auto-declare="true"
                      durable="true"
                      auto-delete="false"
                      exclusive="false"/>
    
        
        <rabbit:direct-exchange id="my_direct_exchange"
                                name="my_direct_exchange"
                                auto-declare="true"
                                durable="true"
                                auto-delete="true">
            <rabbit:bindings>
                <rabbit:binding queue="my_queue" key="my.binding" />
            </rabbit:bindings>
        </rabbit:direct-exchange>
    
    
    </beans>

\# 消息确认回调

package com.learn.config;
    
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.rabbit.support.CorrelationData;
    import org.springframework.beans.factory.annotation.Autowired;
    
    import javax.annotation.PostConstruct;
    
    public class MsgSendConfirmCallback implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @PostConstruct
        public void init() {
            rabbitTemplate.setConfirmCallback(this);
            rabbitTemplate.setReturnCallback(this);
            rabbitTemplate.setMandatory(true);
        }
    
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            if(ack) {
                System.out.println("消息发送成功: " + correlationData);
            } else {
                System.out.println("消息发送失败，id= " + correlationData + " 原因：" + cause);
            }
    
        }
    
        @Override
        public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
            System.out.println("returned message: " + new String(message.getBody()));
            System.out.println("应答码: " + replyCode);
            System.out.println("描述：" + replyText);
            System.out.println("消息使用的交换器 exchange : " + exchange);
            System.out.println("消息使用的路由键 routing : " + routingKey);
        }
        
    
    }

【消费者端确认】

\# 消息消费者配置

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
           xmlns:rabbit="http://www.springframework.org/schema/rabbit"
           xmlns:p="http://www.springframework.org/schema/p"
           xmlns:context="http://www.springframework.org/schema/context"
           xmlns:aop="http://www.springframework.org/schema/aop"
           xmlns:tx="http://www.springframework.org/schema/tx"
           xsi:schemaLocation="http://www.springframework.org/schema/rabbit
                  https://www.springframework.org/schema/rabbit/spring-rabbit.xsd
                  http://www.springframework.org/schema/beans
                 https://www.springframework.org/schema/beans/spring-beans-4.3.xsd
                 http://www.springframework.org/schema/context
                 https://www.springframework.org/schema/context/spring-context-4.3.xsd
                  http://www.springframework.org/schema/aop
                  https://www.springframework.org/schema/aop/spring-aop-4.3.xsd
                  http://www.springframework.org/schema/tx
                  https://www.springframework.org/schema/tx/spring-tx-4.3.xsd">
    
        <context:component-scan base-package="com.learn" />
    
        
        <rabbit:connection-factory id="connectionFactory"
                                   host="192.168.199.237"
                                   port="5672"
                                   username="app_client"
                                   password="client123"
                                   virtual-host="/app"
                                   connection-timeout="3000"
                                   channel-cache-size="5"/>
    
        
        <rabbit:admin connection-factory="connectionFactory" />
    
        
        <rabbit:queue name="my_queue" auto-declare="true" auto-delete="false" exclusive="false" durable="true"/>
    
        
        <rabbit:direct-exchange name="my_direct_exchange" auto-declare="true" durable="true" auto-delete="false">
            <rabbit:bindings>
                <rabbit:binding queue="my_queue" key="my.queue.key" />
            </rabbit:bindings>
        </rabbit:direct-exchange>
    
        
        <bean id="confirmListener" class="com.learn.listeners.ConsumeConfirmListener" />
    
        
        <rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual">
            
            <rabbit:listener ref="confirmListener" method="onMessage" queue-names="my_queue"/>
        </rabbit:listener-container>
    
    
    </beans>

\# 监听器 - 开启消费端手动确认

package com.learn.listeners;
    
    import com.rabbitmq.client.Channel;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
    import org.springframework.stereotype.Component;
    
    import java.io.IOException;
    
    /**
     * 消费端 - 手动确认
     */
    public class ConsumeConfirmListener implements ChannelAwareMessageListener {
    
        @Override
        public void onMessage(Message message, Channel channel) throws Exception {
            try {
                System.out.println("consumer received msg: " + new String(message.getBody()) );
                // 手动确认：ack
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            } catch (IOException e) {
                e.printStackTrace();
                channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);
                channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
            }
            
        }
        
    }