robbitMq 入门学习

砍鸡鸡

修改于 2021-02-24 17:54:21

6380

修改于 2021-02-24 17:54:21

文章被收录于专栏：java 挖矿之路

MQ全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通信方法。MQ是消费-生产者模型的一个典型的代表，一端往消息队列中不断写入消息，而另一端则可以读取队列中的消息。消息中间件最主要的作用是解耦，中间件最标准的用法是生产者生产消息传送到队列，消费者从队列中拿取消息并处理，生产者不用关心是谁来消费，消费者不用关心谁在生产消息，从而达到解耦的目的。在分布式的系统中，消息队列也会被用在很多其它的方面，比如：分布式事务的支持，RPC的调用等等。

RabbitMQ介绍

RabbitMQ是实现AMQP（高级消息队列协议）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。RabbitMQ主要是为了实现系统之间的双向解耦而实现的。当生产者大量产生数据时，消费者无法快速消费，那么需要一个中间层。保存这个数据。

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。

RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

四种交换机(Exchange)

交换机的功能主要是接收消息并且转发到绑定的队列，交换机不存储消息，在启用ack模式后，交换机找不到队列会返回错误。交换机有四种类型：Direct, topic, Headers and Fanout

1. Direct Exchange

direct 类型的行为是"先匹配, 再投送". 即在绑定时设定一个 routingkey, 消息的routingkey 匹配时, 才会被交换器投送到绑定的队列中去.是RabbitMQ默认的交换机模式，也是最简单的模式，根据key全文匹配去寻找队列。

配置：设置一个路由键

public  static  final String QUEUE="queue";
    /**
     * direct 交换机模式
     */
    @Bean
    public Queue queue(){
        return new Queue(QUEUE,true);
    }

发送服务：

@Service
@Slf4j
public class MQSender {

    @Autowired
    AmqpTemplate amqpTemplate;

    public void send(Object message){
        String msg = (String) message;
        log.info("send msg"+message);
        amqpTemplate.convertAndSend(MQConfig.QUEUE,msg);
    }
}

接收服务：

@Service
@Slf4j
public class MQReceiver {

    //监听的queue
    @RabbitListener(queues = MQConfig.QUEUE)
    public void receive(String msg){
        log.info("receive msg "+msg);
    }
}

测试：

@Autowired
    private MQSender sender;

    sender.send("hello direct Exchange");

2. Topic Exchange

按规则转发消息（最灵活）转发消息主要是根据通配符。在这种交换机下，队列和交换机的绑定会定义一种路由模式，那么，通配符就要在这种路由模式和路由键之间匹配后交换机才能转发消息。

路由键必须是一串字符，用句号（.）隔开，

路由模式必须包含一个星号（*），主要用于匹配路由键指定位置的一个单词，井号（#）就表示相当于一个或者多个单词

配置类：

public  static  final String TOPIC_QUEUE1="topic.queue1";
    public  static  final String TOPIC_QUEUE2="topic.queue2";
    public  static  final String ROUTING_KEY1="topic.key1";
    public  static  final String ROUTING_KEY2="topic.#";
    /**
     * Topic 交换机模式  可以用通配符
     */
    @Bean
    public Queue topicQueue1(){
        return new Queue(TOPIC_QUEUE1,true);
    }
    @Bean
    public Queue topicQueue2(){
        return new Queue(TOPIC_QUEUE2,true);
    }

    @Bean
    public TopicExchange topicExchange(){
        return new TopicExchange(TOPIC_EXCHANGE);
    }
    @Bean
    public Binding topicBinding1(){
        return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with(ROUTING_KEY1);
    }
    @Bean
    public Binding topicBinding2(){
        return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with(ROUTING_KEY2);
    }

发送类：

public void sendTopic(Object message){
        String msg = (String) message;
        log.info("send topic message"+msg);
        amqpTemplate.convertAndSend(MQConfig.TOPIC_EXCHANGE,"topic.key1",msg+"1");
        amqpTemplate.convertAndSend(MQConfig.TOPIC_EXCHANGE,"topic.key2",msg+"2");
    }

接收类：

@RabbitListener(queues = MQConfig.TOPIC_QUEUE1)
    public void receiveTopic1(String msg){
        log.info("receive topic1 msg "+msg);
    }

测试：

@Autowired
    private MQSender sender;

    sender.sendTopic("hello topic Exchange");

3. Headers Exchange

设置header attribute参数类型的交换机，相较于 direct 和 topic 固定地使用 routing_key , headers 则是一个自定义匹配规则的类型. 在队列与交换器绑定时, 会设定一组键值对规则, 消息中也包括一组键值对( headers 属性), 当这些键值对有一对, 或全部匹配时, 消息被投送到对应队列.

public static final String HEADER_EXCHANGE="headerExchange";
    /**
     * Header 交换机模式
     */
    @Bean
    public HeadersExchange headersExchange(){
        return new HeadersExchange(HEADER_EXCHANGE);
    }
    @Bean
    public Queue headerQueue(){
        return new Queue(HEADER_QUEUE2,true);
    }
    // 绑定需要指定header，如果不匹配 则不能使用
    @Bean
    public Binding headerBinding(){
        Map<String,Object> map = new HashMap();
        map.put("header1","value1");
        map.put("header2","value2");
        return BindingBuilder.bind(headerQueue()).to(headersExchange()).whereAll(map).match();
    }

public void sendHeader(Object massage){
        String msg = (String) massage;
        log.info("send fanout message: "+msg);
        MessageProperties properties = new MessageProperties();
        properties.setHeader("header1","value1");
        properties.setHeader("header2","value2");
        Message obj = new Message(msg.getBytes(),properties);
        amqpTemplate.convertAndSend(MQConfig.HEADER_EXCHANGE,"",obj);
    }

用MessageProperties来添加Header信息，然后与接收者的header比对。我都设置的是"header1","value1"；"header2","value2"

//监听 header模式的queue
    @RabbitListener(queues = MQConfig.HEADER_QUEUE2)
    //因为发送的是 byte 类型，所以接受也是该数据类型
    public void receiveHeader(byte[] message){
        log.info("header queue message"+new String(message));
    }

测试：

@Autowired
    private MQSender sender;

    sender.sendHeader("hello header Exchange");

4. Fanout Exchange

转发消息到所有绑定队列，消息广播的模式，不管路由键或者是路由模式，会把消息发给绑定给它的全部队列，如果配置了routing_key会被忽略。

public static final String FANOUT_EXCHANGE="fanoutExchange";
    /**
     * Fanout 交换机模式（广播模式）,不用绑定key
     */
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange(FANOUT_EXCHANGE);
    }
    @Bean
    public Binding fanoutBinding1(){
        return BindingBuilder.bind(topicQueue1()).to(fanoutExchange());
    }
    @Bean
    public Binding fanoutBinding2(){
        return BindingBuilder.bind(topicQueue2()).to(fanoutExchange());
    }

public void sendFanout(Object massage){
        String msg = (String) massage;
        log.info("send fanout message: "+msg);
        amqpTemplate.convertAndSend(MQConfig.FANOUT_EXCHANGE,"",msg);
    }

测试：

@Autowired
    private MQSender sender;

    sender.sendFanout("hello fanout Exchange");

补充

这个示例是基于springboot的示例。

pom依赖

<!--rabbbitMQ相关依赖-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

配置文件

rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    # 这个账号密码只能连接本地的mq，远程的话需要配置
    virtual-host: /
    listener:
      simple:
        concurrency: 10
        max-concurrency: 10
        prefetch: 1 # 从队列每次取一个
        auto-startup: true
        default-requeue-rejected: true # 失败后重试

后续会用它来实现一个小小的抢票功能。

END

本文系转载，前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

html