RabbitMQ基本常识

1 消息队列概述

1.1 消息队列MQ

MQ 全称为 Message Queue，消息队列是应用程序和应用程序之间的通信方法。

  • 为什么使用 MQ

在项目中，可将一些无需即时返回且耗时的操作提取出来，进行异步处理，而这种异步处理的方式大大的节省了服务器的请求响应时间，从而提高了系统的吞吐量。

  • 消息队列应用场景

应用解耦、异步处理、流量削峰、日志处理、纯粹通信

1.2 AMQP和JMS

MQ 是消息通信的模型；实现 MQ 的大致有两种主流方式：AMQP、JMS。

  • AMQP

AMQP 高级消息队列协议，是一个进程间传递异步消息的网络协议，更准确的说是一种binary wire-level protocol（链接协议）。这是其和 JMS 的本质差别，AMQP 不从 API 层进行限定，而是直接定义网络交换的数据格式。

  • JMS

JMS 即 Java 消息服务（JavaMessage Service）应用程序接口，是一个 Java 平台中关于面向消息中间件（MOM）的 API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。

  • AMQP和JMS区别

JMS 是定义了统一的接口，来对消息操作进行统一；AMQP 是通过规定协议来统一数据交互的格式JMS 限定了必须使用 Java 语言；AMQP 只是协议，不规定实现方式，因此是跨语言的。JMS 规定了两种消息模式（B2B、发布订阅）；而 AMQP 的消息模式更加丰富

1.3 消息队列产品

  • kafka

Apache 下的一个子项目，使用 scala 实现的一个高性能分布式 Publish/Subscribe 消息队列系统。

1.快速持久化：通过磁盘顺序读写与零拷贝机制，可以在 O(1)的系统开销下进行消息持久化；

2.高吞吐：在一台普通的服务器上既可以达到 10W/s 的吞吐速率；

3.高堆积：支持 topic 下消费者较长时间离线，消息堆积量大；

4.完全的分布式系统：Broker、Producer、Consumer 都原生自动支持分布式，依赖zookeeper 自动实现负载均衡；

5.支持 Hadoop 数据并行加载：对于像 Hadoop 的一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。

  • RocketMQ

RocketMQ 的前身是 Metaq，当 Metaq3.0 发布时，产品名称改为 RocketMQ。RocketMQ 是一款分布式、队列模型的消息中间件，具有以下特点 ：

1.能够保证严格的消息顺序

2.提供丰富的消息拉取模式

3.高效的订阅者水平扩展能力

4.实时的消息订阅机制

5.支持事务消息

6.亿级消息堆积能力

  • RabbitMQ

使用 Erlang 编写的一个开源的消息队列，本身支持很多的协议：AMQP，XMPP, SMTP,STOMP，也正是如此，使的它变的非常重量级，更适合于企业级的开发。同时实现了 Broker 架构，核心思想是生产者不会将消息直接发送给队列，消息在发送给客户端时先在中心队列排队。对路由(Routing)，负载均衡(Load balance)、数据持久化都有很好的支持。多用于进行企业级的 ESB 整合。

1.4 rabbitMQj介绍

RabbitMQ 是由 erlang 语言开发，基于 AMQP（Advanced Message Queue 高级消息队列协议）协议实现的消息队列，它是一种应用程序之间的通信方法，消息队列在分布式系统开发中应用非常广泛。

RabbitMQ 提供了 6 种模式：简单模式，work 模式，Publish/Subscribe 发布与订阅模式，Routing 路由模式，Topics 主题模式，RPC 远程调用模式

alt

2 RabbitMQ入门（简单模式）

alt

P：生产者，也就是要发送消息的程序
C：消费者：消息的接受者，会一直等待消息到来。
Queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。
  • 添加依赖
<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>org.example</groupId>
    <artifactId>rabbitMQ01</artifactId>
    <version>1.0-SNAPSHOT</version>
    <dependencies>
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.6.0</version>
        </dependency>
    </dependencies>
</project>
  • 创建连接工具类
public class RabbitMQConUtil {
    private static final int PORT = 5672;
    private static final String HOST = "localhost";
    private static final String USER_NAME = "guest";
    private static final String PWD = "guest";
    private ConnectionFactory factory;
    private Connection connection;
    public RabbitMQConUtil(String virtualHost) {
        //创建链接工厂
        factory = new ConnectionFactory();
        //参数设置
        factory.setHost(HOST);//IP
        factory.setPort(PORT);//端口号
        factory.setUsername(USER_NAME);//用户
        factory.setPassword(PWD);//密码
        factory.setVirtualHost(virtualHost);//虚拟主机，相当于mysql中db
    }
    public RabbitMQConUtil(String host,int port,String userName,String pwd,String virtualHost){
        factory = new ConnectionFactory();
        factory.setHost(host);
        factory.setPort(port);
        factory.setUsername(userName);
        factory.setPassword(pwd);
        factory.setVirtualHost(virtualHost);
    }
    public Connection getConnection() throws Exception {
        return factory.newConnection();
    }
}
  • 创建简单生产者
public class SimpleProduct {
    public static void main(String[] args) throws Exception {
        RabbitMQConUtil util = new RabbitMQConUtil("/firstVirtualhost");
        Connection connection = util.getConnection();
        //创建频道-channel = connection.createChannel()
        Channel channel = connection.createChannel();
        //声明队列-channel.queueDeclare(名称，是否持久化，是否独占本连接,是否自动删除,附加参数)
        channel.queueDeclare("simple_queue", true, false, false, null);
        //创建消息-String m = xxx
        String message = "欢迎来到我的世界";
        //消息发送-channel.basicPublish(交换机[默认 Default Exchage],路由key[简单模式可以传递队列名称],消息其它属性,消息内容)
        channel.basicPublish("", "simple_queue", null, message.getBytes("utf-8"));
        //关闭资源-channel.close();connection.close()
        channel.close();
        connection.close();
    }
}
  • 结果 alt

alt

  • 创建消费者
public class SimpleConsumer {
    public static void main(String[] args) throws Exception{
        RabbitMQConUtil util = new RabbitMQConUtil("/firstVirtualhost");
        Connection connection = util.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("simple_queue", true, false, false, null);
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                //路由的 key
                String routingKey = envelope.getRoutingKey();
                //获取交换机信息
                String exchange = envelope.getExchange();
                //获取消息 ID
                long deliveryTag = envelope.getDeliveryTag();
                //获取消息信息
                String message = new String(body, "utf-8");
                System.out.println(
                        "routingKey:" + routingKey +
                                ",exchange:" + exchange +
                                ",deliveryTag:" + deliveryTag +
                                ",message:" + message);
            }
        };
        channel.basicConsume("simple_queue",true,consumer);
    }
}

3 RabbitMQ进阶学习

3.1 work queues工作队列模式

alt

Work Queues 与入门程序的简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。 应用场景：对于任务过重或任务较多情况，使用工作队列模式使用多个消费者可以提高任务处理的速度。

  • 代码实现

生产者

public class WorkProduct {
    public static void main(String[] args) throws Exception {
        RabbitMQConUtil util = new RabbitMQConUtil("/firstVirtualhost");
        Connection connection = util.getConnection();
        Channel channel = connection.createChannel();
        for (int i = 0; i < 10; i++) {
            channel.queueDeclare("work_queue",true,false,false,null);
            String mes = "welcome to my house:" + i;
            channel.basicPublish("","work_queue",null,mes.getBytes("UTF-8"));
        }
        channel.close();
        connection.close();
    }
}

两个或多个消费者

public class WorkConsume1 implements Runnable {
    public void run() {
        RabbitMQConUtil util = new RabbitMQConUtil("/firstVirtualhost");
        Connection connection;
        try {
            connection = util.getConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare("work_queue", true, false, false, null);
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String routingKey = envelope.getRoutingKey();

                    String exchange = envelope.getExchange();

                    long deliveryTag = envelope.getDeliveryTag();

                    String message = new String(body, "utf-8");
                    System.out.println(
                            "routingKey:" + routingKey +
                                    ",exchange:" + exchange +
                                    ",deliveryTag:" + deliveryTag +
                                    ",message:" + message);
                }
            };
            channel.basicConsume("work_queue",true,consumer);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

public class RunClass {
    public static void main(String[] args) {
        ExecutorService service = Executors.newCachedThreadPool();
        WorkConsume1 consume1 = new WorkConsume1();
        WorkConsume1 consume2 = new WorkConsume1();
        service.execute(consume1);
        service.execute(consume2);
    }
}

在一个队列中如果有多个消费者，那么消费者之间对于同一个消息的关系是竞争的关系。

3.2 publish/subscribe发布订阅模式

alt 在发布订阅模型中，多了一个 x(exchange)角色，而且过程略有变化。

P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给 X（交换机）C：消费者，消息的接受者，会一直等待消息到来。
Queue：消息队列，接收消息、缓存消息。
Exchange：交换机，图中的 X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange 的类型。Exchange有常见以下 3 种类型：
Fanout：广播，将消息交给所有绑定到交换机的队列
Direct：定向，把消息交给符合指定 routing key 的队列
Topic：通配符，把消息交给符合 routing pattern（路由模式） 的队列

Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

alt

  • 代码实现

生产者，此时生产者连接的是交换机（exchange），不是队列

public class Product {
    public static void main(String[] args) throws Exception {
        RabbitMQConUtil util = new RabbitMQConUtil("/firstVirtualhost");
        Connection connection = util.getConnection();
        Channel channel = connection.createChannel();
        //声明exchange
        channel.exchangeDeclare("fanout_exchange", BuiltinExchangeType.FANOUT);
        for (int i = 0; i < 10; i++) {
            String mes = "hello," + i + " welcome to China";
            channel.basicPublish("fanout_exchange", "", null, mes.getBytes("UTF-8"));
        }
        channel.close();
        connection.close();
    }
}

消费者，声明队列，队列绑定交换机

public class Consume1 implements Runnable {
    public void run() {
        RabbitMQConUtil util = new RabbitMQConUtil("/firstVirtualhost");
        Connection connection;
        try {
            connection = util.getConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare("publish_queue1", true, false, false, null);
            channel.queueBind("publish_queue1","fanout_exchange","key1");
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String routingKey = envelope.getRoutingKey();

                    String exchange = envelope.getExchange();

                    long deliveryTag = envelope.getDeliveryTag();

                    String message = new String(body, "utf-8");
                    System.out.println(
                            "routingKey:" + routingKey +
                                    ",exchange:" + exchange +
                                    ",deliveryTag:" + deliveryTag +
                                    ",message:" + message);
                }
            };
            channel.basicConsume("publish_queue1",true,consumer);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
public class Consume2 implements Runnable {
    public void run() {
        RabbitMQConUtil util = new RabbitMQConUtil("/firstVirtualhost");
        Connection connection;
        try {
            connection = util.getConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare("publish_queue2", true, false, false, null);
            channel.queueBind("publish_queue2","fanout_exchange","key2");
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String routingKey = envelope.getRoutingKey();

                    String exchange = envelope.getExchange();

                    long deliveryTag = envelope.getDeliveryTag();

                    String message = new String(body, "utf-8");
                    System.out.println(
                            "routingKey:" + routingKey +
                                    ",exchange:" + exchange +
                                    ",deliveryTag:" + deliveryTag +
                                    ",message:" + message);
                }
            };
            channel.basicConsume("publish_queue2",true,consumer);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
public class RunConsume {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(2,2,60, TimeUnit.MINUTES,
                new LinkedBlockingDeque<Runnable>());
        Consume1 consume1 = new Consume1();
        Consume2 consume2 = new Consume2();
        executor.execute(consume1);
        executor.execute(consume2);
    }
}
  • 发布订阅模式与 work 队列模式的区别
1、work 队列模式不用定义交换机，而发布/订阅模式需要定义交换机。
2、发布/订阅模式的生产方是面向交换机发送消息，work 队列模式的生产方是面向队列发送消息(底层使用默认交换机)。
3、发布/订阅模式的消费者需要设置队列和交换机的绑定，work 队列模式不需要设置，实际上work队列模式会将队列绑 定到默认的交换机 。

3.3 Routing 路由模式

路由模式特点：

1.队列与交换机的绑定，不能是任意绑定了，而是要指定一个 RoutingKey（路由key）
2.消息的发送方在 向 Exchange 发送消息时，也必须指定消息的 RoutingKey。
3.Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key 完全一致，才会接收到消息

alt

P：生产者，向 Exchange 发送消息，发送消息时，会指定一个 routing key。
X：Exchange（交换机），接收生产者的消息，然后把消息递交给 与 routing key 完全匹配的队列
C1：消费者，其所在队列指定了需要 routing key 为 log.error 的消息
C2：消费者，其所在队列指定了需要 routing key 为 log.info、log.error、log.warning 的消息
  • 代码实现

生产者，创建生产者向三个routing key发信息,在发布步骤添加routing key

public class Product {
    public static void main(String[] args) throws Exception {
        RabbitMQConUtil util = new RabbitMQConUtil("/firstVirtualhost");
        Connection connection = util.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare("direct_exchange", BuiltinExchangeType.DIRECT);
        for (int i = 0; i < 3; i++) {
            String routing_key = "";
            String mes = "hello:"+i;
            switch (i){
                case 0:
                    routing_key = "log.info";
                    break;
                case 1:
                    routing_key = "log.error";
                    break;
                case 2:
                    routing_key = "log.warning";
                    break;
            }
            channel.basicPublish("direct_exchange",routing_key,null,mes.getBytes("UTF-8"));
        }
        channel.close();
        connection.close();
    }
}

消费者，队列绑定时声明routing key

//消费者1 routing key为log.error
public class Consume1 implements Runnable {
    public void run() {
        RabbitMQConUtil util = new RabbitMQConUtil("/firstVirtualhost");
        Connection connection;
        try {
            connection = util.getConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare("publish_queue1", true, false, false, null);
            channel.queueBind("publish_queue1","direct_exchange","log.error");
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String routingKey = envelope.getRoutingKey();

                    String exchange = envelope.getExchange();

                    long deliveryTag = envelope.getDeliveryTag();

                    String message = new String(body, "utf-8");
                    System.out.println(
                            "routingKey:" + routingKey +
                                    ",exchange:" + exchange +
                                    ",deliveryTag:" + deliveryTag +
                                    ",message:" + message);
                }
            };
            channel.basicConsume("publish_queue1",true,consumer);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
//消费者2
public class Consume2 implements Runnable {
    public void run() {
        RabbitMQConUtil util = new RabbitMQConUtil("/firstVirtualhost");
        Connection connection;
        try {
            connection = util.getConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare("publish_queue2", true, false, false, null);
            channel.queueBind("publish_queue2","direct_exchange","log.info");
            channel.queueBind("publish_queue2","direct_exchange","log.error");
            channel.queueBind("publish_queue2","direct_exchange","log.warning");
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String routingKey = envelope.getRoutingKey();

                    String exchange = envelope.getExchange();

                    long deliveryTag = envelope.getDeliveryTag();

                    String message = new String(body, "utf-8");
                    System.out.println(
                            "routingKey:" + routingKey +
                                    ",exchange:" + exchange +
                                    ",deliveryTag:" + deliveryTag +
                                    ",message:" + message);
                }
            };
            channel.basicConsume("publish_queue2",true,consumer);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

Routing 模式要求队列在绑定交换机时要指定 routing key，消息会转发到符合routing key 的队列。

3.4 Topics通配符模式

alt Topic 类型与 Direct 相比，都是可以根据 RoutingKey 把消息路由到不同的队列。只不过Topic 类型Exchange可以让队列在绑定 Routing key 的时候使用通配符！ Routingkey 一般都是有一个或多个单词组成，多个单词之间以“ . ”分割，例如：item.insert

通配符规则：

  • #：匹配一个或多个词
  • *：匹配不多不少恰好 1 个词
举例：
item.#：能够匹配 item.insert.abc 或者 item.insert
item.*：只能匹配 item.insert

生产者，与路由模式类似

public class Product {
    public static void main(String[] args) throws Exception {
        RabbitMQConUtil util = new RabbitMQConUtil("/firstVirtualhost");
        Connection connection = util.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare("topic_exchange", BuiltinExchangeType.TOPIC);
        for (int i = 0; i < 5; i++) {
            String routing_key = "";
            String mes = "hello:"+i;
            switch (i){
                case 0:
                    routing_key = "log.info";
                    break;
                case 1:
                    routing_key = "log.error";
                    break;
                case 2:
                    routing_key = "log.warning";
                    break;
                case 3:
                    routing_key = "log.info.bat";
                    break;
                case 4:
                    routing_key = "log.info.txt";
                    break;
            }
            channel.basicPublish("topic_exchange",routing_key,null,mes.getBytes("UTF-8"));
        }
        channel.close();
        connection.close();
    }
}

消费者

//消费者1，log.*
public class Consume1 implements Runnable {
    public void run() {
        RabbitMQConUtil util = new RabbitMQConUtil("/firstVirtualhost");
        Connection connection;
        try {
            connection = util.getConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare("publish_queue1", true, false, false, null);
            channel.queueBind("publish_queue1","topic_exchange","log.*");
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String routingKey = envelope.getRoutingKey();
                    String exchange = envelope.getExchange();
                    long deliveryTag = envelope.getDeliveryTag();
                    String message = new String(body, "utf-8");
                    System.out.println(
                            "routingKey:" + routingKey +
                                    ",exchange:" + exchange +
                                    ",deliveryTag:" + deliveryTag +
                                    ",message:" + message);
                }
            };
            channel.basicConsume("publish_queue1",true,consumer);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
//消费者2 log.#
public class Consume2 implements Runnable {
    public void run() {
        RabbitMQConUtil util = new RabbitMQConUtil("/firstVirtualhost");
        Connection connection;
        try {
            connection = util.getConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare("publish_queue2", true, false, false, null);
            channel.queueBind("publish_queue2","topic_exchange","log.#");
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String routingKey = envelope.getRoutingKey();
                    String exchange = envelope.getExchange();
                    long deliveryTag = envelope.getDeliveryTag();
                    String message = new String(body, "utf-8");
                    System.out.println(
                            "routingKey:" + routingKey +
                                    ",exchange:" + exchange +
                                    ",deliveryTag:" + deliveryTag +
                                    ",message:" + message);
                }
            };
            channel.basicConsume("publish_queue2",true,consumer);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
//消费者3 log.info.*
public class Consume3 implements Runnable {
    public void run() {
        RabbitMQConUtil util = new RabbitMQConUtil("/firstVirtualhost");
        Connection connection;
        try {
            connection = util.getConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare("publish_queue3", true, false, false, null);
            channel.queueBind("publish_queue3","topic_exchange","log.info.*");
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String routingKey = envelope.getRoutingKey();
                    String exchange = envelope.getExchange();
                    long deliveryTag = envelope.getDeliveryTag();
                    String message = new String(body, "utf-8");
                    System.out.println(
                            "routingKey:" + routingKey +
                                    ",exchange:" + exchange +
                                    ",deliveryTag:" + deliveryTag +
                                    ",message:" + message);
                }
            };
            channel.basicConsume("publish_queue3",true,consumer);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

Topic 主题模式可以实现 Publish/Subscribe 发布订阅模式 和 Routing 路由模式的双重功能；只是Topic在配置 routing key 的时候可以使用通配符，显得更加灵活。

3.5 总结

  • 1、简单模式 HelloWorld 一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）
  • 2、工作队列模式 Work Queue 一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）
  • 3、发布订阅模式 Publish/subscribe 需要设置类型为 fanout 的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列
  • 4、路由模式 Routing 需要设置类型为 direct 的交换机，交换机和队列进行绑定，并且指定 routing key，当发送消息到交换机后，交换机会根据 routing key 将消息发送到对应的队列
  • 5、通配符模式 Topic 需要设置类型为 topic 的交换机，交换机和队列进行绑定，并且指定通配符方式的routing key，当发送消息到交换机后，交换机会根据 routing key 将消息发送到对应的队列

4 SpringBoot整合RabbitMQ

4.1 生产者

生成交换机、队列，绑定交换机与队列，发送消息到队列中

application.properties

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/firstVirtualhost

生成交换机、队列，绑定交换机与队列

@Configuration
public class ProductConfig {
    //创建交换机
    @Bean(name = "topicExchange")
    public TopicExchange topicExchange(){
        return new TopicExchange("topic_exchange_springboot");
    }
    //创建队列
    @Bean(name = "topicQueue")
    public Queue queue(){
        return QueueBuilder.durable("topic_queue_springboot").build();
    }
    //绑定交换机与队列
    @Bean
    public Binding binding(@Qualifier("topicQueue") Queue queue,
                           @Qualifier("topicExchange") Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("log.#").noargs();
    }
}

@RunWith(SpringRunner.class)
@SpringBootTest
public class ProductTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void test(){
        rabbitTemplate.convertAndSend("topic_exchange_springboot","log.info","发送了info消息");
        rabbitTemplate.convertAndSend("topic_exchange_springboot","log.error","发送了error消息");
        rabbitTemplate.convertAndSend("topic_exchange_springboot","log.warning","发送了warning消息");
    }
}

4.2 消费者

重新开启一个session

@Component
public class ConsumerConfig {
    @RabbitListener(queues = "topic_queue_springboot")
    public void listener(String msg){
        System.out.println(msg);
    }
}

5 RabbitMQ的重要问题

1、什么是rabbitmq

RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端的消息队列。

2、为什么要使用rabbitmq

  • 在分布式系统下具备异步,削峰,负载均衡等一系列高级功能;
  • 拥有持久化的机制，进程消息，队列中的信息也可以保存下来。
  • 实现消费者和生产者之间的解耦。
  • 对于高并发场景下，利用消息队列可以使得同步访问变为串行访问达到一定量的限流，利于数据库的操作。
  • 可以使用消息队列达到异步下单的效果，排队中，后台进行逻辑下单。

3、使用rabbitmq的场景

  • 服务间异步通信
  • 顺序消费
  • 定时任务
  • 请求削峰

4、如何确保消息正确地发送至RabbitMQ？ 如何确保消息接收方消费了消息？

发送方确认模式 将信道设置成confirm模式（发送方确认模式），则所有在信道上发布的消息都会被指派一个唯一的ID。一旦消息被投递到目的队列后，或者消息被写入磁盘后（可持久化的消息），信道会发送一个确认给生产者（包含消息唯一ID）。如果RabbitMQ发生内部错误从而导致消息丢失，会发送一条nack（not acknowledged，未确认）消息。发送方确认模式是异步的，生产者应用程序在等待确认的同时，可以继续发送消息。当确认消息到达生产者应用程序，生产者应用程序的回调方法就会被触发来处理确认消息。

接收方确认机制 接收方消息确认机制：消费者接收每一条消息后都必须进行确认（消息接收和消息确认是两个不同操作）。只有消费者确认了消息，RabbitMQ才能安全地把消息从队列中删除。 这里并没有用到超时机制，RabbitMQ仅通过Consumer的连接中断来确认是否需要重新发送消息。也就是说，只要连接不中断，RabbitMQ给了Consumer足够长的时间来处理消息。保证数据的最终一致性；

下面罗列几种特殊情况： 如果消费者接收到消息，在确认之前断开了连接或取消订阅，RabbitMQ会认为消息没有被分发，然后重新分发给下一个订阅的消费者。（可能存在消息重复消费的隐患，需要去重） 如果消费者接收到消息却没有确认消息，连接也未断开，则RabbitMQ认为该消费者繁忙，将不会给该消费者分发更多的消息。

5、如何避免消息重复投递或重复消费？

在消息生产时，MQ内部针对每条生产者发送的消息生成一个inner-msg-id，作为去重的依据（消息投递失败并重传），避免重复的消息进入队列； 在消息消费时，要求消息体中必须要有一个bizId（对于同一业务全局唯一，如支付ID、订单ID、帖子ID等）作为去重的依据，避免同一条消息被重复消费。

6、消息基于什么传输？

由于TCP连接的创建和销毁开销较大，且并发数受系统资源限制，会造成性能瓶颈。RabbitMQ使用信道的方式来传输数据。信道是建立在真实的TCP连接内的虚拟连接，且每条TCP连接上的信道数量没有限制。

7、消息如何分发？

若该队列至少有一个消费者订阅，消息将以循环（round-robin）的方式发送给消费者。每条消息只会分发给一个订阅的消费者（前提是消费者能够正常处理消息并进行确认）。通过路由可实现多消费的功能

8、消息怎么路由？

消息提供方->路由->一至多个队列，消息发布到交换器时，消息将拥有一个路由键（routing key），在消息创建时设定。通过队列路由键，可以把队列绑定到交换器上。 消息到达交换器后，RabbitMQ会将消息的路由键与队列的路由键进行匹配（针对不同的交换器有不同的路由规则）； 常用的交换器主要分为一下三种： fanout：如果交换器收到消息，将会广播到所有绑定的队列上 direct：如果路由键完全匹配，消息就被投递到相应的队列 topic：可以使来自不同源头的消息能够到达同一个队列。 使用topic交换器时，可以使用通配符

9、如何确保消息不丢失？

消息持久化，当然前提是队列必须持久化。RabbitMQ确保持久性消息能从服务器重启中恢复的方式是 ，将它们写入磁盘上的一个持久化日志文件， 当发布一条持久性消息到持久交换器上时，Rabbit会在消息提交到日志文件后才发送响应。一旦消费者从持久队列中消费了一条持久化消息，RabbitMQ会在持久化日志中把这条消息标记为等待垃圾收集。如果持久化消息在被消费之前RabbitMQ重启，那么Rabbit会自动重建交换器和队列（以及绑定），并重新发布持久化日志文件中的消息到合适的队列。

10、使用RabbitMQ有什么好处？

  • 服务间高度解耦
  • 异步通信性能高
  • 流量削峰

11、RabbitMQ的集群

Rabbitmq 集群

集群目的就是为了实现rabbitmq的高可用性，集群分为2种

普通集群：主备架构，只是实现主备方案，不至于主节点宕机导致整个服务无法使用 镜像集群：同步结构，基于普通集群实现的队列同步

普通集群

slave节点复制master节点的所有数据和状态，除了队列数据，队列数据只存在master节点，但是Rabbitmq slave节点可以实现队列的转发，也就是说消息消费者可以连接到slave节点，但是slave需要连接到master节点转发队列，由此说明只能保证了服务可以用，无法达到高可用 alt slave节点队列可以查看到，但是不会同步数据

镜像集群 基于普通集群实现队列的集群主从，消息会在集群中同步（至少三个节点） alt

12、mq的缺点

系统可用性降低 **系统引入的外部依赖越多，越容易挂掉，**本来你就是A系统调用BCD三个系统的接口就好了，本来ABCD四个系统好好的，没啥问题，你偏加个MQ进来，万一MQ挂了咋整？MQ挂了，整套系统崩溃了，你不就完了么。

系统复杂性提高 硬生生加个MQ进来，你怎么保证消息没有重复消费？怎么处理消息丢失的情况？怎么保证消息传递的顺序性？

一致性问题 A系统处理完了直接返回成功了，本来都以为你这个请求就成功了；但是问题是，BD两个系统写库成功了，结果C系统写库失败了，咋整？你这数据就不一致了。

全部评论

相关推荐

01-13 11:44
宁德时代_数据产品经理(准入职员工)
宁德时代内推宁德时代面经
宁德时代校招面经‌宁德时代热管理工程师面经面试时长：25分钟面试流程：无笔试，只有一面，英语测评，智力测评（两个测评应该不刷人，我都乱写的）面试官配置：只有一位，应该是经理面试内容：面试体验非常好，面试官教了我专业之外的很多东西。1.如果领导给你个任务，你有7天的时间，你怎么分配？2.电池液冷板的流量是怎么考虑的？3.实习过程中你学到了什么？4.你觉得实习和学校最大的区别是什么？反问：1.用的什么软件，以及怎么应对在学校用的仿真软件和工作的软件不一样这个转变？面试官说用的floefd，然后又说：其实软件的学习都是次要的，你们校招生进来会培训软件的使用，同时工作中也会天天用软件，你更应该关注的是...
宁德时代
|
校招
|
超多精选岗位
点赞 评论 收藏
分享
今天 13:13
苏州大学 Python
从绝望到重生：软件工程专业的我还能走出困境吗？
大家好，我是江苏某地域性211学校软件工程专业的一名学生，最近深感迷茫与绝望。回顾自己大学这几年的经历，我意识到自己似乎早已把自己困在了一个死胡同里，走到了一个无路可退的境地。或许这就是我应得的结果——没有科研、竞赛，甚至连实习的机会都没有，把自己活生生地困死在这片大学校园中。最令我痛心的，是自己一直以来深陷“学生思维”中，习惯了依赖老师，依赖课程，做事没有方向，没有动力，也没有真正参与到社会和行业的实际需求中去，纯纯在玩乐之中。现在稍微意识到自己已经陷入这种思维方式，感觉一切都太晚了。看着身边的同学们早早进入职场，参加各种项目，积累经验，我却依然徘徊在原地。每一次看到别人有了进展，而自己依旧...
点赞 评论 收藏
分享
2024-12-12 13:14
浙江工业大学 前端工程师
😥
三斤大芒果：切图仔过年回去天塌了
点赞 评论 收藏
分享
昨天 10:58
四川大学 Java
支fu宝出bug了，米不会追回
这个格局很大呀所有消费少花百分之20个人转账都减少可惜我当时在上班亏麻了呀
古怪_GG：个人转账都减少，这要是当时两个号转来转去，那不是左脚踩右脚上天了
点赞 评论 收藏
分享
评论
点赞
1
分享

全站热榜

更多

创作者周榜

更多
正在热议
更多
# 我的2024牛客高光时刻 #
88650次浏览 1490人参与
# 学信网能看师兄师姐就业去向了 #
188868次浏览 535人参与
# 被同事甩锅了怎么办 #
14815次浏览 89人参与
# 实习，投递多份简历没人回复怎么办 #
2686939次浏览 36536人参与
# 入职第三天，晒晒你的工位 #
17012次浏览 98人参与
# 实习工作，你找得还顺利吗？ #
282457次浏览 3645人参与
# 秋招你被哪家公司挂了? #
408730次浏览 3675人参与
# 正在实习的你，有转正机会吗？ #
327300次浏览 2669人参与
# 运营来爆料 #
16343次浏览 200人参与
# 数据人的面试交流地 #
493172次浏览 8275人参与
# 你上一次加班是什么时候？ #
31417次浏览 265人参与
# 硬件打工人的必备素养 #
11043次浏览 73人参与
# 简历中的项目经历要怎么写？ #
103663次浏览 1683人参与
# 想实习转正，又想准备秋招，我该怎么办 #
558950次浏览 5549人参与
# 非技术岗简历怎么写 #
174451次浏览 2581人参与
# 晒一晒你收到的礼盒 #
15830次浏览 173人参与
# 大家都开始春招面试了吗 #
492564次浏览 6572人参与
# 你上一天班会花多少钱 #
21689次浏览 152人参与
# 过年期间可能会经历的尴尬瞬间 #
16663次浏览 184人参与
# 腾讯求职进展汇总 #
264232次浏览 2083人参与
牛客网
牛客企业服务