RabbitMQ整合Spring-AMQP-----Simple

SimpleMessageListenerContainer 这个类主要是为消费端使用！
通过这个类，我们可以做：

• 监听队列（或者多个队列）、自动启动、自动声明
• 设置事物特性、事物回滚、事物属性、事物容量（并发）、是否开启事物、回滚事物等！
• 设置消费者数量、最小最大数量、批量消费
• 设置消息确认和自动确认模式、是否重回队列、异常捕获handler函数。
• 设置消费者标签生成策略、是否独占模式、消费者属性等！

常见的API：

• setQueues 设置监听的队列
• setConcurrentConsumers 设置当前消费者的数量
• setMaxConcurrentConsumers 设置最大的消费者的数量
• setDefaultRequeueRejected 是否重回队列
• setAcknowledgeMode 设置签收模式
• setConsumerTagStrategy 消费端的标签生成策略
• setMessageListener 设置消息的监听

使用：
只声明下这个 @Bean

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        // 只有设置为 true，spring 才会加载 RabbitAdmin 这个类
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }

    @Bean
    public SimpleMessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        // 设置需要监听的队列
        // 可以放入多个 queue 实例
        container.setQueues(queue());
        // 可以设置当前消费者的数量
        container.setConcurrentConsumers(1);
        // 可以设置最大的消费者的数量
        container.setMaxConcurrentConsumers(5);
        // 可以是否重回队列，一般都不允许重回队里
        container.setDefaultRequeueRejected(false);

        // 可以设置签收模式, 比如设置为 自动签收
        container.setAcknowledgeMode(AcknowledgeMode.AUTO);

        ..........
        ..........
        return container;
    }

在上面简单的配置了下 SimpleMessageListenerContainer 这个类！

这个SimpleMessageListenerContainer 可以设置 setConsumerTagStrategy 消费端的标签生成策略:

      // 可以设置消费端的标签，生成策略，我们可以自定义消费端的标签生成策略
        // 标签 可以标识这个消息的唯一性
        container.setConsumerTagStrategy(new ConsumerTagStrategy() {
            @Override
            public String createConsumerTag(String queue) {
                return queue + "_" + UUID.randomUUID().toString();
            }
        });

注意：生成的标签，是唯一的！不重复！这个标签可以标识着一条消息，在消息被手动签收的时候，是根据这个 标签来的！

还可以设置息的监听(也就是消费消息)

        // 监听消息
        container.setMessageListener(new ChannelAwareMessageListener() {
            /**
             * 如果有消息传递过来，就会进入这个 onMessage 方法
             * @param message 消息
             * @param channel 消息管道
             * @throws Exception
             */
            @Override
            public void onMessage(Message message, Channel channel) throws Exception {
                String msg = new String(message.getBody(), "UTF-8");
                System.err.println("-----消费者接收到消息：" + msg);

            }
        });

简单的深入了解一下 SimpleMessageListenerContainer 这个类

关于queue相关的！

5555.png

重点看这个 addQueues 这个队列的添加方法！

666.png

我们发现这个方法里面只有两步:

• 第一个调用了抽象的父类AbstractMessageListenerContainer 的 addQueues 方法，进行添加队列
• 第二步，调用了 queusChanged 方法!

直接看 queusChanged 方法

9999.png

发现里面在遍历消费者，然后调用了 basicCancel 方法设置为 true
看看 basicCancel 方法

1010.png

发现这个 basicCancel 方法先判断了 channel 管道是否打开了，然后调用了 channel.basicCancel(...) 方法!

channel.basicCancel(...) 是 RabbitMQ官方提供的关闭消费者方法！

到这里可以发现 addQueues 队列，实际上把队列添加到父类中的CopyOnWriteArrayList 中，然后把所有的消费者进行了关闭！
其实在最后面还调用了 addAndStartConsumers 这个方法，打开所有的消费者, 也就是唤醒消费者！

在这里其实可以发现，我们可以动态的添加想要监听的队列!
在之前使用的时候 RabbitMQ API 的时候，queue都是确定好的，但是如果要要添加新的监听队列需要重新把这个队列添加进来, 然后在把项目重新启动起来！

除了可以动态的配置queue之外。
SimpleMessageListenerContainer可以进行动态设置，比如在运行中的应用可以动态的修改其消费者数量的大小、接收消息的模式等！

其实在 spring-amqp 上已经有了介绍，SimpleMessageListenerContainer 是可配置的！

很多基于RabbitMQ的自制定化后端管控台在进行动态设置的时候，也是根据这一特点去实现。所以可以看出 SpringAMQP非常的强大！

如有错误地方，欢迎指出！大家共同进步！