SpringBoot集成RabbitMq(二)_Java

RabbitMq四种交换机

RabbitMq模型

名词解释

Broker：简单来说就是消息队列服务器实体。
Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。
Queue：消息队列载体，每个消息都会被投入到一个或多个队列。
Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来。
Routing Key：路由关键字，exchange根据这个关键字进行消息投递。
vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离。
producer：消息生产者，就是投递消息的程序。
consumer：消息消费者，就是接受消息的程序。
channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务。

Topic Exchange

主题交换机:当生产者将消息投递到Topic exchange并指定路由键routing key时,而routing key 由一系列的标识符组成,这些标识符由".“隔开,会分发到与routing key模糊匹配的所有queue中,模糊匹配符为”#“和”*",
若用来绑定交换机与队列的 routing key 中存在"#“表示模糊匹配所有标识符,若存在”*“表示模糊匹配一个标识符.例如若绑定的路由键为"a.#”,当生产者发送的routing key为"a.b"或者"a.b.c"时,消息都会分发到"a.#“绑定的queue中;如若绑定的路由键为"a.*”,当生产者发送的routing key为"a.b"或者"a.b.c"时,只有"a.b"消息都会分发到"a.*"绑定的queue中

RabbitMqConfig

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMqConfig {
    private static final  String topicExchange="topicExchange";
    private static final  String topicRoutingkey="key.#";
    @Bean
    public TopicExchange topicExchange(){
        return  new TopicExchange(topicExchange,true,false);
    }
    @Bean
    public Queue queue1(){
        return  new Queue("queue1",true,false,false);
    }
    @Bean
    public Queue queue2(){
        return  new Queue("queue2",true,false,false);
    }
    @Bean
    public Binding binding1(){
        return BindingBuilder.bind(queue1()).to(topicExchange()).with(topicRoutingkey);
    }
    @Bean
    public Binding binding2(){
        return BindingBuilder.bind(queue2()).to(topicExchange()).with(topicRoutingkey);
    }

}

生产者生产消息

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import java.util.HashMap;

@Component
@RequestMapping("/rabbit")
public class TestRabbitMq {
    @Autowired
    private AmqpTemplate rabbitTemplate;
    @ResponseBody
    @RequestMapping("/1")
    public String test1(){
        rabbitTemplate.convertAndSend("queuedemo","hello world");
        return "1";
    }
    @ResponseBody
    @RequestMapping("/2")
    public String test2(){
        rabbitTemplate.convertAndSend("topicExchange","key.1","key.1 success");
        rabbitTemplate.convertAndSend("topicExchange","key.1.2","key.1.2 success");
        return "2";
    }
}

调用接口localhost:7777/rabbit/2
访问http://www.lhsxpumps.com/_localhost:15672/
web 监控
队列中有我们生产的两条消息

Direct Exchange

直接交换机:当生产者将消息投递到Direct exchange并指定路由键routing key时,只分发到与routing key完全匹配的所有queue中.直连交换机除了完全匹配外,和主题交换机一致,所以就不重复了

Fanout Exchange

扇形交换机:扇形交换机是最基本的交换机类型，它能做的事非常简单——广播消息，扇形交换机会把能接收到的消息全部发送给绑定在自己身上的队列,并且扇形交换机与队列绑定不需要routing key,所有生产者生产的消息也不需要指定routing key。因为广播不需要"思考"，所以扇形交换机处理消息的速度也是所有的交换机类型里面最快的。

RabbitMqConfig

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMqConfig {
    private static final  String topicExchange="topicExchange";
    private static final  String topicRoutingkey="key.#";
    private static final  String fanoutExchange="fanoutExchange";
    @Bean
    public TopicExchange topicExchange(){
        return  new TopicExchange(topicExchange,true,false);
    }
     @Bean
    public FanoutExchange fanoutExchange(){
        return  new FanoutExchange(fanoutExchange,true,false);
    }
    @Bean
    public Queue queue1(){
        return  new Queue("queue1",true,false,false);
    }
    @Bean
    public Queue queue2(){
        return  new Queue("queue2",true,false,false);
    }
    @Bean
    public Queue fanoutqueue1(){
        return  new Queue("fanoutqueue1",true,false,false);
    }
    @Bean
    public Queue fanoutqueue2(){
        return  new Queue("fanoutqueue2",true,false,false);
    }
    @Bean
    public Binding binding1(){
        return BindingBuilder.bind(queue1()).to(topicExchange()).with(topicRoutingkey);
    }
    @Bean
    public Binding binding2(){
        return BindingBuilder.bind(queue2()).to(topicExchange()).with(topicRoutingkey);
    }
    @Bean
    public Binding binding3(){
        return BindingBuilder.bind(fanoutqueue1()).to(fanoutExchange());
    }
    @Bean
    public Binding binding4(){
        return BindingBuilder.bind(fanoutqueue2()).to(fanoutExchange());
    }

}

生产者生产消息

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import java.util.HashMap;

@Component
@RequestMapping("/rabbit")
public class TestRabbitMq {
    @Autowired
    private AmqpTemplate rabbitTemplate;
    @ResponseBody
    @RequestMapping("/1")
    public String test1(){
        rabbitTemplate.convertAndSend("queuedemo","hello world");
        return "1";
    }
    @ResponseBody
    @RequestMapping("/2")
    public String test2(){
        rabbitTemplate.convertAndSend("topicExchange","key.1","key.1 success");
        rabbitTemplate.convertAndSend("topicExchange","key.1.2","key.1.2 success");
        return "2";
    }
    @ResponseBody
    @RequestMapping("/3")
    public String test3(){
        rabbitTemplate.convertAndSend("fanoutExchange","","fanout exchange message");
        return "3";
    }
}

调用接口localhost:7777/rabbit/3
访问http://www.lhsxpumps.com/_localhost:15672/

web 管理
看出消息发送到与fanout exchange 绑定的所有queue 而不通过routing key

Headers Exchange

头部交换机:头部交换机是忽略routing_key的一种路由方式。路由器和交换机路由的规则是通过Headers信息来交换的，当头部交换机绑定一个队列的时候，需要定义一个Map<String,Object>的数据结构，并且要map结构中要求携带一个键"x-match"，这个键的Value可以是any或者all，any为任何一个键值对匹配就行,all为所有键值对都得匹配.生产者生产消息，需要携带Map数据结构的信息，当map内容匹配上的时候，消息就会被写入队列。与其他交换机相比，头部交换机的优势是匹配的规则不被限定为字符串。

RabbitMqConfig


import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

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

@Configuration
public class RabbitMqConfig {

    private static final  String headsExchange="headsExchange";


    @Bean
    public HeadersExchange headersExchange(){
        return  new HeadersExchange(headsExchange,true,false);
    }
    @Bean
    public Queue headsqueue1(){
        return  new Queue("headsqueue1",true,false,false);
    }

    @Bean
    public Binding binding5(){
        Map<String,Object> map=new HashMap<>();
        //map.put("x-match","any");
        map.put("heads1","value1");
        map.put("heads2","value2");
        //return BindingBuilder.bind(headsqueue1()).to(headersExchange()).whereAll(map).match();
        return BindingBuilder.bind(headsqueue1()).to(headersExchange()).whereAny(map).match();
    }

}

生产者生产消息

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import java.util.HashMap;

@Component
@RequestMapping("/rabbit")
public class TestRabbitMq {
    @Autowired
    private AmqpTemplate rabbitTemplate;
    
    @ResponseBody
    @RequestMapping("/4")
    public String test4(){
        MessageProperties messageProperties=new MessageProperties();
        messageProperties.setHeader("heads1","value1");
        String headsMessage="heads exchange success";
        Message msg=new Message(headsMessage.getBytes(),messageProperties);
        rabbitTemplate.convertAndSend("headsExchange","",msg);
        return "4";
    }
}

调用接口localhost:7777/rabbit/4
访问http://www.lhsxpumps.com/_localhost:15672/

web 管理
发现只有当匹配方式为any时消息才会被分发到队列中