关于rabbitmq，你想知道的都在这里（上）

关于rabbitmq，你想知道的都在这里（上）#### rabbitmq在window上安装依赖erlang环境，首先现在erlang

https://www.erlang.org/downloads

然后下载rabbitmq的windows版本

https://www.rabbitmq.com/download.html

首先我们了解下消息队列是由交换机exchange和队列组合构成的，有三种形式

直连型：一个交换机关联一个队列，指定一个路由key，消息通过交换机名称和路由key发送到指定队列，发送一个，队列里面就多一个消息。
扇型：一个交换机关联多个队列。消息通过交换机名称发送，所有关联了这个交换机的队列都将收到消息，发送一个消息再N个消息队列产生N个一模一样的消息数据。
主题型：一个交换机根据规则关联多个队列。这种类型与扇型的很像，但是主题型会根据动态路由key决定消息的投递到哪个队列。这里的路由规则很像正则表达式。会根据事先设定的路由规则动态将消息投递到队列，可能投递到一个队列也可能投递到多个队列。

首先添加依赖：

<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>

yml文件指定rabbitmq连接信息

server:
  port: 8021
spring:
  #给项目来个名字
  application:
    name: rabbitmq-provider
  #配置rabbitMq 服务器
  rabbitmq:
    host: localhost
    port: 5672
#    username: skywalker
#    password: skywalker
#    #虚拟host 可以不设置,使用server默认host
#    virtual-host: skywalker-virtualhost
    #确认消息已发送到交换机(Exchange)
    publisher-confirm-type: correlated
    #确认消息已发送到队列(Queue)
    publisher-returns: true

注意：我们需要创建两个工程，一个生产者producer、一个消费者comsumer，生产者用来生产消息，消费者用来消费生产者将消息投递到rabbitmq中的消息。

两个工程中的pom依赖一样，yml也一样，只需要将server.port设置成不同的端口即可。这里我们将生产者设置为8021端口，消费者设置为8022端口。

直连型：

从上面的讲解中我们知道，有交换机exchange，有队列queue，有路由routing，因此我们需要在生产者端将三者关联起来，然后发送消息，这样消费端才能收到消息。

绑定关联

@Configuration
public class DirectRabbitConfig {
    public static String directRouting = "directRouting";
    public static String directQueue = "directQueue";
    public static String directExchange = "directExchange";
    @Bean
    public Queue DirectQueue() {
        return new Queue(DirectRabbitConfig.directQueue,true);  //true 是否持久
    }
    @Bean
    DirectExchange DirectExchange() {
        return new DirectExchange(DirectRabbitConfig.directExchange);
    }
    @Bean
    Binding bindingDirect() {
//        BindingBuilder.bind(队列A).to(交换机B).with(路由) 将队列A绑定到交换机B，使用路由C传递消息
        return BindingBuilder.bind(DirectQueue()).to(DirectExchange()).with(directRouting);
    }
}

发送消息

@Autowired
private RabbitTemplate rabbitTemplate;  //使用RabbitTemplate,这提供了接收/发送等等方法

@GetMapping("/sendDirectMsg")
public String sendDirectMsg() {
    Map<String,Object> map=new HashMap<String,Object>();
    map.put("id",UUID.randomUUID().toString());
    map.put("data","hello,i am direct msg!");
    map.put("datetime",System.currentTimeMillis());
    //交换机 路由 消息（发送消息的时候不需要管队列，因为队列已经在DirectRabbitConfig中配置了，队列应该是消费者关心的事情）
    rabbitTemplate.convertAndSend(DirectRabbitConfig.directExchange, DirectRabbitConfig.directRouting, map);
    return "ok";
}

注意：在发送完消息后，通过控制台http://localhost:15672 （默认用户名密码都是guest），查看是否产生了消息，如下界面：

接收消息

@Component
@RabbitListener(queues = "directQueue")//监听的队列名称 directQueue，不需要管路由和交换机，因为这些是生产者管理的事情。消费者只需要关心队列即可
public class DirectReceiver {
    @RabbitHandler
    public void handler(Map testMessage) {
        System.out.println("directReceiver消费者收到消息  : " + testMessage.toString());
    }
}

扇型

配置关联

@Configuration
public class FanoutRabbitConfig {
    public static String fanoutQueue1="fanoutQueue1";
    public static String fanoutQueue2="fanoutQueue2";
    public static String fanoutQueue3="fanoutQueue3";
    public static String fanoutExchange = "fanoutExchange";

    @Bean
    public Queue queue1() {
        return new Queue(FanoutRabbitConfig.fanoutQueue1);
    }
    @Bean
    public Queue queue2() {
        return new Queue(FanoutRabbitConfig.fanoutQueue2);
    }
    @Bean
    public Queue queue3() {
        return new Queue(FanoutRabbitConfig.fanoutQueue3);
    }

    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange(FanoutRabbitConfig.fanoutExchange);
    }

    @Bean
    Binding bindingExchange1() {
        //将队列fanoutQueue1 绑定到 fanoutExchange 
        return BindingBuilder.bind(queue1()).to(fanoutExchange());
    }
    @Bean
    Binding bindingExchange2() {
        return BindingBuilder.bind(queue2()).to(fanoutExchange());
    }
    @Bean
    Binding bindingExchange3() {
        return BindingBuilder.bind(queue3()).to(fanoutExchange());
    }
}

发送消息

@GetMapping("/sendFanoutMsg")
public String sendFanoutMsg() {
    Map<String, Object> map = new HashMap<String,Object>();
    map.put("id",UUID.randomUUID().toString());
    map.put("data","hello,i am fanout msg!");
    map.put("datetime",System.currentTimeMillis());
    rabbitTemplate.convertAndSend(FanoutRabbitConfig.fanoutExchange, null, map);//扇型不需要路由key，设置了也无效
    return "ok";
}

接收消息

@Component
@RabbitListener(queues = "fanoutQueue1")
public class FanoutReceiver1 {

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

}

注意：扇型我们配置了3个队列，此处指列出了一个队列的名称，自行加2和3即可，使用postman发送消息后再控制台可以看到：

三个队列都接收到了消息，消费没有先后顺序。

主题型

配置

@Configuration
public class TopicRabbitConfig {
    //绑定键
    public static String topicQueue1 = "topicQueue1";
    public static String topicQueue2 = "topicQueue2";
    public static String topicExchange = "topicExchange";
    public static String topicRoutingApple = "fruit.apple";
//    *  表示1~n个字符 (必须出现的)
//    #  表示0~n个字符 (可能不出现)
//    若队列绑定为#，则无视消息路由，接收所有消息
//    当*和#都未出现时，就相当于直连direct
    public static String topicRoutingFruit = "fruit.#";

    @Bean
    public Queue topicQueue1() {
        return new Queue(TopicRabbitConfig.topicQueue1);
    }
    @Bean
    public Queue topicQueue2() {
        return new Queue(TopicRabbitConfig.topicQueue2);
    }

    @Bean
    TopicExchange exchange() {
        return new TopicExchange(TopicRabbitConfig.topicExchange);
    }

    //只有携带路由key 为fruit.apple,才会分发到该队列
    @Bean
    Binding bindingExchangeMessage() {
        return BindingBuilder.bind(topicQueue1()).to(exchange()).with(TopicRabbitConfig.topicRoutingApple);
    }
    //只要是消息携带的路由键是以fruit.开头,都会分发到该队列
    @Bean
    Binding bindingExchangeMessage2() {
        return BindingBuilder.bind(topicQueue2()).to(exchange()).with(TopicRabbitConfig.topicRoutingFruit);
    }
}

发送消息

@GetMapping("/sendTopicMsg1")
public String sendTopicMsg1() {
    Map<String, Object> map = new HashMap<String,Object>();
    map.put("id",UUID.randomUUID().toString());
    map.put("data","hello,i am topic msg from apple!");
    map.put("datetime",System.currentTimeMillis());
    rabbitTemplate.convertAndSend(TopicRabbitConfig.topicExchange, TopicRabbitConfig.topicRoutingApple, map);
    return "ok";
}

@GetMapping("/sendTopicMsg2")
public String sendTopicMsg2() {
    Map<String, Object> map = new HashMap<String,Object>();
    map.put("id",UUID.randomUUID().toString());
    map.put("data","hello,i am topic msg from fruit!");
    map.put("datetime",System.currentTimeMillis());
    rabbitTemplate.convertAndSend(TopicRabbitConfig.topicExchange, "fruit.orange", map);
    return "ok";
}

接收消息

@Component
@RabbitListener(queues = "topicQueue1")
public class Topic1Receiver {

    @RabbitHandler
    public void handler(Map testMessage) {
        System.out.println("Topic1Receiver消费者收到消息  : " + testMessage.toString());
    }
}
@Component
@RabbitListener(queues = "topicQueue2")
public class Topic2Receiver {

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