• rabbitmq-env.conf 环境信息配置

• rabbitmq.config　核心配置文件

该配置表示是的默认用户guest用户可以远程访问mq(广域网不能访问，内网可以访问)

3. RabbitMQ提供WEB-UI管理控制台，使用 rabbitmq-plugins enable rabbitmq_management命令启用，重启后可以看到

表明WEB-UI控制台启动成功，访问：http://localhost:15672/

• rabbitmqctl add_user test test

• rabbitmqctl set_user_tags test  administrator

  tag分为四种"management", "policymaker", "monitoring" "administrator"　详见　http://www.rabbitmq.com/management.html

RabbitMQ 其他

1. 创建连接工厂ConnectionFactory
2. 获取连接Connection
3. 通过连接获取通信通道Channel
4. 声明交换机Exchange：交换机类型分为四类：

   　　  FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念

   HeadersExchange ：通过添加属性key-value匹配

   DirectExchange:按照routingkey分发到指定队列

   TopicExchange:多关键字匹配

5. 声明队列Queue

6. 将队列和交换机绑定

7. 创建消费者

8. 执行消息的消费

否则回调函数无法调用

@Bean

@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)

//必须是prototype类型

public RabbitTemplate rabbitTemplate() {

    RabbitTemplate template = new RabbitTemplate(connectionFactory());

    return template;

}  

package com.lkl.springboot.amqp;  

import java.util.UUID;  

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.amqp.rabbit.support.CorrelationData;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Component;  

/**

 * 消息生产者

 *

 * @author liaokailin

 * @version $Id: Send.java, v 0.1 2015年11月01日 下午4:22:25 liaokailin Exp $

 */

@Component

public class Send implements RabbitTemplate.ConfirmCallback {  

    private RabbitTemplate rabbitTemplate;  

    /**

     * 构造方法注入

     */

    @Autowired

    public Send(RabbitTemplate rabbitTemplate) {

        this.rabbitTemplate = rabbitTemplate;

        rabbitTemplate.setConfirmCallback(this); //rabbitTemplate如果为单例的话，那回调就是最后设置的内容

    }  

    public void sendMsg(String content) {

        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());

        rabbitTemplate.convertAndSend(AmqpConfig.EXCHANGE, AmqpConfig.ROUTINGKEY, content, correlationId);

    }  

    /**

     * 回调

     */

    @Override

    public void confirm(CorrelationData correlationData, boolean ack, String cause) {

        System.out.println(" 回调id:" + correlationData);

        if (ack) {

            System.out.println("消息成功消费");

        } else {

            System.out.println("消息消费失败:" + cause);

        }

    }  

}  

/**

     * 针对消费者配置

        FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念

        HeadersExchange ：通过添加属性key-value匹配

        DirectExchange:按照routingkey分发到指定队列

        TopicExchange:多关键字匹配

     */

    @Bean

    public DirectExchange defaultExchange() {

        return new DirectExchange(EXCHANGE);

    }  

在Spring Boot中交换机继承AbstractExchange类

@Bean

  public SimpleMessageListenerContainer messageContainer() {

      SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());

      container.setQueues(queue());

      container.setExposeListenerChannel(true);

      container.setMaxConcurrentConsumers(1);

      container.setConcurrentConsumers(1);

      container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认

      container.setMessageListener(new ChannelAwareMessageListener() {  

          @Override

          public void onMessage(Message message, Channel channel) throws Exception {

              byte[] body = message.getBody();

              System.out.println("receive msg : " + new String(body));

              channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); //确认消息成功消费

          }

      });

      return container;

  }  

private volatile CacheMode cacheMode = CacheMode.CHANNEL;  

private volatile List<String> queueNames = new CopyOnWriteArrayList<String>();

1. container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认  

其中的isAutoStartup设置为true时，会自动调用Lifecycle接口中的start方法，既然我们为源码分析，也简单看下这个聪明的声明周期接口是如何实现它的聪明方法的

其中initLifecycleProcessor初始化生命周期处理器，

注册DefaultLifecycleProcessor对应bean

getLifecycleProcessor().onRefresh()调用DefaultLifecycleProcessor中方法onRefresh，调用startBeans(true)

其中

Map<String, Lifecycle> lifecycleBeans = getLifecycleBeans();  

获取所有实现Lifecycle接口bean，执行bean instanceof SmartLifecycle && ((SmartLifecycle)bean).isAutoStartup()判断，如果bean同时也为Phased实例，则加入到LifecycleGroup中，随后phases.get(key).start()调用start方法

接下来要做的事情就很明显：要了解消费者具体如何实现，查看SimpleMessageListenerContainer中的start是如何实现的。

至此～～整合RabbitMQ源码分析准备工作完成，下一篇中正式解读消费者的实现。

==============================

踩坑记录

近日在用spring boot架构一个微服务框架，服务发现与治理、发布REST接口各种轻松惬意。但是服务当设计MQ入口时，就发现遇到无数地雷，现在整理成下文，供各路大侠围观与嘲笑。

版本

当前使用的spring-boot-starter-amqp版本为2016.5发布的1.3.5.RELEASE

也许若干年后，你们版本都不会有这些问题了。：（

RabbitMQ

当需要用到MQ的时候，我的第一反映就是使用RabbitMQ，猫了一眼spring boot的官方说明，上面说spring boot为rabbit准备了spring-boot-starter-amqp，并且为RabbitTemplate和RabbitMQ提供了自动配置选项。暗自窃喜~~

瞅瞅[官方文档]http://docs.spring.io/spring-boot/docs/current-SNAPSHOT/reference/htmlsingle/#boot-features-rabbitmq和例子，SO EASY，再看一眼GITHUB上的官方例了，也有例子。

心情愉悦的照着例子，开干~~。

踩坑

十五分钟后的代码类似这样：

@Service

@RabbitListener(queues = "merchant")

public class MQReceiver  {

    protected Logger logger = Logger.getLogger(MQReceiver.class

            .getName()); 

    @RabbitHandler

    public void process(@Payload UpdateMerchant request) {

        UpdateMerchantResponse response = new UpdateMerchantResponse();

        logger.info(request.getMerchantId() + "->" + response.getReturnCode());

    }

}

消费信息后，应该记录一条日志。
结果得到只有org.springframework.amqp.AmqpException: No method found for class [B 这个异常，并且还无限循环抛出这个异常。。。

记得刚才官方文档好像说了异常什么的，转身去猫一眼，果然有：

If retries are not enabled and the listener throws an exception, by default the delivery will be retried indefinitely. You can modify this behavior in two ways; set the defaultRequeueRejected
 property to false
 and zero re-deliveries will be attempted; or, throw an AmqpRejectAndDontRequeueException
 to signal the message should be rejected. This is the mechanism used when retries are enabled and the maximum delivery attempts are reached.

知道了为啥会无限重试了，下面来看看为啥会抛出这个异常，google搜一下，貌似还有一个倒霉鬼遇到了这个问题。

进去看完问题和大神的解答，豁然开朗。

There are two conversions in the @RabbitListener pipeline.
The first converts from a Spring AMQP Message to a spring-messaging Message.
There is currently no way to change the first converter from SimpleMessageConverter which handles String, Serializable and passes everything else as byte[].
The second converter converts the message payload to the method parameter type (if necessary).
With method-level @RabbitListeners there is a tight binding between the handler and the method.
With class-level @RabbitListener s, the message payload from the first conversion is used to select which method to invoke. Only then, is the argument conversion attempted.
This mechanism works fine with Java Serializable objects since the payload has already been converted before the method is selected.
However, with JSON, the first conversion returns a byte[] and hence we find no matching @RabbitHandler.
We need a mechanism such that the first converter is settable so that the payload is converted early enough in the pipeline to select the appropriate handler method.
A ContentTypeDelegatingMessageConverter is probably most appropriate.
And, as stated in AMQP-574, we need to clearly document the conversion needs for a @RabbitListener, especially when using JSON or a custom conversion.

得嘞，官方示例果然是坑，试试大神的解决方案，手动新增下转换。

  @Bean

    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {

        RabbitTemplate template = new RabbitTemplate(connectionFactory);

        template.setMessageConverter(new Jackson2JsonMessageConverter());

        return template;

    }

    @Bean

    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {

        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();

        factory.setConnectionFactory(connectionFactory);

        factory.setMessageConverter(new Jackson2JsonMessageConverter());

        return factory;

    }

然后在生产和消费信息的地方使用他们：

@RabbitListener(queues = "merchant", containerFactory="rabbitListenerContainerFactory")

public void process(@Payload UpdateMerchant request) {

     UpdateMerchantResponse response = new UpdateMerchantResponse();

    logger.info(request.getMerchantId() + "->" + response.getReturnCode());

 }

再来一次，果然可以了

c.l.s.m.service.MQReceiver : 00000001->null