RabbiMQ原理与SpringBoot使用

rabbimq介绍

具体代码可参考我的github：https://github.com/uniquedong/springboot-study

一、使用场景

rabbitmq是一个消息中间件，所以最主要的作用就是：信息缓冲区，实现应用程序的异步和解耦。

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

amqp，即advanced message queuing protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。amqp的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。详细概念可以参考官方指南 rabbitmq

二、相关概念

通常我们谈到队列服务, 会有三个概念： 发消息者、队列、收消息者，rabbitmq 在这个基本概念之上, 多做了一层抽象, 在发消息者和 队列之间, 加入了交换器 (exchange). 这样发消息者和队列就没有直接联系, 转而变成发消息者把消息给交换器, 交换器根据调度策略再把消息再给队列。

那么，其中比较重要的概念有 4 个，分别为：虚拟主机，交换机，队列，和绑定。

• 虚拟主机v-host：一个虚拟主机持有一组交换机、队列和绑定。为什么需要多个虚拟主机呢？很简单，rabbitmq当中，用户只能在虚拟主机的粒度进行权限控制。 因此，如果需要禁止a组访问b组的交换机/队列/绑定，必须为a和b分别创建一个虚拟主机。每一个rabbitmq服务器都有一个默认的虚拟主机。
• 交换机：exchange 用于转发消息，但是它不会做存储 ，如果没有 queue bind 到 exchange 的话，它会直接丢弃掉 producer 发送过来的消息。这里有一个比较重要的概念：路由键 。消息到交换机的时候，交互机会转发到对应的队列中，那么究竟转发到哪个队列，就要根据该路由键。
• 绑定：也就是交换机需要和队列相绑定，这其中如上图所示

交换机（exchange）

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

• direct：direct 类型的行为是"先匹配, 再投送". 即在绑定时设定一个 routing_key, 消息的routing_key 匹配时, 才会被交换器投送到绑定的队列中去.
• topic：按规则转发消息（最灵活）
• headers：设置header attribute参数类型的交换机
• fanout：转发消息到所有绑定队列

direct exchange

direct exchange是rabbitmq默认的交换机模式，也是最简单的模式，根据key全文匹配去寻找队列。

第一个 x - q1 就有一个 binding key，名字为 orange； x - q2 就有 2 个 binding key，名字为 black 和 green。当消息中的 路由键 和 这个 binding key 对应上的时候，那么就知道了该消息去到哪一个队列中。

ps：为什么 x 到 q2 要有 black，green，2个 binding key呢，一个不就行了吗？ - 这个主要是因为可能又有 q3，而q3只接受 black 的信息，而q2不仅接受black 的信息，还接受 green 的信息。

topic exchange

根据通配符转发消息到队列，在这种交换机下，队列和交换机的绑定会定义一种路由模式，那么，通配符就要在这种路由模式和路由键之间匹配后交换机才能转发消息。

• *（星号）可以替代一个单词。
• ＃（hash）可以替换零个或多个单词。

headers exchange

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

fanout exchange

消息广播的模式，也就是我们的发布订阅模式。fanout exchange 消息广播的模式，不管路由键或者是路由模式，会把消息发给绑定给它的全部队列，如果配置了routing_key会被忽略。

消息确认

消息消费者如何通知 rabbit 消息消费成功？

消息通过 ack 确认是否被正确接收，每个 message 都要被确认（acknowledged），可以手动去 ack 或自动 ack 自动确认会在消息发送给消费者后立即确认，但存在丢失消息的可能，如果消费端消费逻辑抛出异常，也就是消费端没有处理成功这条消息，那么就相当于丢失了消息 如果消息已经被处理，但后续代码抛出异常，使用 spring 进行管理的话消费端业务逻辑会进行回滚，这也同样造成了实际意义的消息丢失 如果手动确认则当消费者调用 ack、nack、reject 几种方法进行确认，手动确认可以在业务失败后进行一些操作，如果消息未被 ack 则会发送到下一个消费者 如果某个服务忘记 ack 了，则 rabbitmq 不会再发送数据给它，因为 rabbitmq 认为该服务的处理能力有限 ack 机制还可以起到限流作用，比如在接收到某条消息时休眠几秒钟 消息确认模式有：

• acknowledgemode.none：自动确认
• acknowledgemode.auto：根据情况确认
• acknowledgemode.manual：手动确认

springboot集成rabbitmq

1. 配置pom，主要添加spring-boot-starter-amqp支持，springboot基于2.1.4版本

<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>
</dependency>
<dependency>
    <groupid>org.projectlombok</groupid>
    <artifactid>lombok</artifactid>
    <optional>true</optional>
</dependency>

1. 配置springboot的yaml文件

   server:
     servlet:
       context-path: /rabbitmq
     port: 9004
   spring:
     application:
       name: rabbitmq
     rabbitmq:
       host: localhost
       virtual-host: /crawl
       username: xxxx
       password: xxx
       port: 5672
       # 消息失败返回，比如路由不到队列时触发回调
       publisher-returns: true
       # 消息正确发送确认
       publisher-confirms: true
       template:
         retry:
           enabled: true
           initial-interval: 2s
       listener:
         simple:
           # 手动ack 不开启自动ack模式,目的是防止报错后未正确处理消息丢失 默认 为 none
           acknowledge-mode: manual

另外我们还要配置ack确认回调的配置，通过实现rabbittemplate.confirmcallback接口，消息发送到broker后触发回调，也就是只能确认是否正确到达exchange中。

import lombok.extern.slf4j.slf4j;
import org.springframework.amqp.rabbit.connection.correlationdata;
import org.springframework.amqp.rabbit.core.rabbittemplate;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.stereotype.component;

import javax.annotation.postconstruct;

/**
 * @author lijianqing
 * @version 1.0
 * @classname rabbittemplateconfirmcallback
 * @date 2019/4/23 12:55
 */
@component
@slf4j
public class rabbittemplateconfirmcallback implements rabbittemplate.confirmcallback {
    @autowired
    private rabbittemplate rabbittemplate;

    @postconstruct
    public void init() {
        //指定 confirmcallback
        rabbittemplate.setconfirmcallback(this);
    }

    @override
    public void confirm(correlationdata correlationdata, boolean ack, string cause) {
        log.info("消息唯一标识：{},确认结果：{},失败原因：{}", correlationdata, ack, cause);
    }
}

消息失败返回，比如路由步到队列就会触发，如果西区奥西发送到交换器成功，但是没有匹配的队列就会触发回调

import lombok.extern.slf4j.slf4j;
import org.springframework.amqp.core.message;
import org.springframework.amqp.rabbit.core.rabbittemplate;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.stereotype.component;

import javax.annotation.postconstruct;

/**
 * @author lijianqing
 * @version 1.0
 * @classname rabbittemplatereturncallback
 * @date 2019/4/23 12:55
 */
@component
@slf4j
public class rabbittemplatereturncallback implements rabbittemplate.returncallback {
    @autowired
    private rabbittemplate rabbittemplate;

    @postconstruct
    public void init() {
        //指定 returncallback
        rabbittemplate.setreturncallback(this);
        rabbittemplate.setmandatory(true);
    }

    @override
    public void returnedmessage(message message, int replycode, string replytext, string exchange, string routingkey) {
        log.info("消息主体 message : " + message);
        log.info("消息主体 message : " + replycode);
        log.info("描述：" + replytext);
        log.info("消息使用的交换器 exchange : " + exchange);
        log.info("消息使用的路由键 routing : " + routingkey);
    }
}

一、简单的开始-简单队列

如下图：

“p”是我们的生产者，“c”是我们的消费者。中间的框是一个队列 - rabbitmq代表消费者保留的消息缓冲区。

新增simpleconfig，创建我们要投放的队列：代码如下

/**
 * 队列直接投放
 * @author lijianqing
 * @version 1.0
 * @classname simpleconfig
 * @date 2019/4/26 15:11
 */
@configuration
public class simpleconfig {
    @bean
    public queue simplequeue() {
        return new queue("simple");
    }
}

再分别创建消息发送者与消息接收者：

• 消息发送者

import lombok.extern.slf4j.slf4j;
import org.springframework.amqp.rabbit.connection.correlationdata;
import org.springframework.amqp.rabbit.core.rabbittemplate;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.stereotype.component;
import zero.springboot.study.rabbitmq.model.user;

import java.util.uuid;

/**
 * @author lijianqing
 * @version 1.0
 * @classname hellosender
 * @date 2019/4/23 11:22
 */
@component
@slf4j
public class hellosender {
    @autowired
    private rabbittemplate rabbittemplate;

    public void send() {
        user user = new user();
        user.setname("青");
        user.setpass("111111");
        //发送消息到hello队列
        log.info("发送消息：{}", user);
        rabbittemplate.convertandsend("hello", user, new correlationdata(uuid.randomuuid().tostring()));

        string msg = "hello qing";
        log.info("发送消息：{}", msg);
        rabbittemplate.convertandsend("simple", msg);
    }
}

• 消息接收者

import com.rabbitmq.client.channel;
import lombok.extern.slf4j.slf4j;
import org.springframework.amqp.rabbit.annotation.rabbithandler;
import org.springframework.amqp.rabbit.annotation.rabbitlistener;
import org.springframework.amqp.support.amqpheaders;
import org.springframework.messaging.handler.annotation.header;
import org.springframework.stereotype.component;
import zero.springboot.study.rabbitmq.model.user;

import java.io.ioexception;

/**
 * 监听hello队列
 *
 * @author lijianqing
 * @version 1.0
 * @classname helloreceiver
 * @date 2019/4/23 11:42
 */
@component
@slf4j
@rabbitlistener(queues = "simple")
public class helloreceiver {

    @rabbithandler
    public void processuser(user user, channel channel, @header(amqpheaders.delivery_tag) long tag) {
        log.info("收到消息：{}", user);
        // 手动ack
        try {
//            //消息确认，代表消费者确认收到当前消息，语义上表示消费者成功处理了当前消息。
            channel.basicack(tag, false);
//             代表消费者拒绝一条或者多条消息，第二个参数表示一次是否拒绝多条消息，第三个参数表示是否把当前消息重新入队
//        channel.basicnack(deliverytag, false, false);

            // 代表消费者拒绝当前消息，第二个参数表示是否把当前消息重新入队
//        channel.basicreject(deliverytag,false);
        } catch (ioexception e) {
            e.printstacktrace();
        }
    }

    @rabbithandler
    public void processstring(string message, channel channel, @header(amqpheaders.delivery_tag) long tag) {
        log.info("收到消息：{}", message);
        // 手动ack
        try {
//            //消息确认，代表消费者确认收到当前消息，语义上表示消费者成功处理了当前消息。
            channel.basicack(tag, false);
//             代表消费者拒绝一条或者多条消息，第二个参数表示一次是否拒绝多条消息，第三个参数表示是否把当前消息重新入队
//        channel.basicnack(deliverytag, false, false);

            // 代表消费者拒绝当前消息，第二个参数表示是否把当前消息重新入队
//        channel.basicreject(deliverytag,false);
        } catch (ioexception e) {
            e.printstacktrace();
        }
    }
}

这样就实现了简单的消息发送到指定队列的模式。我们写一个测试类

二、direct exchange模式

主要配置我们的direct exchange交换机，并且创建队列通过routing key 绑定到交换机上

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;

/**
 *
 * @author lijianqing
 * @version 1.0
 * @classname directconfig
 * @date 2019/4/23 11:15
 */
@configuration
public class directconfig {

    //队列名字
    public static final string queue_name = "direct_name";

    //交换机名称
    public static final string exchange = "zero-exchange";

    //路由键名称
    public static final string routing_key = "routingkey";

    @bean
    public queue bluequeue() {
        return new queue(queue_name, true);
    }

    @bean
    public directexchange defaultexchange() {
        return new directexchange(exchange);
    }

    @bean
    public binding bindingblue() {
        return bindingbuilder.bind(bluequeue()).to(defaultexchange()).with(routing_key);
    }

}

接下来我们创建生产者与消费者

• 生产者

import lombok.extern.slf4j.slf4j;
import org.springframework.amqp.rabbit.connection.correlationdata;
import org.springframework.amqp.rabbit.core.rabbittemplate;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.stereotype.component;
import zero.springboot.study.rabbitmq.config.directconfig;
import zero.springboot.study.rabbitmq.model.user;

import java.util.uuid;

/**
 * @author lijianqing
 * @version 1.0
 * @classname hellosender
 * @date 2019/4/23 11:22
 */
@component
@slf4j
public class directsender {
    @autowired
    private rabbittemplate rabbittemplate;

    public void send() {
        user user = new user();
        user.setname("青");
        user.setpass("111111");
        //发送消息到hello队列
        log.info("directreceiver发送消息：{}", user);
        rabbittemplate.convertandsend(directconfig.exchange, directconfig.routing_key, user, new correlationdata(uuid.randomuuid().tostring()));

        string msg = "hello qing";
        log.info("directreceiver发送消息：{}", msg);
        rabbittemplate.convertandsend(directconfig.exchange, directconfig.routing_key, msg);
    }
}

• 消费者

/**
 *
 * @author lijianqing
 * @version 1.0
 * @classname helloreceiver
 * @date 2019/4/23 11:42
 */
@component
@slf4j
@rabbitlistener(queues = "direct_name")
public class directreceiver {

    @rabbithandler
    public void processuser(user user, channel channel, @header(amqpheaders.delivery_tag) long tag) {
        log.info("directreceiver收到消息：{}", user);
        // 手动ack
        try {
//            //消息确认，代表消费者确认收到当前消息，语义上表示消费者成功处理了当前消息。
            channel.basicack(tag, false);
//             代表消费者拒绝一条或者多条消息，第二个参数表示一次是否拒绝多条消息，第三个参数表示是否把当前消息重新入队
//        channel.basicnack(deliverytag, false, false);

            // 代表消费者拒绝当前消息，第二个参数表示是否把当前消息重新入队
//        channel.basicreject(deliverytag,false);
        } catch (ioexception e) {
            e.printstacktrace();
        }
    }

    @rabbithandler
    public void processstring(string message, channel channel, @header(amqpheaders.delivery_tag) long tag) {
        log.info("收到消息：{}", message);
        // 手动ack
        try {
//            //消息确认，代表消费者确认收到当前消息，语义上表示消费者成功处理了当前消息。
            channel.basicack(tag, false);
//             代表消费者拒绝一条或者多条消息，第二个参数表示一次是否拒绝多条消息，第三个参数表示是否把当前消息重新入队
//        channel.basicnack(deliverytag, false, false);

            // 代表消费者拒绝当前消息，第二个参数表示是否把当前消息重新入队
//        channel.basicreject(deliverytag,false);
        } catch (ioexception e) {
            e.printstacktrace();
        }
    }
}

三、topic exchange模式

创建队列以及交换机。并通过路由匹配规则将队列与交换机绑定上

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.beans.factory.annotation.qualifier;
import org.springframework.context.annotation.bean;
import org.springframework.context.annotation.configuration;

/**
 *  queuemessages 匹配topic.#，queuemessage 只匹配 "topic.message"
 *
 * @author lijianqing
 * @version 1.0
 * @classname topicrabbitconfig
 * @date 2019/4/23 15:03
 */
@configuration
public class topicrabbitconfig {
    final static string message = "topic.message";
    final static string messages = "topic.messages";

    @bean
    public queue queuemessage() {
        return new queue(topicrabbitconfig.message);
    }

    @bean
    public queue queuemessages() {
        return new queue(topicrabbitconfig.messages);
    }

    @bean
    topicexchange exchange() {
        return new topicexchange("topicexchange");
    }

    @bean
    binding bindingexchangemessage(@qualifier("queuemessage") queue queuemessage, topicexchange exchange) {
        return bindingbuilder.bind(queuemessage).to(exchange).with("topic.message");
    }

    @bean
    binding bindingexchangemessages(@qualifier("queuemessages") queue queuemessages, topicexchange exchange) {
        return bindingbuilder.bind(queuemessages).to(exchange).with("topic.#");
    }

}

• 创建生产者

import lombok.extern.slf4j.slf4j;
import org.springframework.amqp.rabbit.core.rabbittemplate;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.stereotype.component;

/**
 * @author lijianqing
 * @version 1.0
 * @classname topicsender
 * @date 2019/4/23 15:10
 */
@component
@slf4j
public class topicsender {
    @autowired
    private rabbittemplate rabbittemplate;

    /**
     * 匹配topic.message，两个队列都会收到
     */
    public void send1() {
        string context = "hi, i am message 1";
        log.info("主题发送 : {}" , context);
        rabbittemplate.convertandsend("topicexchange", "topic.message", context);
    }

    /**
     * 匹配topic.messages
     */
    public void send2() {
        string context = "hi, i am messages 2";
        log.info("主题发送 : {}" , context);
        rabbittemplate.convertandsend("topicexchange", "topic.messages", context);
    }
}

• 创建消费者，这里我们分别创建两个队列的消费者

@component
@rabbitlistener(queues = "topic.message")
@slf4j
public class topicreceiver {

    @rabbithandler
    public void process(string message, channel channel, @header(amqpheaders.delivery_tag) long tag) {
        log.info("topic.message receiver1  {}: ", message);
        // 手动ack
        try {
//            //消息确认，代表消费者确认收到当前消息，语义上表示消费者成功处理了当前消息。
            channel.basicack(tag, false);
//             代表消费者拒绝一条或者多条消息，第二个参数表示一次是否拒绝多条消息，第三个参数表示是否把当前消息重新入队
//        channel.basicnack(deliverytag, false, false);

            // 代表消费者拒绝当前消息，第二个参数表示是否把当前消息重新入队
//        channel.basicreject(deliverytag,false);
        } catch (ioexception e) {
            e.printstacktrace();
        }
    }

}

第二个消费者

@component
@rabbitlistener(queues = "topic.messages")
@slf4j
public class topicreceiver2 {

    @rabbithandler
    public void process(string message, channel channel, @header(amqpheaders.delivery_tag) long tag) {
        log.info("topic.messages receiver2  : {}", message);
        // 手动ack
        try {
//            //消息确认，代表消费者确认收到当前消息，语义上表示消费者成功处理了当前消息。
            channel.basicack(tag, false);
//             代表消费者拒绝一条或者多条消息，第二个参数表示一次是否拒绝多条消息，第三个参数表示是否把当前消息重新入队
//        channel.basicnack(deliverytag, false, false);

            // 代表消费者拒绝当前消息，第二个参数表示是否把当前消息重新入队
//        channel.basicreject(deliverytag,false);
        } catch (ioexception e) {
            e.printstacktrace();
        }
    }

}

四、fanout 模式

也就是发布、订阅。所有绑定在交换机上的队列都会收到消息，发送端指定的routing key的任何字符都会被忽略

配置交换机与队列

@configuration
public class fanoutrabbitconfig {
    @bean
    public queue amessage() {
        return new queue("fanout.a");
    }

    @bean
    public queue bmessage() {
        return new queue("fanout.b");
    }

    @bean
    public queue cmessage() {
        return new queue("fanout.c");
    }

    @bean
    fanoutexchange fanoutexchange() {
        return new fanoutexchange("fanoutexchange");
    }
    @bean
    binding bindingexchangea(queue amessage, fanoutexchange fanoutexchange) {
        return bindingbuilder.bind(amessage).to(fanoutexchange);
    }

    @bean
    binding bindingexchangeb(queue bmessage, fanoutexchange fanoutexchange) {
        return bindingbuilder.bind(bmessage).to(fanoutexchange);
    }

    @bean
    binding bindingexchangec(queue cmessage, fanoutexchange fanoutexchange) {
        return bindingbuilder.bind(cmessage).to(fanoutexchange);
    }
}

• 创建发送者

@component
@slf4j
public class fanoutsender {
    @autowired
    private rabbittemplate rabbittemplate;

    public void send() {
        string context = "hi, fanout msg ";
        rabbittemplate.convertandsend("fanoutexchange", null, context);
    }

}

• 创建a、b、c队列消费者

@component
@rabbitlistener(queues = "fanout.a")
@slf4j
public class fanoutreceivera {

    @rabbithandler
    public void process(string message, channel channel, @header(amqpheaders.delivery_tag) long tag) {
        log.info("fanout receiver a  : {}" , message);
        // 手动ack
        try {
//            //消息确认，代表消费者确认收到当前消息，语义上表示消费者成功处理了当前消息。
            channel.basicack(tag, false);
//             代表消费者拒绝一条或者多条消息，第二个参数表示一次是否拒绝多条消息，第三个参数表示是否把当前消息重新入队
//        channel.basicnack(deliverytag, false, false);

            // 代表消费者拒绝当前消息，第二个参数表示是否把当前消息重新入队
//        channel.basicreject(deliverytag,false);
        } catch (ioexception e) {
            e.printstacktrace();
        }
    }

}

剩下的b、c就不重复贴了。

单元测试

import org.junit.test;
import org.junit.runner.runwith;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.boot.test.context.springboottest;
import org.springframework.test.context.junit4.springrunner;
import zero.springboot.study.rabbitmq.direct.directsender;
import zero.springboot.study.rabbitmq.fanout.fanoutsender;
import zero.springboot.study.rabbitmq.simple.hellosender;
import zero.springboot.study.rabbitmq.topic.topicsender;

@runwith(springrunner.class)
@springboottest(classes = rabbitmqapplication.class)
public class rabbitmqapplicationtests {

    @autowired
    private directsender directsender;

    @autowired
    private topicsender topicsender;

    @autowired
    private fanoutsender fanoutsender;

    @autowired
    private hellosender hellosender;

    @test
    public void testdirect() {
        directsender.send();
    }

    @test
    public void topic1() {
        topicsender.send1();
    }

    @test
    public void topic2() {
        topicsender.send2();
    }

    @test
    public void testfanout() {
        fanoutsender.send();
    }

    @test
    public void testsimple() {
        hellosender.send();
    }

}

所有的代码已在我的github上分享，大家可以具体查看与提出意见。github rabbitmq模块

您的点赞与转发是我最大的肯定。

欢迎关注我的微信公众号 javastorm