https://blog.csdn.net/qq_35387940/article/details/100514134
https://github.com/wangjin123456/2020/tree/master/springboot/rabbitmq

在安装完rabbitmq后，输入：http://localhost:15672/，是可以看到rbbitmq界面的；

image.png

在这个界面里面我们乐意做点什么呢？
可以手动创建虚拟机host,创建用户，分配权限，创建队列等，还有查看队列消息，消费效率，推送效率等。
首先先介绍下一个简单的一个消息推送到接收的流程，提供一个简单的图：

image

黄色的圈圈就是我们的消息推送服务，将消息推送到中间方框里面也就是rabbitMq发服务器，然后经过服务器里面的交换机，队列等各种消息将数据处理入列后，最终右边的消费者将获取对应监听的消息。
常用的交换机有以下三种，因为消费者是从队列获取信息的，对列是绑定交换机的（），所以对应的消息推送/接收模式也会有以下三种：
直连型交换机，根据消息携带的路由键将消息投递给对应的队列；

大致流程，有一个队列绑定到一个直连交换机上，同时赋予了一个路由键routing key .然后当一个消息携带着路由值为X,这个消息通过生产者发送给交换机时交换机就根据这个路由值X去寻找绑定值也是X的队列
Fanout Exchange
扇形交换机，这个交换机没有路由键的概念，就算你绑定了路由键也是无视的。这个交换机在接到消息后：会直接转发到绑定到他上面的所有队列。

Topic Exchange
主题交换机，这个交换机其实跟直连交换机流程差不多，但是他的特点是在他的路由键和绑定建之间是有规则的，简单的介绍下规则

（星号）用来表示一个单词（必须出现的）

（并号）用来表示任意数量（零个或者多个）单词

统配的绑定建是跟队列进行绑定的，举个小例子：
队列Q1 绑定键为：.TT. 队列Q2绑定键为 TT.#
如果一条消息携带的路由键为 A.TT.B，那么队列Q1将会收到；
如果一条消息携带的路由键为TT.AA.BB，那么队列Q2将会收到；
主题交换机时非常强大的，为什么这么膨胀？

当一个队列绑定键为 # 的时候，这个队列将会无视消息的路由键，接收
所有的消息。
当* 和 # 这2个特殊字符都未在绑定建中出现的时候，此时主题交换机就拥有了直连交换机的功能，
所以主题交换机也就实现了扇形交换机和直连交换机的功能。

另外还有Header Exchange 头交换机，default Exchange 默认交换机， Dead Letter Exchange死信交换机
好了，简单的介绍到此为止，接下来进行编码：
本次实例教程需要创建2个 springboot 项目，一个rabbitmq-provider（生产者），consumber 消费者。

首先创建rabbit-provider,
pom.xml 里用到的jar依赖：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.5.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.taotao</groupId>
    <artifactId>rabbitmq-provider</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>rabbitmq-provider</name>
    <description>Demo project for Spring Boot</description>

    <properties>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <!--rabbitmq-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

然后application.yml

server:
  port:  8021
spring:
   application:
      name: rabbitmq-provider
   rabbitmq:
     host: 127.0.0.1
     port:  5672
     username: sy
     password: aaaa1111
     virtual-host:  /yingying_rabbitmq  #这里要去rabbitmq里面创建 /yingying_rabbitmq   并且授权，

接着我们先用下direct exchange(直连型交换机)，创建DirectRabbitConfig.java :


package com.taotao.rabbitmqprovider.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/*
* 使用direct exchange(直连型交换机)
* @autor tom
* @date 2020/3/19 0019 10:46
*/@Configuration
public class DirectRabbitConfig {

 //队列 起名：TestDirectQueue
 @Bean
 public Queue TestDirectQueue() {
     return new Queue("TestDirectQueue",true);  //true 是否持久
 }

 //Direct交换机 起名：TestDirectExchange
 @Bean
 DirectExchange TestDirectExchange() {
     return new DirectExchange("TestDirectExchange");
 }

 //绑定  将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
 @Bean
 Binding bindingDirect() {
     return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
 }



}

写个简单的接口进行消息推送（根据需求也可以改为定时任务等等，具体看需求），
SendMessageController.java：


package com.taotao.rabbitmqprovider.controller;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

/**
 * 接口进行消息推送
 * @autor tom
 * @date 2020/3/19 0019 11:01
 */
@RestController
public class SendMessageController {
    //使用RabbitTemplate 这里提供了接收/发送等方法
    @Autowired
    RabbitTemplate rabbitTemplate;
    @GetMapping("/sendDirectMessage")
    public String sendDirectMessage() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "test message  hello!";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        //将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
        rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", map);
        return "ok";

    }
}

访问 http://localhost:8021/sendDirectMessage

image.png

因为目前没有弄消费者rabbitmq-consumer，消息没有被消费，我们去rabbitmq
管理页面看，是否推送成功。

image.png

再看看队列

image.png

很好，消息已经推送到rabbitmq服务器了
接下来创建 rabbitmq-consumer项目：
pom.xml依赖：


<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.5.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.taotao</groupId>
    <artifactId>rabbitmq-consumer</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>rabbitmq-consumer</name>
    <description>Demo project for Spring Boot</description>

    <properties>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

然后是 yml

server:
  port: 8022
spring:
   application:
      name: rabbitmq-consumer
   rabbitmq:
      host: 127.0.0.1
      port: 5672
      username:  sy
      password:  aaaa1111
      virtual-host:  /yingying_rabbitmq

然后一样，创建 DirectRabbitConfig.java


package com.taotao.rabbitmqconsumer.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @autor tom
 * @date 2020/3/19 0019 15:18
 */
@Configuration
public class DirectRabbitConfig {
    //队列 起名：TestDirectQueue
    @Bean
    public Queue TestDirectQueue() {
        return new Queue("TestDirectQueue", true);
    }

    //Direct交换机 起名：TestDirectExchange
    @Bean
    DirectExchange TestDirectExchange() {
        return new DirectExchange("TestDirectExchange");
    }

    //绑定  将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
    @Bean
    Binding bindingDirect() {
        return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
    }
}

然后创建消息接收监听类, DirectReceiver.java


package com.taotao.rabbitmqconsumer.config;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * 创建消息接收监听类
 * @autor tom
 * @date 2020/3/19 0019 15:29
 */
@Component
@RabbitListener(queues = "TestDirectQueue")//监听的队列消息
public class DirectReceiver {
    @RabbitHandler
    public  void process(Map testMessage){
        System.out.println("DirectReceiver消费者收到消息："+ testMessage.toString() );
    }
}

然后将rabbitmq-consumer项目运行起来，可以看到把之前推送的那条消息消费下来了：
然后可以再继续调用rabbitmq-provider项目的推送消息接口，可以看到消费者即时消费消息：

image.png

接着，我们使用Topic Exchange 主题交换机

在rabbitmq-provider 项目里创建TopicRabbitConfig.java

package com.taotao.rabbitmqprovider.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.management.MXBean;


/**
 * 主题交换机
 * @autor tom
 * @date 2020/3/19 0019 16:25
 */
@Configuration
public class TopicRabbitConfig {
    //绑定建
    public  final static String man="topic.man";
    public final static  String woman="topic.woman";
    @Bean
    public Queue firstQueue(){
        return  new Queue(TopicRabbitConfig.man);
    }

    @Bean
    public  Queue secondQueue(){
        return  new Queue(TopicRabbitConfig.woman);
    }

    @Bean
    TopicExchange exchange(){
        return  new TopicExchange("topicExchange");
    }
    /**
     * 将firstQueue 和 topicExchange绑定，而且绑定的键值为 topic.man
     *z这样只要是消息携带的路由键是topic.man，才会分发到该队列
     */
    @Bean
    Binding bindingExchangeMessage(){
        return  BindingBuilder.bind(firstQueue()).to(exchange()).with(man);
    }
    /**
     * 将SecondQueue 和 topicExchange绑定，而且绑定的键值为 topic.#
     *z这样只要是消息携带的路由键是topic.#，才会分发到该队列
     */
    @Bean
    Binding bindingExchangeMessage2(){
        return  BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");
    }
}

2，然后再加多2个接口，同于推送消息到主题交换机：

package com.taotao.rabbitmqprovider.controller;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

/**
 * 接口进行消息推送
 * @autor tom
 * @date 2020/3/19 0019 11:01
 */
@RestController
public class SendMessageController {
    //使用RabbitTemplate 这里提供了接收/发送等方法
    @Autowired
    RabbitTemplate rabbitTemplate;
    @GetMapping("/sendDirectMessage")
    public String sendDirectMessage() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "test message, hello!";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        //将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
        rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", map);
        return "ok";

    }


    @GetMapping("/sendTopicMessage1")
    public String sendTopicMessage1(){
        String messageId=String.valueOf(UUID.randomUUID());
        String messageData ="message: M A N";
        String createTime=LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String,Object> manMap=new HashMap<>();
        manMap.put("messageId",messageId);
        manMap.put("messageData",messageData);
        manMap.put("createTime",createTime);//topic.man
        rabbitTemplate.convertAndSend("topicExchange","topic.man",manMap);
      return  "ok";
    }


    @GetMapping("/sendTopicMessage2")
    public String sendTopicMessage2(){
        String messageId=String.valueOf(UUID.randomUUID());
        String messageData ="message:  woman is all";
        String createTime=LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String,Object> manMap=new HashMap<>();
        manMap.put("messageId",messageId);
        manMap.put("messageData",messageData);
        manMap.put("createTime",createTime);
        rabbitTemplate.convertAndSend("topicExchange","topic.woman",manMap);
        return  "ok";
    }

}

生产者这边已经完事，先不着急运行，在rabbit-consumer项目上，创建TopicManReceiver.java

package com.taotao.rabbitmqconsumer.config;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * @autor tom
 * @date 2020/3/19 0019 18:29
 */

@Component
@RabbitListener(queues = "topic.man")
public class TopicManReceiver {
    @RabbitHandler
    public void process(Map testMessage){
        System.out.println("TopicManReceiver消费者收到消息："+testMessage.toString());
    }
}

在创建一个TopicTotalReceiver.java

package com.taotao.rabbitmqconsumer.config;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * @autor tom
 * @date 2020/3/19 0019 18:32
 */

@Component
@RabbitListener(queues = "topic.woman")
public class TopicTotalReceiver {

    @RabbitHandler
    public void process(Map testMessage){
        System.out.println("TopicTotalReceiver消费者收到的消息:"+ testMessage.toString());
    }
}

同样加上主题交换机的相关配置， TopicRabbitConfig.java


package com.taotao.rabbitmqconsumer.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


/**
 * 主题交换机
 * @autor tom
 * @date 2020/3/19 0019 16:25
 */
@Configuration
public class TopicRabbitConfig {
    //绑定建
    public  final static String man="topic.man";
    public final static  String woman="topic.woman";
    @Bean
    public Queue firstQueue(){
        return  new Queue(TopicRabbitConfig.man);
    }

    @Bean
    public  Queue secondQueue(){
        return  new Queue(TopicRabbitConfig.woman);
    }

    @Bean
    TopicExchange exchange(){
        return  new TopicExchange("topicExchange");
    }
    /**
     * 将firstQueue 和 topicExchange绑定，而且绑定的键值为 topic.man
     *z这样只要是消息携带的路由键是topic.man，才会分发到该队列
     */
    @Bean
    Binding bindingExchangeMessage(){
        return  BindingBuilder.bind(firstQueue()).to(exchange()).with(man);
    }
    /**
     * 将SecondQueue 和 topicExchange绑定，而且绑定的键值为 topic.#
     *z这样只要是消息携带的路由键是topic.#，才会分发到该队列
     */
    @Bean
    Binding bindingExchangeMessage2(){
        return  BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");
    }
}

然后将rabbitmq-provider, rabbit-cosumer2个项目跑起来，先调用/sendTopicMessage1接口：

image

然后看消费者rabbitmq-consumer的控制台输出情况：
TopicManReceiver 监听队列1，绑定建为 topic.man
TopicTotallReceiver 监听队列2，绑定键为： topic.#
而当前推送的消息，携带的路由键为: topic.man
所以可以看到两个监听消费者receiver都成功消费到了消息，因为这两个recevier监听的队列的绑定键都能与这条消息携带的路由键匹配上。

image.png

接下来调用接口/sendTopicMessage2:

image

然后看消费者rabbitmq-consumer的控制台输出情况：
TopicManReceiver监听队列1，绑定键为：topic.man
TopicTotalReceiver监听队列2，绑定键为：topic.#
而当前推送的消息，携带的路由键为：topic.woman

所以可以看到两个监听消费者只有TopicTotalReceiver成功消费到了消息。

image.png

接下来是使用Fanout Exchang 扇形交换机。
同样地，先在rabbitmq-provider项目上创建FanoutRabbitConfig.java:



package com.taotao.rabbitmqprovider.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;



/**
 * Famout Exchange 扇形交换机
 * @autor tom
 * @date 2020/3/19 0019 19:13
 */
@Configuration
public class FanoutRabbitConfig {

    /***
     * 创建三个队列： famout.A  famout.B famout.c
     * 将三个队列都绑定在交换机 famoutExchange 上
     * 因为是扇形交换机，路由键无需配置，配置也不起作用
     */

    @Bean
    public Queue queueA(){
        return  new Queue("fanout.A");
    }
    @Bean
    public Queue queueB(){
        return  new Queue("fanout.B");
    }
    @Bean
    public Queue queueC(){
        return  new Queue("fanout.C");
    }

    @Bean
    FanoutExchange fanoutExchange(){
        return  new FanoutExchange("fanoutExchange");
    }

    @Bean
    Binding bindingExchangeA(){
        return BindingBuilder.bind(queueA()).to(fanoutExchange());
    }
    @Bean
    Binding bindingExchangeB(){
        return BindingBuilder.bind(queueB()).to(fanoutExchange());
    }
    @Bean
    Binding bindingExchangeC(){
        return BindingBuilder.bind(queueC()).to(fanoutExchange());
    }
}

然后填写一个接口用于推送消息;

package com.taotao.rabbitmqprovider.controller;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

/**
 * 接口进行消息推送
 * @autor tom
 * @date 2020/3/19 0019 11:01
 */
@RestController
public class SendMessageController {
    //使用RabbitTemplate 这里提供了接收/发送等方法
    @Autowired
    RabbitTemplate rabbitTemplate;
    @GetMapping("/sendDirectMessage")
    public String sendDirectMessage() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "test message, hello!";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        //将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
        rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", map);
        return "ok";

    }


    @GetMapping("/sendTopicMessage1")
    public String sendTopicMessage1(){
        String messageId=String.valueOf(UUID.randomUUID());
        String messageData ="message: M A N";
        String createTime=LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String,Object> manMap=new HashMap<>();
        manMap.put("messageId",messageId);
        manMap.put("messageData",messageData);
        manMap.put("createTime",createTime);//topic.man
        rabbitTemplate.convertAndSend("topicExchange","topic.man",manMap);
      return  "ok";
    }


    @GetMapping("/sendTopicMessage2")
    public String sendTopicMessage2(){
        String messageId=String.valueOf(UUID.randomUUID());
        String messageData ="message:  woman is all";
        String createTime=LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String,Object> manMap=new HashMap<>();
        manMap.put("messageId",messageId);
        manMap.put("messageData",messageData);
        manMap.put("createTime",createTime);
        rabbitTemplate.convertAndSend("topicExchange","topic.woman",manMap);
        return  "ok";
    }

    @GetMapping("/sendTopicMessage3")
    public String sendTopicMessage3(){
        String messageId=String.valueOf(UUID.randomUUID());
        String messageData ="message:  woman is all";
        String createTime=LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String,Object> manMap=new HashMap<>();
        manMap.put("messageId",messageId);
        manMap.put("messageData",messageData);
        manMap.put("createTime",createTime);
        rabbitTemplate.convertAndSend("fanoutExchange",null,manMap);
        return  "ok";
    }
}

接着在 rabbitmq-consumer项目里加上消息消费类

FanoutReceiverA.java


package com.taotao.rabbitmqconsumer.config;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * 加上消息消费类
 * @autor tom
 * @date 2020/3/19 0019 19:23
 */
@Component
@RabbitListener(queues = "fanout.A")
public class FanoutReceiverA {

    @RabbitHandler
    public  void process(Map testMessage){
        System.out.println("FamoutReceiverA消费者收到消息："+ testMessage.toString());
    }
}

FanoutReceiverB.java

package com.taotao.rabbitmqconsumer.config;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * 加上消息消费类
 * @autor tom
 * @date 2020/3/19 0019 19:23
 */
@Component
@RabbitListener(queues = "fanout.B")
public class FanoutReceiverB {

    @RabbitHandler
    public  void process(Map testMessage){
        System.out.println("FamoutReceiverB消费者收到消息："+ testMessage.toString());
    }
}

FanoutReceiverC.java

package com.taotao.rabbitmqconsumer.config;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * 加上消息消费类
 * @autor tom
 * @date 2020/3/19 0019 19:23
 */
@Component
@RabbitListener(queues = "fanout.C")
public class FanoutReceiverC {

    @RabbitHandler
    public  void process(Map testMessage){
        System.out.println("FamoutReceiverC消费者收到消息："+ testMessage.toString());
    }
}

然后加上扇形交换机的配置类， FanoutRabbitConfig.java

package com.taotao.rabbitmqconsumer.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


/**
 * Famout Exchange 扇形交换机
 * @autor tom
 * @date 2020/3/19 0019 19:13
 */
@Configuration
public class FanoutRabbitConfig {

    /***
     * 创建三个队列： famout.A  famout.B famout.c
     * 将三个队列都绑定在交换机 famoutExchange 上
     * 因为是扇形交换机，路由键无需配置，配置也不起作用
     */

    @Bean
    public Queue queueA(){
        return  new Queue("fanout.A");
    }
    @Bean
    public Queue queueB(){
        return  new Queue("fanout.B");
    }
    @Bean
    public Queue queueC(){
        return  new Queue("fanout.C");
    }

    @Bean
    FanoutExchange fanoutExchange(){
        return  new FanoutExchange("fanoutExchange");
    }

    @Bean
    Binding bindingExchangeA(){
        return BindingBuilder.bind(queueA()).to(fanoutExchange());
    }
    @Bean
    Binding bindingExchangeB(){
        return BindingBuilder.bind(queueB()).to(fanoutExchange());
    }
    @Bean
    Binding bindingExchangeC(){
        return BindingBuilder.bind(queueC()).to(fanoutExchange());
    }
}

最后将rabbitmq-provider和rabbitmq-consumer项目都跑起来，调用下接口/sendFanoutMessage ：

image

然后看看rabbitmq-consumer项目的控制台情况：

可以看到只要发送到 fanoutExchange 这个扇型交换机的消息，三个队列都绑定这个交换机，所以三个消息接收类都监听到了这条消息。
到了这里其实三个常用交换机的使用我们已经使用完毕了，那么接下来我们继续讲讲消息的回调，其实就是消息确认（生产者推送消息成功，消费这边接收消息成功）。

在rabbitmq-provider项目的application.yml文件上，加上消息确认的配置项后：

···
server:
port: 8021
spring:
application:
name: rabbitmq-provider
rabbitmq:
host: 127.0.0.1
port: 5672
username: sy
password: aaaa1111
virtual-host: /yingying_rabbitmq #这里要去rabbitmq里面创建 /yingying_rabbitmq 并且授权，
#消息确认配置项

 #确认消息已发送到交换机（Exchange)

 publisher-confirms: true
 publisher-returns: true #确认消息已发送到队列（）Queue

···
然后配置相关的信息确认回调函数，RabbitConfig.java

package com.taotao.rabbitmqprovider.config;


import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 配置相关的消息确认回调函数
 *
 * @autor tom
 * @date 2020/3/19 0019 21:19
 */
@Configuration
public class RabbitConfig {
    @Bean
    public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        //设置开启Mandatory, 才能触发回调函数，无论消息推送结果怎么样都强制调用回调函数，
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                System.out.println("ConfirmCallback:  " + " 相关数据:" + correlationData);
                System.out.println("ConfirmCallback:  " + "确认情况： " + b);
                System.out.println("ConfirmCallback:  " + " 原因：" + s);
            }
        });

        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int i, String s, String s1, String s2) {
                System.out.println("ReturnCallback:  " + "消息：" + message);

                System.out.println("ReturnCallback:  " + "回应码：" + i);
                System.out.println("ReturnCallback:  " + "回应消息：" + s);
                System.out.println("ReturnCallback:  " + "交换机：" + s1);
                System.out.println("ReturnCallback:  " + "路由键：" + s2);

            }
        });
        return rabbitTemplate;
    }


}

到这里，生产者推送消息的消息确认调用回调函数已完毕

可以看到上面写了2个回调函数，一个叫ConfirmCallback，一个叫做Retrunallback;
那么以上这2种回调函数都是什么情况会触发呢？
先从总体的情况分析，推送消息存在四种情况：
1，消息推送到server,但是在server里找不到交换机
2，消息推送到server,找到交换机了，但是没找到队列
3，消息推送到server,交换机和队列啥都没找到
4，消息推送成功
那么我先写几个接口来分别测试和认证下4种情况，消息确认触发回调函数情况：
1，消息推送到server，但是在server里找不到交换机
写个测试接口，把消息推送到名为 'non-existent-exchange' 的交换机上（这里没有配置的）：

  /**
     *
     * 消息推送到server ,但是在server里找不到交换机
     */

    @GetMapping("/TestMessageAck")
    public  String TestMessageAck(){
        String messageId =String.valueOf(UUID.randomUUID());
        String messageData ="message: non-existent-exchange  test message";
        String createTime=LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH-mm-ss"));
        Map<String,Object>map =new HashMap<>();
        map.put("messageId",messageId);
        map.put("messageData",messageData);
        map.put("createTime",createTime);
        rabbitTemplate.convertAndSend("non-existent-exchange","TestDirectRouting",map);

        return  "ok";

    }

调用接口，查看rabbitmq-provider项目控制台输出情况（原因里又说，existent-exchange）:

2020-03-19 22:28:08.820 ERROR 10768 --- [ 127.0.0.1:5672] o.s.a.r.c.CachingConnectionFactory       : Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'non-existent-exchange' in vhost '/yingying_rabbitmq', class-id=60, method-id=40)
ConfirmCallback:   相关数据:null
ConfirmCallback:  确认情况： false
ConfirmCallback:   原因：channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'non-existent-exchange' in vhost '/yingying_rabbitmq', class-id=60, method-id=40)

结论： 1 这种情况触发的是ConfirmCallback回调函数

2，消息推送到server, 找到交换机了，但是没有找到队列
这种情况就是需要新增一个交换机，但是不给这个交换机绑定队列，我来简单在
DirectRabbitConfig 里面新增一个交换机，名叫"lonelyDirectExchange",但是没给他做任何绑定配置操作：

  @Bean
    DirectExchange lonelyDirectExchange() {
        return new DirectExchange("lonelyDirectExchange");
    }

然后写个测试接口，把消息推送到名为 'lonelyDirectExchange'的交换机（这个交换机时没有任何队列配置的）：

  @GetMapping("/TestMessageAck2")
    public String TestMessageAck2() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: lonelyDirectExchange test message ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        rabbitTemplate.convertAndSend("lonelyDirectExchange", "TestDirectRouting", map);
        return "ok";
    }

调用接口，查看rabbitmq-provider项目的控制台输出情况：

ReturnCallback:  消息：(Body:'{createTime=2020-03-20 04:01:26, messageId=80341f93-fff8-4cf2-bd03-cea65ab9b589, messageData=message: lonelyDirectExchange test message }' MessageProperties [headers={}, contentType=application/x-java-serialized-object, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0])
ReturnCallback:  回应码：312
ReturnCallback:  回应消息：NO_ROUTE
ReturnCallback:  交换机：lonelyDirectExchange
ReturnCallback:  路由键：TestDirectRouting

ConfirmCallback:   相关数据:null
ConfirmCallback:  确认情况： true
ConfirmCallback:   原因：null

可以看到这种情况，2个函数都被调用了；
这种情况下，消息是推送成功到服务器了的，所以ConfirmCallback对消息确认情况是true;而在ReturnCalllback回调函数的打印参数里面可以看到，消息是推送到了交换机成功了，但是在路由分发给队列的时候，找不到队列，所以报了错误 No_Route
结论 2 这种情况触发的是 Confirm和 ReturnCallback 2个回调函数。
3，消息推送到server，交换机和队列啥都没找到，
这种情况其实一看觉得跟1 好像，没错，3和1的回调是一致的，所以不做结果说明了
结论： 3 这种情况触发的是ConfirmCallback 回调函数。
4 消息推送成功
那么测试下，按照正常调用之前消息推送到的接口就行，就调用下/sendFanoutMessage接口，可以看到控制台输出：

ConfirmCallback:   相关数据:null
ConfirmCallback:  确认情况： true
ConfirmCallback:   原因：null

以上是生产者推送消息的消息确认，回调函数的使用介绍（可以在回调函数根据雪球做对应的扩展或者业务数据处理）。
接下来我们继续，消费者接受到消息的消息确认机制。
和生产者的消息确认不同，因为消息接收本来就是在监听消息，符合条件的消息就会消费下来。所以，消息接收的确认机制主要存在三种模式
1，自动确认，这也是默认的消息确认情况， AcknowledgeMode.NONE
RabbitMQ成功将消息发出（即将消息成功写入 Tcp Socket）中立即认为本次投递已经被正确处理。不管消费者端是否成功处理本次投递；
所以这种情况如果消费端消费逻辑抛出异常，也就是消费端没有处理成功这条消息，那么相当于丢失了消息。
一般这种情况我们都是使用try catch 捕捉异常后，打印日志用于追踪数据，这样找出对应数据在做后续处理。

2，根据情况确认，这个不做介绍
3，手动确认，这个比较关键，也是我们配置接收消息确认机制时，多数选择的模式。消费者接收到消息后，手动调用用 basic.ack/basic.nack/basic.reject后，RabbitMQ收到消息后，才认为本次投递成功。
basic.ack 用于肯定确认，
basic.nack 用于否定确认，
basic.reject 用于否定确认，但与basic.nack相比有一个限制，一次只能拒绝单条消息，消费者端以上的3个方法都表示消息已经被正确投递，但是basic.ack表示消息已经被正确处理，但是basic.nack,basic.reject 表示没有被正确处理，但是RabbitMQ中仍然需要删除这条消息。

看了上面那么多介绍，接下来我们一起配置下，看看一般的消息接收，手动确认是怎样的。
之前介绍用了很多个交换，现在我们就先给直连型交换机添加信息接收确认机制。
新建 MessageListenerConfig.java 上添加代码相关的配置代码（可以看到注释掉的代码，就是给扇型交换机配置消息确认，只用在这个里面继续添加对应的队列和对应的接收类即可，当然对应的接收类也需要跟后面介绍一个加上对应方法）：

package com.taotao.rabbitmqconsumer.config;

import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @autor tom
 * @date 2020/3/20 0020 7:35
 */
@Configuration
public class MessageListenerConfig {
    @Autowired
    private CachingConnectionFactory cachingConnectionFactory;
    @Autowired
    private DirectReceiver directReceiver;//Direct消息接收处理类
    //@Autowired
    //FanoutReceiverA fanoutReceiverA;//消息接收处理类A

    @Autowired
    DirectRabbitConfig directRabbitConfig;

    //@Autowired
    //   FanoutRabbitConfig fanoutRabbitConfig;
    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(cachingConnectionFactory);
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(1);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默认是自动确认，这里改为手动确认消息
        container.setQueues(directRabbitConfig.TestDirectQueue());
        container.setMessageListener(directReceiver);
//        container.addQueues(fanoutRabbitConfig.queueA());
//        container.setMessageListener(fanoutReceiverA);
        return container;
    }
}

然后在直连型交换机的消息接收处理类上需要添加相关的消息手动确认代码
DirectReceiver.java

package com.taotao.rabbitmqconsumer.config;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * 创建消息接收监听类
 * @autor tom
 * @date 2020/3/19 0019 15:29
 */
@Component
@RabbitListener(queues = "TestDirectQueue")//监听的队列名称 TestDirectQueue
public class DirectReceiver implements ChannelAwareMessageListener {

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //因为传递消息的时候用的map传递,所以将Map从Message内取出需要做些处理
            String msg = message.toString();
            String[] msgArray = msg.split("'");//可以点进Message里面看源码,单引号直接的数据就是我们的map消息数据
            Map<String, String> msgMap = mapStringToMap(msgArray[1].trim());
            String messageId=msgMap.get("messageId");
            String messageData=msgMap.get("messageData");
            String createTime=msgMap.get("createTime");
            System.out.println("messageId:"+messageId+"  messageData:"+messageData+"  createTime:"+createTime);
            channel.basicAck(deliveryTag, true);
//          channel.basicReject(deliveryTag, true);//为true会重新放回队列
        } catch (Exception e) {
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }

    //{key=value,key=value,key=value} 格式转换成map
   private  Map<String,String>mapStringToMap(String str){
       str = str.substring(1, str.length() - 1);
       String[] strs = str.split(",");
       Map<String, String> map = new HashMap<String, String>();
       for (String string : strs) {
           String key = string.split("=")[0].trim();
           String value = string.split("=")[1];
           System.out.println(key+"*****");
           map.put(key, value);
       }
       return map;
   }
}

然后在将rabbitmq-provider, rabbitmq-consumer俩个项目跑起来，访问下/sendDirectMessage接口往直连型交换机推送一条消息，看看监听到的消息结果：

createTime*****
messageId*****
messageData*****
messageId:81469828-9008-4218-ba6a-29f6d2c0c8c8  messageData:test message. hello!  createTime:2020-03-20 10:00:28

手动的确认模式的投递消息略低于自动，但是可以弥补自动确认模式的不足，更加准确的去记录消费情况，
那么如果需要有些消息接收类设置自动确认，有些消息接收类设置需要手动确认的话，那么只需要将需要设置手动确认的相关对垒加到之前的MessageListenerConfig的SimpleMessageListenerContainer里就行，
PS: 我这个就是扇形交换机A,B,C里面仅仅对于FanoutReceiverA设置了手动确认，推送一条消息，情况如下：

createTime*****
messageId*****
messageData*****
messageId:92fd2837-8ebb-43cc-83cf-b84ef78b2451  messageData:message:  woman is all  createTime:2020-03-20 10:19:50
FamoutReceiverB消费者收到消息：{createTime=2020-03-20 10:19:50, messageId=92fd2837-8ebb-43cc-83cf-b84ef78b2451, messageData=message:  woman is all}
FamoutReceiverC消费者收到消息：{createTime=2020-03-20 10:19:50, messageId=92fd2837-8ebb-43cc-83cf-b84ef78b2451, messageData=message:  woman is all}

ok, nice.

4,springboot 整合RabbitMq

4,springboot 整合RabbitMq

（并号）用来表示任意数量（零个或者多个）单词