rabbitmq-spring使用小计

rabbitmq是为数不多的spring官方开放集成插件的消息队列插件，rabbitmq-spring网上教程很多，但是之前我一直没找到想关于ack的代码实现，之前一个项目中有使用就贴出来分享一下，那时候比较懒没有去官网找到相关的api文档，有资料的朋友发出来分享一下。

pom
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit</artifactId>
            <version>1.2.2.RELEASE</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"  
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd  
    http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd">  
    
    <rabbit:connection-factory  virtual-host="***" id="connectionFactory" host="rabbitmq.com" username="***"
        password="***" port="***"/>
          
    <rabbit:admin connection-factory="connectionFactory"/>

    <!--add mq for cooperation record-->


    <!-- queue 队列声明-->
    <rabbit:queue id="addque" durable="true" auto-delete="false" exclusive="false" name="addque"/>


    <!-- exchange queue binging key 绑定 -->
    <rabbit:direct-exchange name="queExchange" durable="true" auto-delete="false" id="queExchange">
        <rabbit:bindings>
            <rabbit:binding queue="addque" key="addque"/>
        </rabbit:bindings>
    </rabbit:direct-exchange>


    <!-- spring template声明-->
    <rabbit:template exchange="queExchange" id="amqpTemplate"  connection-factory="connectionFactory" />

    <bean id="queListener" class="com.rabbitmq.listener.QueListener"/>

    <rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual" prefetch="1">
        <rabbit:listener queues="addque" ref="queListener"  method="onMessage"/>
    </rabbit:listener-container>
    
</beans>

package com.rabbitmq.listener;

import java.io.IOException;

import org.apache.log4j.Logger;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.bean.Que;
import com.bean.dto.QueDto;
import com.enums.RecordErrEnum;
import com.exception.QueException;
import com.service.QueService;
import com.util.queConversionUtil;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;

@Service("queListener")
public class QueListener implements ChannelAwareMessageListener {

    private static final Logger logger = Logger.getLogger(queListener.class);
    private static ObjectMapper MAPPER = new ObjectMapper();
    @Autowired
    private QueService queService;

    @Override
    public void onMessage(Message message, Channel channel) throws IOException {
        logger.info("get mq message for que " + new String(message.getBody()));
        try {
            QueDto queDto = MAPPER.readValue(message.getBody(), QueDto.class);
            //业务逻辑
        }
        catch (QueException ce) {
            //特定exception重发
            if (ce.getErrCode() == RecordErrEnum.HTTP_SEND_ERR.getErrCode()) {
                logger.info("由于访问passport获取snsId接口失败,3秒后重新记录 " + new String(message.getBody()));
                try {
                    Thread.sleep(3000);
                    //ack表示失败，要求重新发送
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                    return;
                }
                catch (Exception e) {
                    logger.error("重发mq消息" + new String(message.getBody()) + " 失败", e);
                }
            }
        }
        catch (Exception e) {
            logger.error("save que " + new String(message.getBody()) + " error", e);
        }
        try {
            //最终应答rabbitmq返回，防止消息堵塞
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
        catch (IOException e) {
            logger.error("deal Record message is " + new String(message.getBody()) + "error ", e);
        }
    }
}

rabbitmq的ack机制很简单，继承接口ChannelAwareMessageListener ，重写onMessage ,Channel channel为一个控制ack的变量， channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true)来告知mq重新发送。

发送消息的方式也很简单

    @Autowired
    private AmqpTemplate amqpTemplate;

注入AmqpTemplate，然后根据路由发送特定对象。

amqpTemplate.convertAndSend("addque", MAPPER.writeValueAsString(xxxx));