JMS RabbitMQ消息代理

分手后的思念是犯贱 2022-06-13 05:56 123阅读 0赞

queue(队列)：存储消息，消费者从队列中消费消息，durable表示队列是否持久化，持久化队列在rabbitmq服务重启后还在；

exchange(交换)：生产者将消息生产给交换机，再由交换机路由到绑定队列，AMQP规范将“默认Exchange（direct）”定义为没有名称。并且所有队列都使用其名称作为绑定值自动绑定到该默认Exchange(即直接Exchange),所以也可以设置routingKey为队列名称发送消息到指定队列。交换机也可以设置持久化，交换机有以下几种类型：

*  fanout：所有发送到该Exchange的消息都会路由到所有与它绑定的Queue中；
 *  direct：把消息路由到那些binding key与routing key完全匹配的Queue中；
 *  topic：将消息路由到binding key与routing key相匹配的Queue中，但这里为模糊匹配；
 *  headers：headers类型的Exchange不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。

binding(绑定)：连接exchange和queue；

bindingKey：绑定exchange和queue的句点号“. ”分隔的字符串；

routingKey：发送消息时指定的路由关键字，也是句点号“. ”分隔的字符串，通过它来匹配bindingKey从而发送到指定队列；

autoAck：设置队列是否自动应答，默认为true，消费者应答表示该消费者消费消息并处理完成，rabbitmq会删除该消息，如果一个消费者挂掉而没有应答，rabbitmq会将任务交付给另一个消费者去处理。如果设置为false，而消费者处理完消息后没有手动应答，rabbitmq认为消息处理失败，而不会删除消息，可通过basicAck进行手动应答，

channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false)

Delivery Tag： 用来标识信道中投递的消息。RabbitMQ 推送消息给 Consumer 时，会附带一个 Delivery Tag，multiple 取值为 false 时，表示通知 RabbitMQ 当前消息被确认；如果为 true，则额外将比第一个参数指定的 delivery tag 小的消息一并确认。

如果消费者处理消息中抛出异常并且没有捕获，程序没有主动应答或自动应答，rabbitmq认为该消息没有被正确处理，而不会删出该消息，因此消费者会一直消费该条消息，并持续抛出异常，

SimpleMessageListenerContainer： 最简单的消息监听器容器，只能处理固定数量的JMS会话，且不支持事务；

DefaultMessageListenerContainer：这个消息监听器容器建立在SimpleMessageListenerContainer容器之上，添加了对事务的支持；

ListenerContainer：这是功能最强大的消息监听器，与DefaultMessageListenerContainer相同，它支持事务，但是它还允许动态地管理JMS会话。

rabbitmq两种监听器消费消息的方法：

1、@RabbitListener注解的方式

@Component
    public class OutBoundListener {
      
      @RabbitListener(queues = "MSG_OUTBOUND_APP8075", containerFactory = "rlcFactory")
      public void process(Message message) {
         log.info("接收消息：{}", message.toString());
      }
    }

rlcFactory是在DefaultRabbitConfig中配置的SimpleRabbitListenerContainerFactory

package com.spring.component.rabbitmq.config;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    import org.springframework.amqp.core.AcknowledgeMode;
    import org.springframework.amqp.core.TopicExchange;
    import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import com.spring.component.rabbitmq.Properties;
    
    /**
     * 默认rabbitmq配置类
     * 
     * @author liushihua
     * @since 2018年04月11日
     */
    @Configuration
    public class DefaultRabbitConfig extends RabbitConfig {
      @Autowired
      private Properties rabbitProperties;
    
      @Bean("connectionFactory")
      public ConnectionFactory connectionFactory() {
        if (rabbitProperties.validQueue()) {
          return this.instanceConnectionFactory(rabbitProperties.getHost(),
              Integer.parseInt(rabbitProperties.getPort()), rabbitProperties.getUsername(),
              rabbitProperties.getPassword(), rabbitProperties.getVirtualHost());
        } else {
          log.info("没有配置默认mq参数，将不能使用默认mq");
          return new CachingConnectionFactory();
        }
      }
    
      @Bean("rabbitTemplate")
      public RabbitTemplate rabbitTemplate(
          @Qualifier("connectionFactory") ConnectionFactory connectionFactory) {
        return this.instanceRabbitTemplate(connectionFactory);
      }
    
      @Bean("rlcFactory")
      public SimpleRabbitListenerContainerFactory rlcFactory(
          @Qualifier("connectionFactory") ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = this.instanceFactory(connectionFactory);
        return factory;
      }
    
      @Bean("topicExchange")
      public TopicExchange topicExchange(
          @Qualifier("connectionFactory") CachingConnectionFactory connectionFactory) {
        if ("true".equals(rabbitProperties.getExchange())) {
          try {
            this.topicExchangeDeclare(connectionFactory);
            log.info("exchangeDeclare成功：TOPIC.EXCHANGE");
          } catch (IOException e) {
            log.error("exchangeDeclare失败：" + e.getMessage());
          } catch (TimeoutException e) {
            log.error("exchangeDeclare失败, 访问超时：" + e.getMessage());
          }
        }
        return new TopicExchange(Properties.TOPIC_EXCHANGE);
      }
    }

package com.spring.component.rabbitmq.config;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.core.AcknowledgeMode;
    import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.spring.component.utils.CollectionUtil;
    
    /**
     * 自定义rabbitmq配置
     * 
     * @author liushihua
     * @since 2018年04月10日
     */
    public class RabbitConfig {
    
      protected final Logger log = LoggerFactory.getLogger(this.getClass());
    
      public CachingConnectionFactory instanceConnectionFactory(String host, int port, String username,
          String password, String virtualHost) {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setHost(host);
        factory.setPort(port);
        factory.setUsername(username);
        factory.setPassword(password);
        if (CollectionUtil.isNotBlank(virtualHost)) {
          factory.setVirtualHost(virtualHost);
        }
        return factory;
      }
    
      public RabbitTemplate instanceRabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        // 为RabbitTemplate指定发送消息是的转换器
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        return rabbitTemplate;
      }
    
      /**
       * 配置RabbitListenerContainer的工厂类，可配置在@RabbitListener注解中的containerFactory属性
       * 
       * @param connectionFactory
       * @return
       */
      public SimpleRabbitListenerContainerFactory instanceFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        // 指定了我们接受消息的时候，以 JSON 传输的消息可以转换成对应的类型传入到方法中
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        // 设置为手动应答
        // factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        // 为每个listener配置并发的消费者个数的。
        // factory.setConcurrentConsumers(5);
        // 最大个数
        // factory.setMaxConcurrentConsumers(10);
        return factory;
      }
    
      /**
       * 配置rabbitTemplate的交换机，该模板默认发送消息到TOPIC.EXCHANGE交换机
       * 
       * @param connectionFactory
       * @throws IOException
       * @throws TimeoutException
       */
      public void topicExchangeDeclare(CachingConnectionFactory connectionFactory)
          throws IOException, TimeoutException {
        Connection connection = connectionFactory.getRabbitConnectionFactory().newConnection();
        Channel channel = connection.createChannel();
        // 交换机名称，交换机类型，是否持久化
        channel.exchangeDeclare("TOPIC.EXCHANGE", "topic", true);
      }
    }

package com.spring.component.rabbitmq;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    import java.util.stream.Stream;
    
    import javax.annotation.Resource;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.core.TopicExchange;
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.Bean;
    import org.springframework.stereotype.Component;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.DefaultConsumer;
    import com.spring.component.utils.CollectionUtil;
    import com.spring.component.utils.StringUtil;
    
    /**
     * 绑定队列和交换机
     * @author liushihua
     * @since 2018年04月11日
     */
    @Component
    public class QueueBuilder {
      private final Logger log = LoggerFactory.getLogger(this.getClass());
    
      @Resource
      private CachingConnectionFactory connectionFactory;
      @Resource(name = "iotbusConnectionFactory")
      private CachingConnectionFactory iotBusConnectionFactory;
      @Resource(name = "logserviceConnectionFactory")
      private CachingConnectionFactory logserviceConnectionFactory;
      @Autowired
      private Properties rabbitProperties;
      @Autowired
      private TopicExchange topicExchange;
      @Autowired
      private TopicExchange iotbusTopicExchange;
    
      @Bean
      public QueueBuilder intiQueue() {
        log.info("系统加载定义queue......");
        // 初始化默认mq队列
        if (rabbitProperties.validQueue()) {
          // 声明队列
          queueDeclare(connectionFactory, rabbitProperties.getQueues());
          // 声明订阅topic的队列
          queueDeclare(connectionFactory, rabbitProperties.getTopics());
        }
        // 初始化iotbus mq队列
        if (rabbitProperties.validIotbusQueue()) {
          queueDeclare(iotBusConnectionFactory, rabbitProperties.getIotbusQueues());
          queueDeclare(iotBusConnectionFactory, rabbitProperties.getIotbusTopics());
        }
        // 初始化logservice mq队列
        if (rabbitProperties.validLogQueue()) {
          queueDeclare(logserviceConnectionFactory, rabbitProperties.getLogQueues());
        }
        // 绑定默认mq topic routingKey到指定交换机上
        bindExchange(connectionFactory, topicExchange.getName(), rabbitProperties.getRouting());
        // 绑定iotbus mq topic routingKey
        bindExchange(iotBusConnectionFactory, iotbusTopicExchange.getName(),
            rabbitProperties.getIotbusRouting());
        return new QueueBuilder();
      }
    
      /**
       * 声明队列,根据队列是否配置auotAck自动应答创建队列
       * 
       * @param connectionFactory
       * @param queues
       */
      private void queueDeclare(CachingConnectionFactory connectionFactory, String queues) {
        Channel channel = channel(connectionFactory);
        if (null == channel) {
          log.error("channel为空，queueDeclare失败");
          return;
        }
        if (StringUtil.isNotBlank(queues)) {
          Stream.of(queues.trim().split(",")).filter(StringUtil::isNotBlank).forEach(queue -> {
            String queueName = queue;
            // 是否自动ack，如果不自动ack，需要使用channel.ack、channel.nack、channel.basicReject 进行消息应答
            String autoAck = null;
            int index = queue.indexOf(":");
            if (index > -1) {
              queueName = queue.trim().substring(0, index);
              autoAck = queue.trim().substring(index + 1);
            }
            try {
              // 创建队列,durable表示队列是否持久化，持久化队列在rabbitmq服务重启后还在
              channel.queueDeclare(queueName, true, false, false, null);
              if (false && StringUtil.equals("false", autoAck)) {
                // 设置队列是否自动ack，默认自动应答，
                channel.basicConsume(queueName, false, new DefaultConsumer(channel));
              }
              StringBuilder sb = new StringBuilder("queueDeclare成功:").append(queueName);
              if (StringUtil.isNotBlank(autoAck)) {
                sb.append(", autoAck = ").append(autoAck);
              }
              log.info(sb.toString());
            } catch (IOException e) {
              log.error("定义queue失败[{}],异常：{}", queueName, e.getMessage());
            }
          });
        }
      }
    
      /**
       * 为队列绑定交换机topic
       * 
       * @param connectionFactory
       * @param exchange
       * @param rotuing
       */
      private void bindExchange(CachingConnectionFactory connectionFactory, String exchange,
          String rotuing) {
        Channel channel = channel(connectionFactory);
        if (null == channel) {
          log.error("channel为空，bindExchange失败");
          return;
        }
        if (StringUtil.isAllNotBlank(exchange, rotuing)) {
          Stream.of(rotuing.trim().split("&")).filter(StringUtil::isNotBlank).forEach(rotuingName -> {
            String[] ss = rotuingName.trim().split(":");
            if (null != ss && ss.length > 1) {
              String routingKey = ss[0].trim();
              String routingValues = ss[1].trim();
              if (StringUtil.isNotBlank(routingKey) && CollectionUtil.isNotBlank(routingValues)) {
                Stream.of(routingValues.trim().split(",")).filter(StringUtil::isNotBlank)
                    .forEach(queue -> {
                      try {
                        // 为队列绑定交换机topic和bindKey
                        channel.queueBind(queue.trim(), exchange, routingKey);
                        log.info(
                            "#####################      绑定topic成功: topicName={} & queue= {} & routingKey= {}",
                            exchange, queue.trim(), routingKey);
                      } catch (IOException e) {
                        log.error("bindExchange失败routing[{}],异常: {}", rotuing, e.getMessage());
                      }
                    });
              } else {
                log.error("routing 格式不正确，rotuing={}", rotuingName);
              }
            } else {
              log.error("routing 格式不正确，rotuing={}", rotuingName);
            }
          });
        }
      }
    
      /**
       * 获得channel,并设置同一时间每次发给一个消息给一个worker
       * 
       * @param connectionFactory
       * @return
       */
      private Channel channel(CachingConnectionFactory connectionFactory) {
        Channel channel = null;
        try {
          Connection connection = connectionFactory.getRabbitConnectionFactory().newConnection();
          channel = connection.createChannel();
          // 设置同一时间每次发给一个消息给一个worker
          channel.basicQos(1, false);
        } catch (IOException e) {
          log.error("createChannel失败，异常：{}", e.getMessage());
        } catch (TimeoutException e) {
          log.error("createChannel超时，异常：{}", e.getMessage());
        }
        return channel;
      }
    }

package com.spring.component.rabbitmq;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.Bean;
    import org.springframework.core.env.Environment;
    import org.springframework.stereotype.Component;
    
    import com.spring.component.utils.StringUtil;
    
    /**
     * 多rabbitmq环境配置
     * 
     * @author liushihua
     * @since 2018年04月11日
     */
    @Component
    public class Properties {
      // 默认rabbitmq配置
      private String host;
      private String port;
      private String username;
      private String password;
      private String virtualHost;
      private String queues;
      private String topics;
      private String routing;
      private String exchange;
      private String iotbusHost;
      // 自定义rabbitmq配置
      private String iotbusPort;
      private String iotbusUsername;
      private String iotbusPassword;
      private String iotbusVirtualHost;
      private String iotbusQueues;
      private String iotbusTopics;
      private String iotbusRouting;
      private String iotbusExchange;
      // 日志rabbitmq配置
      private String logHost;
      private String logPort;
      private String logUsername;
      private String logPassword;
      private String logVirtualHost;
      private String logQueues;
      private String logEnable;
    
      @Autowired
      private Environment env;
      public static final String TOPIC_EXCHANGE = "TOPIC.EXCHANGE";
    
      @Bean(name = "rabbitProperties")
      public Properties properties() {
        Properties properties = new Properties();
        this.host = this.env.getProperty("spring.rabbitmq.host", "");
        this.port = this.env.getProperty("spring.rabbitmq.port", "");
        this.username = this.env.getProperty("spring.rabbitmq.username", "");
        this.password = this.env.getProperty("spring.rabbitmq.password", "");
        this.virtualHost = this.env.getProperty("spring.rabbitmq.virtualHost", "");
        this.queues = this.env.getProperty("spring.rabbitmq.queues", "");
        this.topics = this.env.getProperty("spring.rabbitmq.topics", "");
        this.routing = this.env.getProperty("spring.rabbitmq.topic.routing", "");
        this.exchange = this.env.getProperty("spring.rabbitmq.topic.exchange", "true");
        this.iotbusHost = this.env.getProperty("iotbus.rabbitmq.host", "");
        this.iotbusPort = this.env.getProperty("iotbus.rabbitmq.port", "");
        this.iotbusUsername = this.env.getProperty("iotbus.rabbitmq.username", "");
        this.iotbusPassword = this.env.getProperty("iotbus.rabbitmq.password", "");
        this.iotbusVirtualHost = this.env.getProperty("iotbus.rabbitmq.virtualHost", "");
        this.iotbusQueues = this.env.getProperty("iotbus.rabbitmq.queues", "");
        this.iotbusTopics = this.env.getProperty("iotbus.rabbitmq.topics", "");
        this.iotbusRouting = this.env.getProperty("iotbus.rabbitmq.topic.routing", "");
        this.iotbusExchange = this.env.getProperty("iotbus.rabbitmq.topic.exchange", "true");
        this.logHost = this.env.getProperty("log.rabbitmq.host", "");
        this.logPort = this.env.getProperty("log.rabbitmq.port", "");
        this.logUsername = this.env.getProperty("log.rabbitmq.username", "");
        this.logPassword = this.env.getProperty("log.rabbitmq.password", "");
        this.logVirtualHost = this.env.getProperty("log.rabbitmq.virtualHost", "");
        this.logQueues = this.env.getProperty("log.rabbitmq.queues", "");
        this.logEnable = this.env.getProperty("log.enable", "");
        properties.setHost(this.host);
        properties.setPort(this.port);
        properties.setUsername(this.username);
        properties.setPassword(this.password);
        properties.setVirtualHost(this.virtualHost);
        properties.setQueues(this.queues);
        properties.setTopics(this.topics);
        properties.setRouting(this.routing);
        properties.setExchange(this.exchange);
        properties.setIotbusHost(this.iotbusHost);
        properties.setIotbusPort(this.iotbusPort);
        properties.setIotbusUsername(this.iotbusUsername);
        properties.setIotbusPassword(this.iotbusPassword);
        properties.setIotbusVirtualHost(this.iotbusVirtualHost);
        properties.setIotbusQueues(this.iotbusQueues);
        properties.setIotbusTopics(this.iotbusTopics);
        properties.setIotbusRouting(this.iotbusRouting);
        properties.setIotbusExchange(this.iotbusExchange);
        properties.setLogHost(this.logHost);
        properties.setLogPort(this.logPort);
        properties.setLogUsername(this.logUsername);
        properties.setLogPassword(this.logPassword);
        properties.setLogVirtualHost(this.logVirtualHost);
        properties.setLogQueues(this.logQueues);
        properties.setLogEnable(this.logEnable);
        return properties;
      }
    
      public boolean validQueue() {
        return StringUtil.isNumeric(this.port)
            && StringUtil.isAllNotBlank(this.host, this.username, this.password);
      }
    
      public boolean validIotbusQueue() {
        return StringUtil.isNumeric(this.iotbusPort)
            && StringUtil.isAllNotBlank(this.iotbusHost, this.iotbusUsername, this.iotbusPassword);
      }
    
      public boolean validLogQueue() {
        return StringUtil.isNumeric(this.logPort)
            && StringUtil.isAllNotBlank(this.logHost, this.logUsername, this.logPassword);
      }
    
      public String getHost() {
        return host;
      }
    
      public void setHost(String host) {
        this.host = host;
      }
    
      public String getPort() {
        return port;
      }
    
      public void setPort(String port) {
        this.port = port;
      }
    
      public String getUsername() {
        return username;
      }
    
      public void setUsername(String username) {
        this.username = username;
      }
    
      public String getPassword() {
        return password;
      }
    
      public void setPassword(String password) {
        this.password = password;
      }
    
      public String getVirtualHost() {
        return virtualHost;
      }
    
      public void setVirtualHost(String virtualHost) {
        this.virtualHost = virtualHost;
      }
    
      public String getQueues() {
        return queues;
      }
    
      public void setQueues(String queues) {
        this.queues = queues;
      }
    
      public String getTopics() {
        return topics;
      }
    
      public void setTopics(String topics) {
        this.topics = topics;
      }
    
      public String getRouting() {
        return routing;
      }
    
      public void setRouting(String routing) {
        this.routing = routing;
      }
    
      public String getExchange() {
        return exchange;
      }
    
      public void setExchange(String exchange) {
        this.exchange = exchange;
      }
    
      public String getIotbusHost() {
        return iotbusHost;
      }
    
      public void setIotbusHost(String iotbusHost) {
        this.iotbusHost = iotbusHost;
      }
    
      public String getIotbusPort() {
        return iotbusPort;
      }
    
      public void setIotbusPort(String iotbusPort) {
        this.iotbusPort = iotbusPort;
      }
    
      public String getIotbusUsername() {
        return iotbusUsername;
      }
    
      public void setIotbusUsername(String iotbusUsername) {
        this.iotbusUsername = iotbusUsername;
      }
    
      public String getIotbusPassword() {
        return iotbusPassword;
      }
    
      public void setIotbusPassword(String iotbusPassword) {
        this.iotbusPassword = iotbusPassword;
      }
    
      public String getIotbusVirtualHost() {
        return iotbusVirtualHost;
      }
    
      public void setIotbusVirtualHost(String iotbusVirtualHost) {
        this.iotbusVirtualHost = iotbusVirtualHost;
      }
    
      public String getIotbusQueues() {
        return iotbusQueues;
      }
    
      public void setIotbusQueues(String iotbusQueues) {
        this.iotbusQueues = iotbusQueues;
      }
    
      public String getIotbusTopics() {
        return iotbusTopics;
      }
    
      public void setIotbusTopics(String iotbusTopics) {
        this.iotbusTopics = iotbusTopics;
      }
    
      public String getIotbusRouting() {
        return iotbusRouting;
      }
    
      public void setIotbusRouting(String iotbusRouting) {
        this.iotbusRouting = iotbusRouting;
      }
    
      public String getIotbusExchange() {
        return iotbusExchange;
      }
    
      public void setIotbusExchange(String iotbusExchange) {
        this.iotbusExchange = iotbusExchange;
      }
    
      public String getLogHost() {
        return logHost;
      }
    
      public void setLogHost(String logHost) {
        this.logHost = logHost;
      }
    
      public String getLogPort() {
        return logPort;
      }
    
      public void setLogPort(String logPort) {
        this.logPort = logPort;
      }
    
      public String getLogUsername() {
        return logUsername;
      }
    
      public void setLogUsername(String logUsername) {
        this.logUsername = logUsername;
      }
    
      public String getLogPassword() {
        return logPassword;
      }
    
      public void setLogPassword(String logPassword) {
        this.logPassword = logPassword;
      }
    
      public String getLogVirtualHost() {
        return logVirtualHost;
      }
    
      public void setLogVirtualHost(String logVirtualHost) {
        this.logVirtualHost = logVirtualHost;
      }
    
      public String getLogQueues() {
        return logQueues;
      }
    
      public void setLogQueues(String logQueues) {
        this.logQueues = logQueues;
      }
    
      public String getLogEnable() {
        return logEnable;
      }
    
      public void setLogEnable(String logEnable) {
        this.logEnable = logEnable;
      }
    }

2、实现ChannelAwareMessageListener，并注册到SimpleMessageListenerContainer的方式，这种方式可以实现消息的手动应答。

MessageListener实现：

package com.spring.component.rabbitmq.listener;
    
    import com.rabbitmq.client.Channel;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
    import org.springframework.stereotype.Component;
    
    /**
     * 消费者，手动ack
     * 
     * @author liushihua
     * @since 2018年04月25日
     */
    @Component
    public class Receiver implements ChannelAwareMessageListener {
    
      private final Logger log = LoggerFactory.getLogger(this.getClass());
    
      @Override
      public void onMessage(Message message, Channel channel) throws Exception {
        log.info("接收消息：{}", message.toString());
        // 手动应答ack
        // channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        log.info("处理完成");
      }
    }

注册到SimpleMessageListenerContainer：

package com.spring.component.rabbitmq.config;
    
    import com.spring.component.rabbitmq.listener.Receiver;
    import org.springframework.amqp.core.AcknowledgeMode;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
    import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /**
     * 配置rabbitmq监听器
     * 
     * @author liushihua
     * @since 2018年04月25日
     */
    @Configuration
    public class ListenerConfig {
    
      @Autowired
      private Receiver receiver;
    
      /**
       * 注册一个简单的监听器容器 1、SimpleMessageListenerContainer 简单的监听器容器，不支持事务 2、DefaultMessageListenerContainer
       * 支持事务 3、ListenerContainer 不仅支持事务，还允许动态地管理JMS会话
       * 
       * @param connectionFactory
       * @return
       */
      @Bean
      public SimpleMessageListenerContainer listenerContainer(
          @Qualifier("connectionFactory") ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        // 设置connectionFactory
        container.setConnectionFactory(connectionFactory);
        // 设置消息json转换器
        container.setMessageConverter(new Jackson2JsonMessageConverter());
        // 队列
        container.setQueueNames("MSG_INBOUND_APP8075");
        container.setExposeListenerChannel(true);
        // 设置手动ack
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        // 设置容器的消费者的线程数
        container.setConcurrentConsumers(1);
        // 设置容器的最大消费者数
        container.setMaxConcurrentConsumers(10);
        // 讲一个MessageListener实现注入到监听容器
        container.setMessageListener(receiver);
        return container;
      }
    }

Rabbitmq支持多线程消费消息，可以在消息监听容器SimpleMessageListenerContainer设置线程数：

*  container.setConcurrentConsumers(1);设置容器的消费者的线程数
 *  container.setMaxConcurrentConsumers(10);设置容器的最大消费者数

也可以在消息监听容器设置消息手动应答： container.setAcknowledgeMode(AcknowledgeMode.MANUAL);