一·简介

1丶为什么要使用消息队列

https://wenku.baidu.com/view/e297236f83c4bb4cf7ecd193.html

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

①异步处理(高并发)

②系统解耦

③流量削锋

2丶为什么使用RabbitMQ

①给予AMQP协议

②高并发

③高可用

④强大的社区支持，以及很多公司都在使用

⑤高性能

⑥支持插件(监控管理界面的插件，安装插件支持jms)

⑦支持多语言(PHP,Python,.net)

3丶为什么使用Spring AMQP

①基于Spring之上，社区活跃

②对AMQP协议进行了高度的封装

③极大简化了RabbitMQ的操作

④易用性，可扩展性

二·AMQP协议介绍

AMQP协议是一个二进制协议,一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准,为面向消息的中间件设计。并不受客户端/中间件不同产品，不同的开发语言等条件的限制。Erlang中的实现有 RabbitMQ等。

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

三·RabbitMQ安装

1丶特点

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

2丶Windows下安装

①安装Erlang

Erlang:是一种通用的面向并发的编程语言,Erlang是运行于虚拟机的解释性语言,Erlang支持脚本式解释器,Erlang属于多重范型编程语言，涵盖函数式、并发式及分布式。顺序执行的Erlang是一个及早求值, 单次赋值和动态类型的函数式编程语言。

下载地址:http://www.erlang.org/downloads

配置环境变量: ERLANG_HOME: D:\Develop\Erlang\erl9.3

Path:%ERLANG_HOME%\bin;

测试安装:

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

②安装RabbitMQ

下载地址: http://www.rabbitmq.com/install-windows.html(注意安装目录不要有空格)

Erlang与RabbitMQ版本兼容:

http://www.rabbitmq.com/which-erlang.html#unsupported versions

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

③安装插件

在D:\Develop\RabbitMQ\rabbitmq_server-3.7.6\sbin运行cmd:

rabbitmq-plugins enable rabbitmq_management

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

四·exchange详解

1丶exchange介绍

接受生产者的消息，根据路由键转发消息到绑定队列。

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

类型：

Direct exchange

Topic exchange

Fanout exchange

Headers exchange

属性：

Name: 名字(一个virtual host里面有若干个exchange，但是名字不能重复)

Type: exchange的类型

Durability: 是否持久化(保证消息的可靠性不会被丢失)

[(默认)Durable(持久化)，Transient(不持久化)]

Auto delete：当最后一个绑定消息被删除，exchange是否自动删除

[(默认)No,Yes]

Internal: 这个exchange只能被内部使用，生产者发送消息只能发送到internal为true的交换机上

[(默认)No,Yes]

Arguments: 在AMQP协议的参数(上面的那些属性)上进行扩展

例如:alternate-exchange : 当我们交换机不能通过路由键路由到指定的消息队列上去的时候，就可以把消息转发到另一个exchange上面去，保证消息不会被丢失

2丶Direct Exchange

将消息中的Routing Key(在发送一个消息的时候会指定routing key)与exchange中关联的所有binding中的Routing Key进行比较(exchange与queue进行binding的时候会指定一个routingkey)，如果相等，则发送到该binding对应的queue中。

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

一个routing key可以binding多个queue

一个exchange可以binding多个queue，在binding多个queue的时候可以指定相同的 routing key 也可以指定不同的routing key

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

默认交换隐式地绑定到每个队列，并使用与队列名相等的路由键。不可能显式地绑定到默认交换，或从默认交换中解除绑定。它也不能被删除。

不能进行binding操作
任何发送到该exchange中的消息都会被转发到Routing key指定的queue中(在发送消息时会指定一个routing key ，但是default exchange又没有与queue进行binding，那么发送到default exchange的消息就会发送到与发送消息相同的routing key的消息队列中)（这里routing key的名称即:queue的名称）；
如果vhost不存在routing key中指定的队列名，则该消息会被抛弃；

3丶topic Exchage

将消息中的Routing key与该Exchange 关联的所有binding中的Routing key进行比较，若果匹配上了，则发送到该binding对应的queue中。

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

特殊情况:

如果binding中的Routing key没有*/#，那就是相当于direct exchage(相等转发)

如果binding中的Routing key为#.#就相当于 fanout exchange(全转发 )

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

4丶fanout Exchang

直接将消息转发到所有binding的queue中，忽略Routing key.

特点：

效率最高

Fanout>direct>topic

5丶Headers Exchaneg

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

转发时，是让发送时的headers与binding时arguments进行比较，binding时至少要有两个参数其中一个必须是x-match=all/any，如果是all那么，发送的参数必须大于等于binding时的参数（参数类型也需要一致），如果没有设置x-match那么默认为all。

五·binding详解

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

六·queue详解

也称为Message Queue,消息队列，保存消息并将他们转发给消费者。

属性:

Virtualhost：指定是属于哪个vhost，一个vhost有很多queue,但名字是唯一的；

Name: queue的名字；

Durability：是否持久化，当rabbitmq重启或最后一个消息被消费者消费queue是否移除。 (durable：是,transient:否)

Auto delete：yes,当最后一个监听被移除后，该queue会自动删除；

七·message详解

服务器和应用程序之间发送的数据，由properties和payload（body）组成

属性：

Routing Key: 在消息转发时需要(fanout exchang,headers exchange不需要指定)

Delivery mode：消息是否持久化(1不持久化，2持久化)，（当消息发到queue中，当计算机宕机或者重启的时候消息就会丢失）

Header: 消息头(键值对)

Payload: 发送的消息

保证消息持久化:exchange,queue,message三者持久化

八·rabbitmq java client

1丶基本操作

①：创建工厂以及连接和信道

//创建工厂
ConnectionFactory cf=new ConnectionFactory();
cf.setHost("localhost");
cf.setPort(5672);
cf.setUsername("guest");
cf.setPassword("guest");

//另一种创建工厂的方式默认的vhost是"/"可以在5672后面指定vhost

cf.setUri("amqp://guest:[email protected]:5672");

//创建连接
// Connection con=cf.newConnection();
Connection con=cf.newConnection("测试连接");

//创建信道
Channel channel = con.createChannel();

注：使用guest用户只能登陆localhost不能带ip(192.168...)

默认端口是5672而15672是可视化界面的端口

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

②：创建交换机

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

//创建交换机
Map<String,Object> map=new HashMap<String,Object>();
map.put("alternate-exchange","direct_test");
channel.exchangeDeclare("exchange-Test", BuiltinExchangeType.DIRECT, true, false, map);

//判断exchange是否存在(不存在就抛异常)
channel.exchangeDeclarePassive("exchange-Test");

//删除Exchange
channel.exchangeDelete("exchange-Test");

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

③：创建消息队列

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

//创建消息队列
/**
* exclusive：只允许当前connection访问,
* 如果当前connection(程序关闭)没有访问就会自动删除
*/
channel.queueDeclare("queue-Test", true, false, false, new HashMap<String, Object>());

//判断队列是否存在(不存在就抛异常)
channel.queueDeclarePassive("queue-Test");

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

④：binding

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

//exchange和queue进行binding
channel.queueBind("queue-Test","exchange-Test","test");

//exchange和exchange进行binding
/**
* 把resource交换机与destinatio交换机进行binding
* 当想source交换机发消息时消息会被转发到detination交换机上
*/
channel.exchangeBind("exchange-Test","direct_test","test");

//exchange与queue进行解绑
channel.queueUnbind("queue-Test","exchange-Test","test");
//exchange与exchange进行解绑
channel.exchangeUnbind("exchange-Test","direct_test","test");

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

⑤：关闭连接释放资源

channel.close();
con.close();

2丶消息的发送与消费

①：使用默认的exchange发送消息(default Exchange)

//向默认的exchange发送消息
/**
* BasicProperties设置消息的属性(propertis)
* Builder后面跟参数
*/
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().deliveryMode(2).contentEncoding("UTF-8").build();

/**
* 默认的交换机发消息routing key的名称就是queue的名称
*/
channel.basicPublish("","queue-Test",props,"生成者发送消息".getBytes());

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

如果requeue选择为false在 getMessage过后消息就从消息队列中移除。

②：向topic exchange发送消息

//创建exchange
channel.exchangeDeclare("push", BuiltinExchangeType.TOPIC, true, false, null);
//binding
channel.queueBind("queue-Test","push","#");
//发送消息
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().deliveryMode(2).contentEncoding("UTF-8").build();
channel.basicPublish("push","我不管指定什么都会发送",props,"我又来了".getBytes());

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

③：消费者消费消息

//创建工厂
ConnectionFactory cf=new ConnectionFactory();
cf.setHost("localhost");
cf.setPort(5672);
cf.setUsername("guest");
cf.setPassword("guest");
//创建连接
Connection con=cf.newConnection();
//创建信道
Channel channel = con.createChannel();
//消费者消费消息
channel.basicConsume("queue-Test",true,new ConsumerService(channel));

Thread.sleep(30000);

channel.close();
con.close();

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

注：使用channel.basicConsume（“对列名称”，”是否确认消息”，”回调函数”）（要睡够才会出现效果）

回调函数：

public class ConsumerService extends DefaultConsumer {

    public ConsumerService(Channel channel) {
        super(channel);
    }

    /**
     * 获取消息后的回调函数
     * @param consumerTag
     * @param envelope
     * @param properties
     * @param body
     * @throws IOException
     */
    @Override
    public void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)
            throws IOException{
        System.out.println(consumerTag);
        System.out.println("消息接收到了");
        System.out.println("-------------------------");
        System.out.println("routing key:"+envelope.getRoutingKey()+"-----exchange:"+envelope.getExchange());
    }

}

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

九·Spring AMQP

1丶简介

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

依赖：

<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit</artifactId>
    <version>1.7.3.RELEASE</version>
</dependency>

2丶RabbitAdmin创建详解

①：使用

Java配置类

@Configuration
public class MQConfig {

    @Bean
    public ConnectionFactory connectionFactory(){
        CachingConnectionFactory factory=new CachingConnectionFactory();
        factory.setUri("amqp://guest:[email protected]:5672");
        return factory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        return rabbitAdmin;
    }
}

开启注解扫描

@ComponentScan
public class Start {
    public static void main(String[] args) {
        //开启注解扫描
        AnnotationConfigApplicationContext context=new AnnotationConfigApplicationContext(Start.class);
        RabbitAdmin rabbitAdmin=context.getBean(RabbitAdmin.class);

        System.out.println(rabbitAdmin);
        context.close();
    }
}

使用rabbitAdmin创建与绑定

//创建exchange
rabbitAdmin.declareExchange(new DirectExchange("rabbitAdmin-TopicExchange",
                                true,false));
//创建queue
rabbitAdmin.declareQueue(new Queue("rabbitAdmin-queue",
                    true,false,false));
//exchange与queue进行binding
rabbitAdmin.declareBinding(new Binding("rabbitAdmin-queue",
        Binding.DestinationType.QUEUE,"rabbitAdmin-TopicExchange",
        "#",new HashMap<String,Object>()));

//使用bindingBuilder进行binding
rabbitAdmin.declareBinding

(BindingBuilder.bind(new Queue("rabbitAdmin-queue")).
to(new DirectExchange("rabbitAdmin-TopicExchange")).with("#"));

Exchange

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

Queue

Binding

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

BindingBuilder

注：

exchange与exchange进行binding

使用binding时需要修改String destination

Binding.DestinationType.EXCHANGE；

使用BingdingBuilder的时候:(exchange1).to(exchange2).with(“routingkey”)是exchange2绑定exchange1

②：exchange,binding,queue的自动声明

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

配置类

}

@Configuration
public class DecalreConfig {
    @Bean
    public Queue q1(){
        return new Queue("q1");
    }
    @Bean
    public Exchange e1(){
        return new TopicExchange("e1",true,false);
    }
    @Bean
    public Binding b1(){
        HashMap<String, Object> map = new HashMap<String, Object>();
        return new Binding("q1", Binding.DestinationType.QUEUE, "e1", "#",map);
    }
}

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

3丶RabbitTemplate进行消息发送

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

配置类

@Configuration
public class MQConfig {

@Bean
public ConnectionFactory connectionFactory(){
    CachingConnectionFactory factory=new CachingConnectionFactory();
    factory.setUri("amqp://guest:[email protected]:5672");
    factory.setCacheMode(CachingConnectionFactory.CacheMode.CHANNEL);
    return factory;
}

@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

        //设置tempalte默认的Routingkey
        rabbitTemplate.setRoutingKey("anyKey");

        //设置template默认的exchange
        rabbitTemplate.setExchange("e1");

        return rabbitTemplate;
    }
}

使用send发送消息

AnnotationConfigApplicationContext context=new AnnotationConfigApplicationContext(TemplateSatrt.class);
RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);
System.out.println(rabbitTemplate);

//设置message对象
MessageProperties mp = new MessageProperties();

mp.setHeader("type","2");
mp.setHeader("id","haha");

Message message = new Message("hello".getBytes(), mp);

//向template默认的exchange和routingkey发消息
rabbitTemplate.send(message);

//向template默认的exchange，指定的routing key发消息
rabbitTemplate.send("test.anyKey",message);

//向指定的exchange和routingkey发消息
rabbitTemplate.send("","q1",message);

//设置唯一标识的消息
rabbitTemplate.send("e1","anykey",message,new CorrelationData("messageId"));

context.close();

使用convertAndSend发消息

//向template默认的exchange和routingkey发消息
rabbitTemplate.convertAndSend("anything");

//向template默认的exchange，指定的routing key发消息
rabbitTemplate.convertAndSend("test.anyKey","haha");

//向指定的exchange和routingkey发消息
rabbitTemplate.convertAndSend("","q1","haha");

//设置唯一标识的消息
rabbitTemplate.convertAndSend("e1","anykey","haha",new CorrelationData("messageId"));

//使用message后置处理器发送消息
rabbitTemplate.convertAndSend("", "q1", "hahhaaha", new MessagePostProcessor() {
    public Message postProcessMessage(Message message) throws AmqpException {
        System.out.println("我先对消息处理过后才会发消息");
        message.getMessageProperties().getHeaders().put("type","1");
        message.getMessageProperties().getHeaders().put("messageId","monitor");
        System.out.println(message);
        return message;
    }
});

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

4丶消息的消费

①：使用容器的方式进行创建消费者消费

首先需要一个MessageListenerContainer

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

选择它的实现类：SimpleMessageListenerContainer

创建SimpleMessageListenerContainer需要三个东西：

1.连接工厂：ConnectionFactory

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

2.监听队列的名称：setQueueName(可以是多个队列)

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

3.收到消息的回调函数

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

checkMessageListener:

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

创建

使用

@ComponentScan
public class Consumer {
    public static void main(String[] args) {
        AnnotationConfigApplicationContext context=new AnnotationConfigApplicationContext(Consumer.class);
       context.getBean(SimpleMessageListenerContainer.class);
        //设置了不自动启动，这里就就要手动启动
//       context.getBean(SimpleMessageListenerContainer.class).start();
    }
}

注：

当同一个queue上有多个消息者时，一个消息发送过来只会有一个消费者收到消息

SimpleMessageListenerContainer支持一次监听多个queue

SimpleMessageListenerContainer支持运行时动态增加/移除queue

@Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer
            (ConnectionFactory connectionFactory, RabbitTemplate rb) {
        SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer(connectionFactory);
        //是否自动启动
//        simpleMessageListenerContainer.setAutoStartup(f

        //设置从哪个队列取消息(可以取多个)
//        simpleMessageListenerContainer.setQueueNames("q1", "q2", "q3");
        simpleMessageListenerContainer.setQueueNames("q1");

        //设置接受消息后的回调
        simpleMessageListenerContainer.setMessageListener(new MessageListener() {
            public void onMessage(Message message) {
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("***********消费者接受到了" +
                        message.getMessageProperties().getConsumerQueue() +
                        "消息了**************");
                System.out.println(new String(message.getBody()));
            }
        });
        return simpleMessageListenerContainer;
    }

AnnotationConfigApplicationContext context=new AnnotationConfigApplicationContext(Consumer.class);

        SimpleMessageListenerContainer bean = context.getBean(SimpleMessageListenerContainer.class);
        bean.setQueueNames("q2","q3");
        bean.setMessageListener(new MessageListener() {
            public void onMessage(Message message) {
                System.out.println("收到"+message.getMessageProperties().getConsumerQueue()+"的消息");
                System.out.println(new String(message.getBody()));
            }
        });

bean.removeQueueNames("q2","q3");
//设置了不自动启动，这里就就要手动启动
// context.getBean(SimpleMessageListenerContainer.class).start();

//设置多少个并发消费者一起消费
bean.setConcurrentConsumers(5);
//设置最大的并发消费者数
bean.setMaxConcurrentConsumers(10);

5丶MessageListenerAdaptor

①：基本操作

配置类

代理方法

public class MesageAdaptor {
    public void handleMessage(byte bytes[]){
        System.out.println("----------------handleMessage消息:"+new String(bytes)+"-------");
    }
    public void send2(byte bytes[]){
        System.out.println("----------------send2消息:"+new String(bytes)+"-------");
    }
    public void send3(byte bytes[]){
        System.out.println("----------------send3消息:"+new String(bytes)+"-------");
    }
}

启动类

@ComponentScan
public class Consumer {
public static void main(String[] args) throws Exception{
AnnotationConfigApplicationContext context=new AnnotationConfigApplicationContext(Consumer.class);

}

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

②：MessageConvertor

MessageListenerAdaptor内部通过MessageConvertor把Message对象转换成java对象，然后去消息适配器内找到对应的方法(参数为转换后的java对象的类型)。

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

实现类：

public class MessageConvertorImpl implements MessageConverter{
    /**
     * 把Object转换为Message对象
     * @param object
     * @param messageProperties
     * @return
     * @throws MessageConversionException
     */
    public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
        System.out.println("----------toMessage----------");
        return new Message(object.toString().getBytes(),messageProperties);
    }

    /**
     * 把Message对象装换为object对象
     * @param message
     * @return
     * @throws MessageConversionException
     */
    public Object fromMessage(Message message) throws MessageConversionException {
        System.out.println("----------fromMessage----------");
        String contentType = message.getMessageProperties().getContentType();
        if(contentType!=null){
            if(contentType.startsWith("text")){
                return new String(message.getBody());
            }
        }
        return message.getBody();
    }
}

消息适配器类：

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

/**
* Converts from a AMQP Message to an Object.
*/
@Override
public Object fromMessage(Message message) throws MessageConversionException {
   Object content = null;
   MessageProperties properties = message.getMessageProperties();
   if (properties != null) {
      String contentType = properties.getContentType();
      if (contentType != null && contentType.startsWith("text")) {
         String encoding = properties.getContentEncoding();
         if (encoding == null) {
            encoding = this.defaultCharset;
         }
         try {
            content = new String(message.getBody(), encoding);
         }
         catch (UnsupportedEncodingException e) {
            throw new MessageConversionException(
                  "failed to convert text-based Message content", e);
         }
      }
      else if (contentType != null &&
            contentType.equals(MessageProperties.CONTENT_TYPE_SERIALIZED_OBJECT)) {
         try {
            content = SerializationUtils.deserialize(
                  createObjectInputStream(new ByteArrayInputStream(message.getBody()), this.codebaseUrl));
         }
         catch (IOException e) {
            throw new MessageConversionException(
                  "failed to convert serialized Message content", e);
         }
         catch (IllegalArgumentException e) {
            throw new MessageConversionException(
                  "failed to convert serialized Message content", e);
         }
         catch (IllegalStateException e) {
            throw new MessageConversionException(
                  "failed to convert serialized Message content", e);
         }
      }
   }
   if (content == null) {
      content = message.getBody();
   }
   return content;
}

③：Jackson2MessageConverter

生产者在发送json数据的时候，需要指定这个json是哪个对象，否则消费者收到消息之后，不知道要转换成哪个java对象。

指定方法在消息header中，增加一个__TypeId__，value就是具体的java对象全称（包名+类名）(一定要是消费者所在系统的java对象全称)

发送消息的时候，__TypeId__的值可以是java对象全称，也可以是映射的key，当消费者有配置映射key的时候，生产者既可以指定java对象全称，又可以是映射的key，如果消费者没有配置映射key，则只能指定java对象全称。

如果生产者发送的是list的json数据，则还需要增加一个__ContentTypeId__的header，用于指明List里面的具体的对象

如果生产者发送的是map的json数据，则需要指定__KeyTypeId__、__ContentTypeId__的header，用于指明map里面的key，value的具体对象

生产者：

配置文件

@Configuration
public class MQConfig {

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setUri("amqp://guest:[email protected]:5672");
        factory.setCacheMode(CachingConnectionFactory.CacheMode.CHANNEL);
        return factory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

        //设置tempalte默认的Routingkey
        rabbitTemplate.setRoutingKey("q2.aaa");

        //设置template默认的exchange
        rabbitTemplate.setExchange("e1");

        return rabbitTemplate;
    }

}

实体

public class Order {
    private Integer id;
    private Integer userId;
    private double amount;
    private String time;
    public Integer getId() {
        return id;
    }
    public void setId(Integer id) {
        this.id = id;
    }
    public Integer getUserId() {
        return userId;
    }
    public void setUserId(Integer userId) {
        this.userId = userId;
    }
    public double getAmount() {
        return amount;
    }
    public void setAmount(double amount) {
        this.amount = amount;
    }
    public String getTime() {
        return time;
    }
    public void setTime(String time) {
        this.time = time;
    }
    @Override
    public String toString() {
        return "Order [id=" + id + ", userId=" + userId + ", amount=" + amount + ", time=" + time + "]";
    }
}

启动类

@ComponentScan
public class Productor {
    public static void main(String[] args) throws Exception{
        AnnotationConfigApplicationContext context=new AnnotationConfigApplicationContext(Productor.class);
        RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);
        Order order = new Order();
        order.setId(1);
        order.setUserId(1000);
        order.setAmount(88d);
        order.setTime(LocalDateTime.now().toString());

        Order order1 = new Order();
        order1.setId(1);
        order1.setUserId(1000);
        order1.setAmount(88d);
        order1.setTime(LocalDateTime.now().toString());

        HashMap<String, Order> orderMap = new HashMap<String, Order>();
        orderMap.put("order",order);
        List<Order> list= Arrays.asList(order,order1);
        ObjectMapper objectMapper = new ObjectMapper();

        String jsonList = objectMapper.writeValueAsString(list);
        String jsonMap=objectMapper.writeValueAsString(orderMap);
        String jsonOrder = objectMapper.writeValueAsString(order);

        MessageProperties messageProperties = new MessageProperties();
        //设置contentType
        messageProperties.setContentType("application/json");

        //消费者实体对象的全称
                //对象
//        messageProperties.setHeader("__TypeId__", "springAMQP.Order");
        messageProperties.setHeader("__TypeId__", "order");
                //List
//         messageProperties.setHeader("__TypeId__","java.util.List");
//         messageProperties.setHeader("__ContentTypeId__","order");
                 //map
//        messageProperties.setHeader("__TypeId__","java.util.Map");
//        messageProperties.setHeader("__KeyTypeId__","java.lang.String");

//        rabbitTemplate.send(new Message(jsonList.getBytes(),messageProperties));
//        rabbitTemplate.send(new Message(jsonMap.getBytes(),messageProperties));
        rabbitTemplate.send(new Message(jsonOrder.getBytes(),messageProperties));

        context.close();
    }
}

消费者

配置类

@Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer(ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer simpleMessageListenerContainer =
                new SimpleMessageListenerContainer(connectionFactory);
        simpleMessageListenerContainer.setQueueNames("q1", "q2", "q3");

        //使用MessageListeneradaptor适配器来代理消息监听器
        MessageListenerAdapter adapter = new MessageListenerAdapter(new MesageAdaptor());

        //设置jackson接受消息
        Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();

        DefaultJackson2JavaTypeMapper javaTypeMapper = new DefaultJackson2JavaTypeMapper();
        //配置映射key(这样生产者在发送消息的时候就不用指定全类名)
        HashMap<String, Class<?>> idClassMap = new HashMap<String, Class<?>>();
        idClassMap.put("order",Order.class);
        javaTypeMapper.setIdClassMapping(idClassMap);

        jackson2JsonMessageConverter.setJavaTypeMapper(javaTypeMapper);

        //设置消息转换器
        adapter.setMessageConverter(jackson2JsonMessageConverter);

        //设置默认方法
//        adapter.setDefaultListenerMethod("send2");
        //把不同的消息交给不同的方法来处理
        HashMap<String, String> map = new HashMap<String, String>();
        map.put("q1", "handleMessage");
        map.put("q2", "send2");
        map.put("q3", "send3");
        adapter.setQueueOrTagToMethodName(map);
        simpleMessageListenerContainer.setMessageListener(adapter);

        return simpleMessageListenerContainer;
    }
}

对应生产者的实体

适配器类

public class MesageAdaptor {
    public void handleMessage(byte bytes[]){
        System.out.println("----------------handleMessage消息(bytes[]):"+new String(bytes)+"-------");
    }

    /**
     * 接受String类型
     * @param bytes
     */
    public void send2(String bytes){
        System.out.println("----------------send2消息(String):"+new String(bytes)+"-------");
    }

    /**
     * 接受字节类型
     * @param bytes
     */
    public void send2(byte bytes[]){
        System.out.println("----------------send2消息(bytes[]):"+new String(bytes)+"-------");
    }
    /**
     * 接受字节类型
     * @param order
     */
    public void send2(List order){
        System.out.println("----------------send2消息(List order):"+order.get(0)+"-------");
    }
    /**
     * 接受字节类型
     * @param order
     */
    public void send2(Order order){
        System.out.println("----------------send2消息(Order order):"+order.toString()+"-------");
    }
    /**
     * 接受字节类型
     * @param order
     */
    public void send2(Map order){
        System.out.println("----------------send2消息(Map order):"+order.toString()+"-------");
    }

    public void send3(byte bytes[]){
        System.out.println("----------------send3消息:"+new String(bytes)+"-------");
    }
}

启动类

}
}

④：ContentTypeDelegatingMessageConverter

ContentTypeDelegatingMessageConverter 是一个代理的MessageConverter。

ContentTypeDelegatingMessageConverter本身不做消息转换的具体动作，而是把消息转换委托给具体的MessageConverter，我们可以设置ContentType和MessageConverter的映射关系。

ContentTypeDelegatingMessageConverter还有一个默认的MessageConverter，

也就是说当根据ContentType没有找到映射的MessageConverter的时候，就会使用默认的MessageConverter。

@Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer(ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer simpleMessageListenerContainer =
                new SimpleMessageListenerContainer(connectionFactory);
        simpleMessageListenerContainer.setQueueNames("q1", "q2", "q3");

        //使用MessageListeneradaptor适配器来代理消息监听器
        MessageListenerAdapter adapter = new MessageListenerAdapter(new MesageAdaptor());

        //设置jackson接受消息
        Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
        DefaultJackson2JavaTypeMapper javaTypeMapper = new DefaultJackson2JavaTypeMapper();
        //配置映射key(这样生产者在发送消息的时候就不用指定全类名)
        HashMap<String, Class<?>> idClassMap = new HashMap<String, Class<?>>();
        idClassMap.put("order",Order.class);
        javaTypeMapper.setIdClassMapping(idClassMap);
        jackson2JsonMessageConverter.setJavaTypeMapper(javaTypeMapper);

        //使用ContentTypeDelegatingMessageConverter代理MessageConvertor
        ContentTypeDelegatingMessageConverter converter = new ContentTypeDelegatingMessageConverter();
        //json的convertor
        converter.addDelegate("json",jackson2JsonMessageConverter);
        //字符串的convertor
        converter.addDelegate("",new MessageConvertorImpl());

        //设置消息转换器
        adapter.setMessageConverter(converter);

        //设置默认方法
//        adapter.setDefaultListenerMethod("send2");
        //把不同的消息交给不同的方法来处理
        HashMap<String, String> map = new HashMap<String, String>();
        map.put("q1", "handleMessage");
        map.put("q2", "send2");
        map.put("q3", "send3");
        adapter.setQueueOrTagToMethodName(map);
        simpleMessageListenerContainer.setMessageListener(adapter);

        return simpleMessageListenerContainer;
    }

6丶RabbitListenerConfigurer进行消费

1：实现RabbitListenerConfigurer接口，并把实现类托管到spring容器中

2：在spring容器中，托管一个RabbitListenerContainerFactory的bean（SimpleRabbitListenerContainerFactory）
3：在启动类上面加上@EnableRabbit

配置类

@Configuration
public class MQconfig {

    @Bean
    /**
     * 连接工厂
     */
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setUri("amqp://guest:[email protected]:5672");
        factory.setCacheMode(CachingConnectionFactory.CacheMode.CHANNEL);
        return factory;
    }

    @Bean
    /**
     * 连接以及binding
     */
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }

    @Bean
    /**
     * 发送消息
     */
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

        //设置tempalte默认的Routingkey
//        rabbitTemplate.setRoutingKey("anyKey");

        //设置template默认的exchange
//        rabbitTemplate.setExchange("e1");
        return rabbitTemplate;
    }

    @Bean
    public RabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        return factory;
    }

    @Bean
    /**
     * 接收消息
     */
    public RabbitListenerConfigurer rabbitListenerConfigurer() {
        return new RabbitListenerConfigurer() {
            public void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) {

                SimpleRabbitListenerEndpoint endpoint1 = new SimpleRabbitListenerEndpoint();
                endpoint1.setId("aaa");
                endpoint1.setQueueNames("q1");
                //接受消息后的回调
                endpoint1.setMessageListener(new MessageListener() {
                    public void onMessage(Message message) {
                        System.out.println("-------接收到消息了："+new String(message.getBody())+"--------");
                    }
                });

                SimpleRabbitListenerEndpoint endpoint2 = new SimpleRabbitListenerEndpoint();
                endpoint2.setId("bbb");
                endpoint2.setQueueNames("q2","q3");
                endpoint2.setMessageListener(new MessageListenerAdapter(new MesageAdaptor()));

                registrar.registerEndpoint(endpoint1);
                registrar.registerEndpoint(endpoint2);
            }
        };
    }
}

消息适配器

启动类

@EnableRabbit
@ComponentScan
public class Start {
    public static void main(String[] args) {
        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Start.class);
//        context.close();
    }
}

7丶使用@rabbitlistener注解进行消费

①：实现步骤

在启动类里加一个@EnableRabbit的注解

在spring容器中托管一个RabbitListenerContanierFactory的Bean(默认的实现是SimpleRabbitListenerContaninerFactory)

写一个消息处理类托管到spring容器中,然后在具体方法上加上注解@RabbitListenner

启动类：

配置类：

    @Bean
    public RabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        return factory;
    }
}

消息处理类：

@Component
public class MesageAdaptor {
    @RabbitListener(queues={"q1"})
    public void handleMessage(byte bytes[]){
        System.out.println("----------------handleMessage消息(bytes[]):"+new String(bytes)+"-------");
    }

    /**
     * 接受String类型
     * @param bytes
     */
    @RabbitListener(queues = {"q1"})
    public void sendString(String bytes){
        System.out.println("----------------handleMessage(String):"+new String(bytes)+"-------");
    }
}

/**
* 与适配器不同的是可以接受参数为Message
* @param message
*/
@RabbitListener(queues = {"q1"})
public void handleMessage(Message message){
System.out.println(message.getMessageProperties());
System.out.println("----------handleMessage消息:"+new String(message.getBody()));
}

可以在接受的消息前打上注解@Payload说明他是是个消息，也可以打上注解@Headers Map<String,Object> map来接受消息头。

②：进阶用法

使用读取配置文件的方式监听队列

配置文件(queueName.properties)：

test.QueueName=q1

消息处理类：

@Component
public class MesageAdaptor {

    /**
     * 与适配器不同的是可以接受参数为Message
     * @param message
     */
    @RabbitListener(queues = {"${test.QueueName}"})
    public void handleMessage(Message message){
        System.out.println(message.getMessageProperties());
        System.out.println("----------handleMessage消息:"+new String(message.getBody()));
    }

}

启动类：

@EnableRabbit
@ComponentScan
@PropertySource("classpath:queueName.properties")
public class Start {
    public static void main(String[] args) {
        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Start.class);
    }
}

使用bindings实现自动声明

配置类：

@Configuration
public class MQconfig {
    @Bean
    public ConnectionFactory connectionFactory(){
        CachingConnectionFactory cachingConnectionFactory=new CachingConnectionFactory();
        cachingConnectionFactory.setUri("amqp://guest:[email protected]:5672");
        cachingConnectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CHANNEL);
        return cachingConnectionFactory;
    }

    /**
     * 使用自动申明的条件就是要有Rabbitadmin
     * @param connectionFactory
     * @return
     */
    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
//        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }
    @Bean
    public RabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        return factory;
    }
}

消息处理类：

@Component
public class MesageAdaptor {

/**
     * 自动申明与监听
     * @param message
     */
    @RabbitListener(bindings =@QueueBinding(value = @Queue(value = "listenerQueue",durable = "true"), exchange = @Exchange(value = "listenerExchange",durable = "true",type = ExchangeTypes.TOPIC),key = "test.#"))
    public void declare(Message message){
        System.out.println(message.getMessageProperties());
        System.out.println("----------handleMessage消息:"+new String(message.getBody()));
    }
}

启动类：

@EnableRabbit
@ComponentScan
public class Start {
    public static void main(String[] args) throws Exception{
        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Start.class);

    }
}

@RabbitListenner写在类上面

消息处理类

@Component
@RabbitListener(queues={"q1"})
public class MesageAdaptor {

@RabbitHandler
    public void handleMessage(@Payload byte bytes[]){
        System.out.println("----------------handleMessage消息(bytes[]):"+new String(bytes)+"-------");
    }

    @RabbitHandler
    public void handleMessage(String bytes){
        System.out.println("----------------handleMessage(String):"+new String(bytes)+"-------");
    }

注：写在类上@RabbitListenner注解要配合@RabbitHandler一起使用

@RabbitListenner注解设置containerfactory

配置类:

@Configuration
public class MQconfig {
    @Bean
    public ConnectionFactory connectionFactory(){
        CachingConnectionFactory cachingConnectionFactory=new CachingConnectionFactory();
        cachingConnectionFactory.setUri("amqp://guest:[email protected]:5672");
        cachingConnectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CHANNEL);
        return cachingConnectionFactory;
    }
       @Bean
    public RabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        return factory;
    }
    @Bean
    public RabbitListenerContainerFactory rabbitListenerContainerFactory2(ConnectionFactory connectionFactory){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        //设置消息转换器
        factory.setMessageConverter(new MessageConverter() {
            public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
                return null;
            }
            //把message对象装换为user对象
            public Object fromMessage(Message message) throws MessageConversionException {
                return new User(1,new String(message.getBody()));
            }
        });
        return factory;
    }
}

消息处理类：

@Component
@RabbitListener(queues={"q1"},containerFactory="rabbitListenerContainerFactory2")
public class MesageAdaptor {

    @RabbitHandler
    public void handleMessage(@Payload byte bytes[]){
        System.out.println("----------------handleMessage消息(bytes[]):"+new String(bytes)+"-------");
    }

    @RabbitHandler
    public void handleMessage(String bytes){
        System.out.println("----------------handleMessage(String):"+new String(bytes)+"-------");
    }

    @RabbitHandler
    public void xxxxx(User u){
        System.out.println(u.toString());
    }
}

注：@RabbitListenner里默认的containerfactory是rabbitListenerContainerFactory

(名字必须是这个)如果有特殊需求就可以修改这个名字但是在对应的配置文件里必须有对应名字的containerfactory.

十一·消息确认(可靠消息)

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

1丶Publisher Confirms(发送者确认消息)

Publisher Confirms机制用于解决生产者与RabbitMQ服务器之间消息可靠传输。它在消息服务器持久化消息后通知消息生产者发送成功

生产者：

public class ConfirmSend {
    static Long id = 0L;

    static TreeSet<Long> tags = new TreeSet<Long>();

    private static Long send(Channel channel, byte[] body) throws Exception {
        //设置消息属性
        BasicProperties props = new BasicProperties.Builder().deliveryMode(2).contentEncoding("UTF-8").build();
        //发送消息
        channel.basicPublish("", "q1", props, body);
        return ++id;
    }

    public static void main( String[] args ) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUri("amqp://guest:[email protected]:5672");

        Connection conn = factory.newConnection();

        Channel channel = conn.createChannel();

        //使当前Channel处于确认模式（处于事务模式的Channel，不能设置为确认模式）
        channel.confirmSelect();

        channel.addConfirmListener(new ConfirmListener() {
            /**
             * deliveryTag 消息ID
             * multiple 是否批量
             *   如果是true，就意味着，小于等于deliveryTag的消息都处理成功了
             *   如果是false，就意味着，只是成功了deliveryTag这一条消息
             */
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                //消息处理成功触发该方法
                System.out.println("+++++++++++++消息处理成功++++++++++");
                System.out.println("deliveryTag: " + deliveryTag);
                System.out.println("   multiple: " + multiple);
                try {
                    //处理成功发送的消息
                    if(multiple) {//是否批量确认
                        //批量确认
                        for(Long _id : new TreeSet<Long>(tags.headSet(deliveryTag+1))) {
                            tags.remove(_id);
                        }
                    } else {
                        //单个确认
                        tags.remove(deliveryTag);
                    }
                }catch(Exception e) {
                    e.printStackTrace();
                }
                System.out.println("未确认的：" + tags);
            }

            /**
             * deliveryTag 消息ID
             * multiple 是否批量
             *   如果是true，就意味着，小于等于deliveryTag的消息都处理失败了
             *   如果是false，就意味着，只是失败了deliveryTag这一条消息
             */
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                //消息处理失败触发该方法
                System.out.println("+++++++++++++失败了++++++++++");
                System.out.println("deliveryTag: " + deliveryTag);
                System.out.println("   multiple: " + multiple);
                //处理发送失败的消息
            }
        });

        Long id = send(channel, "123".getBytes());
        //ID是需要入库或者保存的
        tags.add(id);

        id = send(channel, "123".getBytes());
        //ID是需要入库或者保存的
        tags.add(id);

        id = send(channel, "123".getBytes());
        //ID是需要入库或者保存的
        tags.add(id);

        channel.waitForConfirmsOrDie();

        TimeUnit.SECONDS.sleep(2);

        channel.close();
        conn.close();
        System.out.println(tags.toString());
    }
}

消费者：

public class Consume {

    public static void main( String[] args ) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setVirtualHost("/");

        Map<String, Object> clientProperties = new HashMap<String, Object>();
        clientProperties.put("desc", "移动支付系统V2.0");
        clientProperties.put("author", "zhangsan");
        clientProperties.put("user", "[email protected]");

        factory.setClientProperties(clientProperties);

//     Connection conn = factory.newConnection();
        Connection conn = factory.newConnection("q1队列的消费者");

        Channel channel = conn.createChannel(10);

        String consumerTag = channel.basicConsume("q1", true, "pay_all_log_consumer", new SimpleConsumer(channel));
        System.out.println(consumerTag);

        TimeUnit.SECONDS.sleep(120);

        channel.close();
        conn.close();
    }
}

回调函数：

2丶springAMQP发送可靠的消息

配置类：

@Configuration
public class MQConfig {
    @Bean
    public ConnectionFactory connectionFactory(){
        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
        cachingConnectionFactory.setUri("amqp://guest:[email protected]:5672");
        cachingConnectionFactory.setPublisherConfirms(true);
        return cachingConnectionFactory;
    }
    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        //设置确认后的回调函数
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
             * correlationData唯一标识
             * 有了这个唯一标识，我们就知道可以确认（失败）哪一条消息了
             */
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println("-------------收到消息确认啦-----------");
                if(ack) {
                    System.out.println(correlationData.getId() + " 已经成功发送");
                } else {
                    System.out.println(correlationData.getId() + " 发送失败，原因为：" + cause);
                }
                System.out.println(correlationData.toString());
            }
        });
        return rabbitTemplate;
    }
}

启动类：

@ComponentScan
public class App {
    private static Order createOrder() {
        Order order = new Order();
        order.setUserId(1);
        order.setCreateDate(LocalDateTime.now().toString());
        order.setPrice(100d);
        return order;
    }

    private static void save(Order order) {
        //入库操作
        order.setId(UUID.randomUUID().toString());
    }

    public static void main(String[] args) throws Exception {
        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(App.class);
        RabbitTemplate rabbit = context.getBean(RabbitTemplate.class);

        Order order = createOrder();
        save(order);

        ObjectMapper mapper = new ObjectMapper();
        byte[] body = mapper.writeValueAsBytes(order);

        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setContentType("json");

        Message message = new Message(body, messageProperties);

        System.out.println("id : " + order.getId());
        rabbit.send("", "q1", message, new CorrelationData(order.getId()));

        TimeUnit.SECONDS.sleep(2);
        context.close();
    }
}

消息类：

public class Order {
    private String id;
    private Integer userId;
    private double price;
    private String createDate;
    public String getId() {
        return id;
    }
    public void setId(String id) {
        this.id = id;
    }
    public Integer getUserId() {
        return userId;
    }
    public void setUserId(Integer userId) {
        this.userId = userId;
    }
    public double getPrice() {
        return price;
    }
    public void setPrice(double price) {
        this.price = price;
    }
    public String getCreateDate() {
        return createDate;
    }
    public void setCreateDate(String createDate) {
        this.createDate = createDate;
    }

    @Override
    public String toString() {
        return "Order{" +
                "id='" + id + '\'' +
                ", userId=" + userId +
                ", price=" + price +
                ", createDate='" + createDate + '\'' +
                '}';
    }
}

注:

1：CachingConnectionFactory.setPublisherConfirms(true);

2：RabbitTemplate.setConfirmCallback() 设置回调方法

3：发送消息的时候，需要指定CorrelationData，用于标识该次发送的唯一ID

比较：

1：Spring AMQP不支持批量确认，底层的RabbitMQ Java client方式支持批量

2：Spring AMQP提供的方式非常简单明了

3丶Consumer Acknowledgements(消费者确认消息)

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

消费者：

public class Consume {

    public static void main( String[] args ) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setVirtualHost("/");

        Map<String, Object> clientProperties = new HashMap<String, Object>();
        clientProperties.put("desc", "移动支付系统V2.0");
        clientProperties.put("author", "zhangsan");
        clientProperties.put("user", "[email protected]");

        factory.setClientProperties(clientProperties);

//     Connection conn = factory.newConnection();
        Connection conn = factory.newConnection("q1队列的消费者");
        Channel channel = conn.createChannel(10);
        //把自动确认消息(autoAck)设置为false
        String consumerTag = channel.basicConsume("q1", false, "pay_all_log_consumer", new SimpleConsumer(channel));
        System.out.println(consumerTag);

        TimeUnit.SECONDS.sleep(120);

        channel.close();
        conn.close();
    }
}

消息处理回调函数：

public class SimpleConsumer extends DefaultConsumer {

    public SimpleConsumer(Channel channel) {
        super(channel);
    }

    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
            throws IOException {
        if(properties.getHeaders().get("type")!=null){
            //拒绝消息(最后一个参数:是否把消息放回队列)支持批量拒绝
            this.getChannel().basicNack(envelope.getDeliveryTag(),false,false);

            //只支持拒绝单条消息
            this.getChannel().basicReject(envelope.getDeliveryTag(), false);

            System.out.println("消息属性为：" + properties);
            System.out.println("拒绝消息了");

            System.out.println(new String(body));
        }else{
            //确认消息第一个参数：消息id号，第二个参数：是否批量确认
            this.getChannel().basicAck(envelope.getDeliveryTag(),false);
            System.out.println("----------收到消息啦----------");
            System.out.println("消息属性为：" + properties);
            System.out.println("消息内容为：" + new String(body));
        }

    }
}

4丶springAMQP消费者确认消息

配置类：

@Configuration
public class MQConfig {
    @Bean
    public ConnectionFactory connectionFactory(){
        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
        cachingConnectionFactory.setUri("amqp://guest:[email protected]:5672");
        return cachingConnectionFactory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        return rabbitTemplate;
    }

    /**
     * 消息转换器
     * @param connectionFactory
     * @return
     */
    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer(ConnectionFactory connectionFactory){
        SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer(connectionFactory);
        simpleMessageListenerContainer.setQueueNames("q1");
        //设置手动确认消息
        simpleMessageListenerContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        simpleMessageListenerContainer.setMessageListener(new ChannelAwareMessageListener() {
            public void onMessage(Message message, Channel channel) throws Exception {
                Object type = message.getMessageProperties().getHeaders().get("type");
                if(type==null){
                    //参数1:消息id,参数2：是否把消息返回队列
         channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
                    //参数1：消息id，参数2:是否批量拒绝，参数3：是否把消息返回队列
//                    channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);
                    System.out.println("消息拒绝了");
                }else{
                    System.out.println("消息已确认");
                    //参数1：消息id,参数2：是否批量确认
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                }
                System.out.println(new String(message.getBody()));
                System.out.println(message.getMessageProperties());
            }
        });
        return simpleMessageListenerContainer;
    }
}

启动类：

/**
* AcknowledgeMode.NONE 自动确认，等效于 autoAck=true
*
* AcknowledgeMode.MANUAL 手动确认，等效于 autoAck=false
*
*/

@ComponentScan
public class App {
    public static void main(String[] args) throws Exception {
        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(App.class);
    }
}

注：AcknowledgeMode.NONE 自动确认，等效于 autoAck=true

AcknowledgeMode.MANUAL 手动确认，等效于 autoAck=false
使用spring默认的AcknowledgeMode.AUTO

配置类：

@Configuration
public class MQConfig {
    @Bean
    public ConnectionFactory connectionFactory(){
        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
        cachingConnectionFactory.setUri("amqp://guest:[email protected]:5672");
        return cachingConnectionFactory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        return rabbitTemplate;
    }

    /**
     * 消息转换器
     * @param connectionFactory
     * @return
     */
    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer(ConnectionFactory connectionFactory){
        SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer(connectionFactory);
        simpleMessageListenerContainer.setQueueNames("q1");
        //设置拒绝消息切消息不会被放回队列
        simpleMessageListenerContainer.setDefaultRequeueRejected(false);

        simpleMessageListenerContainer.setMessageListener(new ChannelAwareMessageListener() {
            public void onMessage(Message message, Channel channel) throws Exception {
                System.out.println(new String(message.getBody()));
                System.out.println(message.getMessageProperties());
                //消息会被确认
//                throw new ImmediateAcknowledgeAmqpException("consume fail");

                //消息会被拒绝，且requeue=false
//                throw new AmqpRejectAndDontRequeueException("consume fail");

                //其他异常消息会被拒绝，且requeue=true
                //requeue=true可以通过setDefaultRequeueRejected（默认为true）去设置
                throw new Exception("");

            }
        });
        return simpleMessageListenerContainer;
    }
}

启动类：

/**
* AcknowledgeMode.NONE 自动确认，等效于 autoAck=true
*
* AcknowledgeMode.MANUAL 手动确认，等效于 autoAck=false
*
* AcknowledgeMode.AUTO 根据方法的执行情况来决定是否是确认还是拒绝（是否重新入queue）
* 1: 如果消息成功被消费了，则自动确认
* 2：
*    (1)当抛出AmqpRejectAndDontRequeueException异常的时候，则消息会被拒绝，且requeue=false
*    (2)当抛出ImmediateAcknowledgeAmqpException异常，则消息会被确认
*    (3)其他的异常，则消息会被拒绝，且requeue=true（如果此时只有一个消费者监听该队列，则有发生死循环的风险）
*       其中，requeue=true可以通过setDefaultRequeueRejected（默认为true）去设置
*
*    要想了解更新详细的代码，则可以分析SimpleMessageListenerContainer.doReceiveAndExecute方法
*
*/
@ComponentScan
public class App {
    public static void main(String[] args) throws Exception {
        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(App.class);
    }
}

注：

当抛出AmqpRejectAndDontRequeueException异常的时候，则消息会被拒绝，且requeue=false

当抛出ImmediateAcknowledgeAmqpException异常，则消息会被确认

其他的异常，则消息会被拒绝，且requeue=true

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

private boolean doReceiveAndExecute(BlockingQueueConsumer consumer) throws Throwable { //NOSONAR
   Channel channel = consumer.getChannel();
   for (int i = 0; i < this.txSize; i++) {
      logger.trace("Waiting for message from consumer.");
      Message message = consumer.nextMessage(this.receiveTimeout);
      if (message == null) {
         break;
      }
      try {
         executeListener(channel, message);
      }
      catch (ImmediateAcknowledgeAmqpException e) {
         if (this.logger.isDebugEnabled()) {
            this.logger.debug("User requested ack for failed delivery: "
                  + message.getMessageProperties().getDeliveryTag());
         }
         break;
      }
      catch (Throwable ex) { //NOSONAR
         if (causeChainHasImmediateAcknowledgeAmqpException(ex)) {
            if (this.logger.isDebugEnabled()) {
               this.logger.debug("User requested ack for failed delivery: "
                     + message.getMessageProperties().getDeliveryTag());
            }
            break;
         }
         if (this.transactionManager != null) {
            if (this.transactionAttribute.rollbackOn(ex)) {
               RabbitResourceHolder resourceHolder = (RabbitResourceHolder) TransactionSynchronizationManager
                     .getResource(getConnectionFactory());
               if (resourceHolder != null) {
                  consumer.clearDeliveryTags();
               }
               else {
                  /*
                   * If we don't actually have a transaction, we have to roll back
                   * manually. See prepareHolderForRollback().
                   */
                  consumer.rollbackOnExceptionIfNecessary(ex);
               }
               throw ex; // encompassing transaction will handle the rollback.
            }
            else {
               if (this.logger.isDebugEnabled()) {
                  this.logger.debug("No rollback for " + ex);
               }
               break;
            }
         }
         else {
            consumer.rollbackOnExceptionIfNecessary(ex);
            throw ex;
         }
      }
   }
   return consumer.commitIfNecessary(isChannelLocallyTransacted(channel));
}

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

消费者拒绝消息并不代表消息没被确认所以生产的ack还是会为true

（把生产者当做卖家，消费者当做买家，rabbitmq当做快递公司，卖家拒绝收货卖家不能怪快递公司没发货，说明卖家还是受到货物了的）

十二·alternate exchange

当我们交换机不能通过路由键路由到指定的消息队列上去的时候，就可以把消息转发到alternate exchange设置的exhcange上面去，保证消息不会被丢失。

配置类：

@Configuration
public class MQConifg {
    @Bean
    public ConnectionFactory connectionFactory(){
        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
        cachingConnectionFactory.setUri("amqp://guest:[email protected]:5672");
        //设置消息回复
        cachingConnectionFactory.setPublisherConfirms(true);
        return cachingConnectionFactory;
    }
    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        rabbitAdmin.setAutoStartup(true);
//        binding
        rabbitAdmin.declareBinding(new Binding("alternate-queue", Binding.DestinationType.QUEUE,
                "warn","test",new HashMap<String,Object>()));
        rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue("alternate-queue")).to
                (new FanoutExchange("debug",true,false)));

        return rabbitAdmin;
    }

    @Bean
    public Exchange exchange(){
        HashMap<String, Object> arguments = new HashMap<String, Object>();
        arguments.put("alternate-exchange","debug");
        return new TopicExchange("warn",true,false,arguments);
    }
    @Bean
    public Queue queue(){
       return new Queue("alternate-queue");
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

        //消息确认后的回调函数
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback(){
            public void confirm(CorrelationData correlationData, boolean ack, String cause){
                if(ack){
                    System.out.println("生产者消息发送成功");
                }else{
                    System.out.println("消息发送失败");
                    System.out.println(cause);
                }
            }
        });
        return rabbitTemplate;
    }

    /**
     * 消息转换器
     * @param connectionFactory
     * @return
     */
    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer(ConnectionFactory connectionFactory){
        SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer(connectionFactory);
        simpleMessageListenerContainer.setQueueNames("alternate-queue");
        simpleMessageListenerContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL);

        simpleMessageListenerContainer.setMessageListener(new ChannelAwareMessageListener(){
            public void onMessage(Message message, Channel channel) throws Exception{
                try {
                    System.out.println("消费者收到消息了");
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                    System.out.println("消息："+new String(message.getBody()));
                    System.out.println("消息id："+message.getMessageProperties().getDeliveryTag());

                }catch (IOException e){
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);
                    channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
                    System.out.println("出现错误返回消息到队列");
                }

            }
        });
        return simpleMessageListenerContainer;
    }
}

启动类：

@ComponentScan
public class App {
    public static void main(String[] args) {
        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(App.class);
        RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);
        //发送消息(warn根据xxx不能路由消息就向alternate指定的excahnge发)
        rabbitTemplate.send("warn","xxx",
                new Message("测试alternate".getBytes(),new MessageProperties()));
    }
}

注：错误demo,不要把生产者和消费者放在一个服务端。要想去启动这个demo需要注释bingding和send。然后再启动打开binding，然后再启动打开send.

注：根据视频资料以及网上零散资料的整合，仅供参考。

RabbitMQ详解以及spring对RabbitMQ的集成(附带部分源码解读)

一·简介

1丶为什么要使用消息队列

2丶为什么使用RabbitMQ

3丶为什么使用Spring AMQP

二·AMQP协议介绍

三·RabbitMQ安装

1丶特点

2丶Windows下安装

①安装Erlang

②安装RabbitMQ

③安装插件

四·exchange详解

1丶exchange介绍

2丶Direct Exchange

3丶topic Exchage

4丶fanout Exchang

5丶Headers Exchaneg

五·binding详解

六·queue详解

七·message详解

八·rabbitmq java client

1丶基本操作

①：创建工厂以及连接和信道

②：创建交换机

③：创建消息队列

④：binding

⑤：关闭连接释放资源

2丶消息的发送与消费

①：使用默认的exchange发送消息(default Exchange)

②：向topic exchange发送消息

③：消费者消费消息

九·Spring AMQP

1丶简介

2丶RabbitAdmin创建详解

①：使用

②：exchange,binding,queue的自动声明

3丶RabbitTemplate进行消息发送

4丶消息的消费

①：使用容器的方式进行创建消费者消费

5丶MessageListenerAdaptor

①：基本操作

②：MessageConvertor

③：Jackson2MessageConverter

④：ContentTypeDelegatingMessageConverter

6丶RabbitListenerConfigurer进行消费

7丶使用@rabbitlistener注解进行消费

①：实现步骤

②：进阶用法

十一·消息确认(可靠消息)

1丶Publisher Confirms(发送者确认消息)

2丶springAMQP发送可靠的消息

3丶Consumer Acknowledgements(消费者确认消息)

4丶springAMQP消费者确认消息

十二·alternate exchange

相关推荐