[开发工具]RabbitMQ学习分目录整理万字笔记（有图有代码附带源码项目git）

RabbitMQ

项目源码git共享在 RabbitMQ项目源码git地址

RabbitMQ介绍

2007年，Rabbit技术公司基于AMQP标准开发的RabbitMQ1.0 发布。RabbitMQ采用Erlang语言开发。Erlang 语言由Ericson设计，专门为开发高并发和分布式系统的一种语言，在电信领域使用广泛。

rabbitmq基础架构图

Connection 连接（由连接工厂创建，连接工厂需要设置参数）

Channel 管道，用来通信

VirtualHost 虚拟主机

Exchange 交换机

Queue队列

RabbitMQ工作模式

最基本的单生产者单消费者队列模式

一个生产者，一个消费者，项目示范：单生产者单消费者项目使用示范

Work queues工作队列模式

一个生产者，若干消费者，项目示范：单生产者多消费者项目使用示范

• worker queues：与入门程序的简单模式相比，多了一个或者一些消费端，多个消费端消费同一个队列里的消息

• 应用场景：对于人物过重或者任务较多情况使用工作队列可以提高任务处理的速度

Pub/Sub订阅模式

一个生产者，多个队列，多消费者，项目示范：Pub/Sub模式项目使用示范

在订阅模型中，多了一个Exchange角色，而且过程略有变化:

P:生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X(交换机)

• C:消费者，消息的接收者，会一直等待消息到来
• Queue:消息队列，接收消息、缓存消息
• Exchange:交换机(X)。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有常见以下3种类型:

Fanout:广播，将消息交给所有绑定到交换机的队列
Direct:定向，把消息交给符合指定routing key 的队列
Topic:通配符，把消息交给符合routing pattern(路由模式)的队列

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

Routing路由模式

• 队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey(路由key)

• 消息的发送方在向Exchange 发送消息时，也必须指定消息的 RoutingKey

• Exchange 不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey 与消息的 Routing key完全—致，才会接收到消息

• P:生产者，向Exchange发送消息，发送消息时，会指定一个routing key

• X: Exchange(交换机)，接收生产者的消息，然后把消息递交给与routing key完全匹配的队列

• C1:消费者，其所在队列指定了需要routing key为error的消息

• C2:消费者，其所在队列指定了需要routing key为info、error、warning 的消息

Topic通配符模式

同一交换机下，只要发的消息的routingkey能和绑定的队列的通配符能匹配上这个消息就会被送进队列，项目示范：Topic模式项目使用示范

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

下载和安装RabbitMQ

安装docker

为了屏蔽环境，操作系统不同，我使用docker，没有docker直接去菜鸟教程的docker安装教程看，有自动安装的命令，这里我摘取centos和ubuntu的。

# centos | ubuntu 都是一样的命令
curl -fsSL https://get.docker.com | bash -s docker --mirror Aliyun

启动和将docker加入开机启动的服务

# 启动docker
systemctl start docker
# 启用服务
systemctl enable docker

拉取镜像并启动容器

# 拉取镜像（后面带-management的是带管理界面的）
docker pull rabbitmq:3.6.5-management
# 启动容器(设置后台账户密码为admin了)  5672队列服务的端口 15672是管理控制台的端口  4369和25672是后续集群通讯的端口
docker run -dit --name rabbitmq -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 5672:5672 -p 15672:15672 -p 4369:4369 -p 25672:25672 rabbitmq:3.6.5-management

访问管理后台

我的linux是个虚拟机放，IP是192.168.0.105，你自己的就用127.0.0.1就可以了，rabbitmq服务的端口是5672，后台的端口是15672。

单生产者单消费者项目使用示范

创建生产者

创建的是maven项目

pom.xml

<?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">
    <parent>
        <artifactId>rabbitmq-demo</artifactId>
        <groupId>org.example</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>producer</artifactId>
    <dependencies>
<!--        rabbitmq 客户端依赖-->
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.6.0</version>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

Producer.java

package com.example.producer;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @Author:humorchen
 * @Date 2021/9/25 9:33
 */
public class Producer {
    public static void main(String[] args) {
        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.0.105");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        connectionFactory.setVirtualHost("/");
        try {
            //创建连接
            Connection connection = connectionFactory.newConnection();
            //创建Channel
            Channel channel = connection.createChannel();
            //创建队列
            //queueDeclare(String var1, boolean var2, boolean var3, boolean var4, Map<String, Object> var5)
            //              队列名        是否持久化     是否独占(单消费者监听)  当没有消费者时是否自动删除  参数信息
            channel.queueDeclare("first_queue",true,false,false,null);
            //发送消息到队列
            // basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
            //                交换机                路由键          配置参数               数据
            String msg = "hello rabbitmq,this is my first  message.";
            channel.basicPublish("","first_queue",null,msg.getBytes());
            //释放资源
            channel.close();
            connection.close();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

运行结果

去控制面板查看

点开first_queue，再点Get messages里的Get Message(s)

创建消费者

也是maven项目

pom.xml

<?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">
    <parent>
        <artifactId>rabbitmq-demo</artifactId>
        <groupId>org.example</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>consumer</artifactId>

    <dependencies>
        <!--        rabbitmq 客户端依赖-->
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.6.0</version>
        </dependency>
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.9.0</version>
            <scope>compile</scope>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

ConsumerDemo.java

package com.example.consumer;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @Author:humorchen
 * @Date 2021/9/25 10:07
 */
public class ConsumerDemo {
    public static void main(String[] args) {

        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.0.105");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        connectionFactory.setVirtualHost("/");
        try {
            //创建连接
            Connection connection = connectionFactory.newConnection();
            //创建Channel
            Channel channel = connection.createChannel();
            //创建队列(消费端可以不写的)
            //queueDeclare(String var1, boolean var2, boolean var3, boolean var4, Map<String, Object> var5)
            //              队列名        是否持久化     是否独占(单消费者监听)  当没有消费者时是否自动删除  参数信息
//            channel.queueDeclare("first_queue",true,false,false,null);
            //从队列拿消息
            // basicConsume(String queue, boolean autoAck, Consumer callback)
            //               队列名称      是否自动确认       消费对象
            Consumer consumer = new DefaultConsumer(channel){
                /**
                 * 收到消息的回调方法
                 * @param consumerTag 标识
                 * @param envelope  获取一些信息，如交换机，路由key
                 * @param properties 获取一些配置信息
                 * @param body 消息主体
                 * @throws IOException
                 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("consumerTag: "+consumerTag);
                    System.out.println("env: "+envelope.toString());
                    System.out.println("消费者收到消息：");
                    System.out.println(new String(body));
                }
            };
            //开启监听，消费消息
            channel.basicConsume("first_queue",true,consumer);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

运行结果

单生产者多消费者项目使用示范

创建生产者

给之前的代码加个for循环执行，并加一个sleep即可

SingleLoopProducer.java

package com.example.producer;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.Date;
import java.util.concurrent.TimeoutException;

/**
 * @Author:humorchen
 * @Date 2021/9/25 9:33
 */
public class SingleLoopProducer {
    public static void main(String[] args) {
        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.0.105");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        connectionFactory.setVirtualHost("/");
        try {
            //创建连接
            Connection connection = connectionFactory.newConnection();
            //创建Channel
            Channel channel = connection.createChannel();
            //创建队列
            //queueDeclare(String var1, boolean var2, boolean var3, boolean var4, Map<String, Object> var5)
            //              队列名        是否持久化     是否独占(单消费者监听)  当没有消费者时是否自动删除  参数信息
            channel.queueDeclare("first_queue",true,false,false,null);
            for (;;){
                try {
                    //发送消息到队列
                    // basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
                    //                交换机                路由键          配置参数               数据
                    String msg = "hello rabbitmq,this is my first  message. "+new Date().toString();
                    channel.basicPublish("","first_queue",null,msg.getBytes());
                    //每隔一段时间发一个
                    Thread.sleep(1500);
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

创建消费者

SingleLoopConsumerDemo1.java 消费者2号

package com.example.consumer;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @Author:humorchen
 * @Date 2021/9/25 10:07
 */
public class SingleLoopConsumerDemo1 {
    public static void main(String[] args) {

        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.0.105");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        connectionFactory.setVirtualHost("/");
        try {
            //创建连接
            Connection connection = connectionFactory.newConnection();
            //创建Channel
            Channel channel = connection.createChannel();
            //创建队列(消费端可以不写的)
            //queueDeclare(String var1, boolean var2, boolean var3, boolean var4, Map<String, Object> var5)
            //              队列名        是否持久化     是否独占(单消费者监听)  当没有消费者时是否自动删除  参数信息
//            channel.queueDeclare("first_queue",true,false,false,null);
            //从队列拿消息
            // basicConsume(String queue, boolean autoAck, Consumer callback)
            //               队列名称      是否自动确认       消费对象
            Consumer consumer = new DefaultConsumer(channel){
                /**
                 * 收到消息的回调方法
                 * @param consumerTag 标识
                 * @param envelope  获取一些信息，如交换机，路由key
                 * @param properties 获取一些配置信息
                 * @param body 消息主体
                 * @throws IOException
                 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("消费者1号收到消息：");
                    System.out.println("consumerTag: "+consumerTag);
                    System.out.println("env: "+envelope.toString());
                    System.out.println("消息主体：");
                    System.out.println(new String(body));
                }
            };
            //开启监听，消费消息
            channel.basicConsume("first_queue",true,consumer);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

SingleLoopConsumerDemo2.java 消费者2号

package com.example.consumer;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @Author:humorchen
 * @Date 2021/9/25 10:07
 */
public class SingleLoopConsumerDemo2 {
    public static void main(String[] args) {

        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.0.105");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        connectionFactory.setVirtualHost("/");
        try {
            //创建连接
            Connection connection = connectionFactory.newConnection();
            //创建Channel
            Channel channel = connection.createChannel();
            //创建队列(消费端可以不写的)
            //queueDeclare(String var1, boolean var2, boolean var3, boolean var4, Map<String, Object> var5)
            //              队列名        是否持久化     是否独占(单消费者监听)  当没有消费者时是否自动删除  参数信息
//            channel.queueDeclare("first_queue",true,false,false,null);
            //从队列拿消息
            // basicConsume(String queue, boolean autoAck, Consumer callback)
            //               队列名称      是否自动确认       消费对象
            Consumer consumer = new DefaultConsumer(channel){
                /**
                 * 收到消息的回调方法
                 * @param consumerTag 标识
                 * @param envelope  获取一些信息，如交换机，路由key
                 * @param properties 获取一些配置信息
                 * @param body 消息主体
                 * @throws IOException
                 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("消费者2号收到消息：");
                    System.out.println("consumerTag: "+consumerTag);
                    System.out.println("env: "+envelope.toString());
                    System.out.println("消息主体：");
                    System.out.println(new String(body));
                }
            };
            //开启监听，消费消息
            channel.basicConsume("first_queue",true,consumer);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

运行结果

启动两个消费者，再启动生产者

可以看到两个消费者不断的在消费队列里的消息，并且不会重复消费。

Pub/Sub模式项目使用示范

创建pub生产者端

PubProducer.java

package com.example.producer;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @Author:humorchen
 * @Date 2021/9/25 16:38
 */
public class PubProducer {
    public static void main(String[] args) {
        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.0.105");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        connectionFactory.setVirtualHost("/");
        try {
            //创建连接
            Connection connection = connectionFactory.newConnection();
            //创建Channel
            Channel channel = connection.createChannel();
            //创建交换机
            // exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
            //                  交换机名字           交换机类型             是否持久化            是否自动删除          是否内部             参数，直接null
            //交换机类型
            //    DIRECT("direct"),  定向
            //    FANOUT("fanout"),  扇形广播
            //    TOPIC("topic"),    通配符的方式
            //    HEADERS("headers"); 参数（不常用）
            String exchangeName = "pub_demo_exchange";
            channel.exchangeDeclare(exchangeName,BuiltinExchangeType.FANOUT,true,true,false,null);

            //创建队列
            //queueDeclare(String var1, boolean var2, boolean var3, boolean var4, Map<String, Object> var5)
            //              队列名        是否持久化     是否独占(单消费者监听)  当没有消费者时是否自动删除  参数信息
            String queue1Name = "pub_demo_queue1";
            String queue2Name = "pub_demo_queue2";
            channel.queueDeclare(queue1Name,true,false,false,null);
            channel.queueDeclare(queue2Name,true,false,false,null);

            //队列绑定到交换机
            // queueBind(String queue, String exchange, String routingKey, Map<String, Object> arguments)
            //            队列名字            交换机名字     路由键                    参数
            channel.queueBind(queue1Name,exchangeName,"",null);
            channel.queueBind(queue2Name,exchangeName,"",null);

            String msg = "hello ,send to exchange "+exchangeName;
            channel.basicPublish(exchangeName,"",null,msg.getBytes());
            //释放资源
            channel.close();
            connection.close();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

运行结果

创建队列1的消费者

PubSubConsumerDemo1.java

package com.example.consumer;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @Author:humorchen
 * @Date 2021/9/25 10:07
 */
public class PubSubConsumerDemo1 {
    public static void main(String[] args) {

        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.0.105");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        connectionFactory.setVirtualHost("/");
        String queueName = "pub_demo_queue1";
        try {
            //创建连接
            Connection connection = connectionFactory.newConnection();
            //创建Channel
            Channel channel = connection.createChannel();
            //创建队列(消费端可以不写的)
            //queueDeclare(String var1, boolean var2, boolean var3, boolean var4, Map<String, Object> var5)
            //              队列名        是否持久化     是否独占(单消费者监听)  当没有消费者时是否自动删除  参数信息
//            channel.queueDeclare("first_queue",true,false,false,null);
            //从队列拿消息
            // basicConsume(String queue, boolean autoAck, Consumer callback)
            //               队列名称      是否自动确认       消费对象
            Consumer consumer = new DefaultConsumer(channel){
                /**
                 * 收到消息的回调方法
                 * @param consumerTag 标识
                 * @param envelope  获取一些信息，如交换机，路由key
                 * @param properties 获取一些配置信息
                 * @param body 消息主体
                 * @throws IOException
                 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("消费者1号从队列"+queueName+"收到消息：");
                    System.out.println("consumerTag: "+consumerTag);
                    System.out.println("env: "+envelope.toString());
                    System.out.println("消息主体：");
                    System.out.println(new String(body));
                }
            };
            //开启监听，消费消息
            channel.basicConsume(queueName,true,consumer);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

运行结果

创建队列2的消费者

PubSubConsumerDemo2.java

package com.example.consumer;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @Author:humorchen
 * @Date 2021/9/25 10:07
 */
public class PubSubConsumerDemo2 {
    public static void main(String[] args) {

        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.0.105");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        connectionFactory.setVirtualHost("/");
        String queueName = "pub_demo_queue2";
        try {
            //创建连接
            Connection connection = connectionFactory.newConnection();
            //创建Channel
            Channel channel = connection.createChannel();
            //创建队列(消费端可以不写的)
            //queueDeclare(String var1, boolean var2, boolean var3, boolean var4, Map<String, Object> var5)
            //              队列名        是否持久化     是否独占(单消费者监听)  当没有消费者时是否自动删除  参数信息
//            channel.queueDeclare("first_queue",true,false,false,null);
            //从队列拿消息
            // basicConsume(String queue, boolean autoAck, Consumer callback)
            //               队列名称      是否自动确认       消费对象
            Consumer consumer = new DefaultConsumer(channel){
                /**
                 * 收到消息的回调方法
                 * @param consumerTag 标识
                 * @param envelope  获取一些信息，如交换机，路由key
                 * @param properties 获取一些配置信息
                 * @param body 消息主体
                 * @throws IOException
                 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("消费者2号从队列"+queueName+"收到消息：");
                    System.out.println("consumerTag: "+consumerTag);
                    System.out.println("env: "+envelope.toString());
                    System.out.println("消息主体：");
                    System.out.println(new String(body));
                }
            };
            //开启监听，消费消息
            channel.basicConsume(queueName,true,consumer);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

运行结果

Routing模式项目使用示范

创建生产者

RoutingProducer.java

package com.example.producer;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @Author:humorchen
 * @Date 2021/9/25 16:38
 */
public class RoutingProducer {
    public static void main(String[] args) {
        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.0.105");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        connectionFactory.setVirtualHost("/");
        try {
            //创建连接
            Connection connection = connectionFactory.newConnection();
            //创建Channel
            Channel channel = connection.createChannel();
            //创建交换机
            // exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
            //                  交换机名字           交换机类型             是否持久化            是否自动删除          是否内部             参数，直接null
            //交换机类型
            //    DIRECT("direct"),  定向
            //    FANOUT("fanout"),  扇形广播
            //    TOPIC("topic"),    通配符的方式
            //    HEADERS("headers"); 参数（不常用）
            String exchangeName = "routing_demo_exchange";
            channel.exchangeDeclare(exchangeName,BuiltinExchangeType.DIRECT,true,true,false,null);

            //创建队列
            //queueDeclare(String var1, boolean var2, boolean var3, boolean var4, Map<String, Object> var5)
            //              队列名        是否持久化     是否独占(单消费者监听)  当没有消费者时是否自动删除  参数信息
            String queue1Name = "routing_demo_queue1";
            String queue2Name = "routing_demo_queue2";
            channel.queueDeclare(queue1Name,true,false,false,null);
            channel.queueDeclare(queue2Name,true,false,false,null);

            //队列绑定到交换机
            // queueBind(String queue, String exchange, String routingKey, Map<String, Object> arguments)
            //            队列名字            交换机名字     路由键                    参数
            channel.queueBind(queue1Name,exchangeName,"error",null);

            channel.queueBind(queue2Name,exchangeName,"debug",null);
            channel.queueBind(queue2Name,exchangeName,"info",null);
            channel.queueBind(queue2Name,exchangeName,"warning",null);

            String msg = "hello ,send to exchange "+exchangeName+" level: error";
            channel.basicPublish(exchangeName,"error",null,msg.getBytes());
            //释放资源
            channel.close();
            connection.close();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

运行结果

创建消费者1消费routingkey=“error”

RoutingConsumerDemo1.java

package com.example.consumer;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @Author:humorchen
 * @Date 2021/9/25 10:07
 */
public class RoutingConsumerDemo1 {
    public static void main(String[] args) {

        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.0.105");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        connectionFactory.setVirtualHost("/");
        String queueName = "routing_demo_queue1";
        try {
            //创建连接
            Connection connection = connectionFactory.newConnection();
            //创建Channel
            Channel channel = connection.createChannel();
            //创建队列(消费端可以不写的)
            //queueDeclare(String var1, boolean var2, boolean var3, boolean var4, Map<String, Object> var5)
            //              队列名        是否持久化     是否独占(单消费者监听)  当没有消费者时是否自动删除  参数信息
//            channel.queueDeclare("first_queue",true,false,false,null);
            //从队列拿消息
            // basicConsume(String queue, boolean autoAck, Consumer callback)
            //               队列名称      是否自动确认       消费对象
            Consumer consumer = new DefaultConsumer(channel){
                /**
                 * 收到消息的回调方法
                 * @param consumerTag 标识
                 * @param envelope  获取一些信息，如交换机，路由key
                 * @param properties 获取一些配置信息
                 * @param body 消息主体
                 * @throws IOException
                 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("消费者1号从队列"+queueName+"收到消息：");
                    System.out.println("consumerTag: "+consumerTag);
                    System.out.println("env: "+envelope.toString());
                    System.out.println("消息主体：");
                    System.out.println(new String(body));
                }
            };
            //开启监听，消费消息
            channel.basicConsume(queueName,true,consumer);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

运行结果

创建消费者2消费routingkey为debug info error

RoutingConsumerDemo2.java

package com.example.consumer;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @Author:humorchen
 * @Date 2021/9/25 10:07
 */
public class RoutingConsumerDemo2 {
    public static void main(String[] args) {

        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.0.105");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        connectionFactory.setVirtualHost("/");
        String queueName = "routing_demo_queue2";
        try {
            //创建连接
            Connection connection = connectionFactory.newConnection();
            //创建Channel
            Channel channel = connection.createChannel();
            //创建队列(消费端可以不写的)
            //queueDeclare(String var1, boolean var2, boolean var3, boolean var4, Map<String, Object> var5)
            //              队列名        是否持久化     是否独占(单消费者监听)  当没有消费者时是否自动删除  参数信息
//            channel.queueDeclare("first_queue",true,false,false,null);
            //从队列拿消息
            // basicConsume(String queue, boolean autoAck, Consumer callback)
            //               队列名称      是否自动确认       消费对象
            Consumer consumer = new DefaultConsumer(channel){
                /**
                 * 收到消息的回调方法
                 * @param consumerTag 标识
                 * @param envelope  获取一些信息，如交换机，路由key
                 * @param properties 获取一些配置信息
                 * @param body 消息主体
                 * @throws IOException
                 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("消费者2号从队列"+queueName+"收到消息：");
                    System.out.println("consumerTag: "+consumerTag);
                    System.out.println("env: "+envelope.toString());
                    System.out.println("消息主体：");
                    System.out.println(new String(body));
                }
            };
            //开启监听，消费消息
            channel.basicConsume(queueName,true,consumer);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

运行结果

Topic模式项目使用示范

创建生产者

TopicProducer.java

package com.example.producer;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @Author:humorchen
 * @Date 2021/9/25 16:38
 */
public class TopicProducer {
    public static void main(String[] args) {
        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.0.105");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        connectionFactory.setVirtualHost("/");
        try {
            //创建连接
            Connection connection = connectionFactory.newConnection();
            //创建Channel
            Channel channel = connection.createChannel();
            //创建交换机
            // exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
            //                  交换机名字           交换机类型             是否持久化            是否自动删除          是否内部             参数，直接null
            //交换机类型
            //    DIRECT("direct"),  定向
            //    FANOUT("fanout"),  扇形广播
            //    TOPIC("topic"),    通配符的方式
            //    HEADERS("headers"); 参数（不常用）
            String exchangeName = "topic_demo_exchange";
            channel.exchangeDeclare(exchangeName,BuiltinExchangeType.TOPIC,true,true,false,null);

            //创建队列
            //queueDeclare(String var1, boolean var2, boolean var3, boolean var4, Map<String, Object> var5)
            //              队列名        是否持久化     是否独占(单消费者监听)  当没有消费者时是否自动删除  参数信息
            String queue1Name = "topic_demo_queue1";
            String queue2Name = "topic_demo_queue2";
            channel.queueDeclare(queue1Name,true,false,false,null);
            channel.queueDeclare(queue2Name,true,false,false,null);

            //队列绑定到交换机
            // queueBind(String queue, String exchange, String routingKey, Map<String, Object> arguments)
            //            队列名字            交换机名字     路由键                    参数
            // #一个或者多个词  * 匹配一个词  (注意，匹配的是词！)
            //和日志有关的
            channel.queueBind(queue1Name,exchangeName,"*.log",null);

            //和订单有关的
            channel.queueBind(queue2Name,exchangeName,"order.#",null);

            String msg = "hello ,this is an order data,send to exchange "+exchangeName;
            channel.basicPublish(exchangeName,"order.log",null,msg.getBytes());
            //释放资源
            channel.close();
            connection.close();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

运行结果

创建日志消费者

TopicConsumerDemo1.java

package com.example.consumer;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @Author:humorchen
 * @Date 2021/9/25 10:07
 */
public class TopicConsumerDemo1 {
    public static void main(String[] args) {

        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.0.105");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        connectionFactory.setVirtualHost("/");
        String queueName = "topic_demo_queue1";
        try {
            //创建连接
            Connection connection = connectionFactory.newConnection();
            //创建Channel
            Channel channel = connection.createChannel();
            //创建队列(消费端可以不写的)
            //queueDeclare(String var1, boolean var2, boolean var3, boolean var4, Map<String, Object> var5)
            //              队列名        是否持久化     是否独占(单消费者监听)  当没有消费者时是否自动删除  参数信息
//            channel.queueDeclare("first_queue",true,false,false,null);
            //从队列拿消息
            // basicConsume(String queue, boolean autoAck, Consumer callback)
            //               队列名称      是否自动确认       消费对象
            Consumer consumer = new DefaultConsumer(channel){
                /**
                 * 收到消息的回调方法
                 * @param consumerTag 标识
                 * @param envelope  获取一些信息，如交换机，路由key
                 * @param properties 获取一些配置信息
                 * @param body 消息主体
                 * @throws IOException
                 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("消费者1号从队列"+queueName+"收到消息：");
                    System.out.println("consumerTag: "+consumerTag);
                    System.out.println("env: "+envelope.toString());
                    System.out.println("消息主体：");
                    System.out.println(new String(body));
                }
            };
            //开启监听，消费消息
            channel.basicConsume(queueName,true,consumer);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

运行结果

创建订单消费者

TopicConsumerDemo2.java

package com.example.consumer;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @Author:humorchen
 * @Date 2021/9/25 10:07
 */
public class TopicConsumerDemo2 {
    public static void main(String[] args) {

        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.0.105");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        connectionFactory.setVirtualHost("/");
        String queueName = "topic_demo_queue2";
        try {
            //创建连接
            Connection connection = connectionFactory.newConnection();
            //创建Channel
            Channel channel = connection.createChannel();
            //创建队列(消费端可以不写的)
            //queueDeclare(String var1, boolean var2, boolean var3, boolean var4, Map<String, Object> var5)
            //              队列名        是否持久化     是否独占(单消费者监听)  当没有消费者时是否自动删除  参数信息
//            channel.queueDeclare("first_queue",true,false,false,null);
            //从队列拿消息
            // basicConsume(String queue, boolean autoAck, Consumer callback)
            //               队列名称      是否自动确认       消费对象
            Consumer consumer = new DefaultConsumer(channel){
                /**
                 * 收到消息的回调方法
                 * @param consumerTag 标识
                 * @param envelope  获取一些信息，如交换机，路由key
                 * @param properties 获取一些配置信息
                 * @param body 消息主体
                 * @throws IOException
                 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("消费者1号从队列"+queueName+"收到消息：");
                    System.out.println("consumerTag: "+consumerTag);
                    System.out.println("env: "+envelope.toString());
                    System.out.println("消息主体：");
                    System.out.println(new String(body));
                }
            };
            //开启监听，消费消息
            channel.basicConsume(queueName,true,consumer);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

运行结果

Springboot 整合RabbitMQ

创建生产者

项目结构图

创建一个springboot-rabbitmq-producer模块,是一个springboot项目

pom.xml

<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.example</groupId>
    <artifactId>springboot-rabbitmq-producer</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>springboot-rabbitmq-producer</name>
    <description>Demo project for Spring Boot</description>

    <properties>
        <java.version>1.8</java.version>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <spring-boot.version>2.3.7.RELEASE</spring-boot.version>
    </properties>

    <dependencies>
        <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-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-configuration-processor</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-dependencies</artifactId>
                <version>${spring-boot.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <version>2.3.7.RELEASE</version>
                <configuration>
                    <mainClass>com.example.springbootrabbitmqproducer.SpringbootRabbitmqProducerApplication</mainClass>
                </configuration>
                <executions>
                    <execution>
                        <id>repackage</id>
                        <goals>
                            <goal>repackage</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

</project>

application.yml

spring:
  application:
    name: springboot-rabbitmq-producer
  rabbitmq:
    host: 192.168.0.105
    port: 5672
    virtual-host: /
    username: admin
    password: admin
server:
  port: 8001

rabbitmq配置类

package com.example.springbootrabbitmqproducer.config;

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

/**
 * RabbitMQ 配置类
 * @Author:humorchen
 * @Date 2021/9/26 17:44
 */
@Configuration
public class RabbitmqConfig {
    public static final String EXCHANGE_NAME = "springboot_topic_exchange";
    public static final String QUEUE_NAME = "springboot_queue";
    // 交换机创建
    //队列创建
    //队列交换机绑定


    /**
     * 配置一个topic类型的交换机并命名
     * 启用持久化
     * @return
     */
    @Bean(EXCHANGE_NAME)
    public Exchange exchange(){
        return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();
    }

    /**
     * 配置一个持久化的队列并命名
     * @return
     */
    @Bean(QUEUE_NAME)
    public Queue queue(){
        return QueueBuilder.durable(QUEUE_NAME).build();
    }

    /**
     * 队列绑定到交换机
     * 注入队列和交换机
     * @param queue
     * @param exchange
     * @return
     */
    @Bean
    public Binding binding(@Qualifier(QUEUE_NAME) Queue queue,@Qualifier(EXCHANGE_NAME)  Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("springboot.#").noargs();
    }
}

测试运行

package com.example.springbootrabbitmqproducer;

import com.example.springbootrabbitmqproducer.config.RabbitmqConfig;
import org.junit.jupiter.api.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@SpringBootTest
@RunWith(SpringRunner.class)
class SpringbootRabbitmqProducerApplicationTests {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 测试使用rabbitmq发送消息
     */
    @Test
    public void testSend(){
        rabbitTemplate.convertAndSend(RabbitmqConfig.EXCHANGE_NAME,"springboot.hello","hello,springboot rabbitmq!");
    }

}

运行结果

创建消费者

pom.xml

<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.example</groupId>
    <artifactId>springboot-rabbitmq-consumer</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>springboot-rabbitmq-consumer</name>
    <description>Demo project for Spring Boot</description>

    <properties>
        <java.version>1.8</java.version>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <spring-boot.version>2.3.7.RELEASE</spring-boot.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-configuration-processor</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-dependencies</artifactId>
                <version>${spring-boot.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <version>2.3.7.RELEASE</version>
                <configuration>
                    <mainClass>com.example.springbootrabbitmqconsumer.SpringbootRabbitmqConsumerApplication</mainClass>
                </configuration>
                <executions>
                    <execution>
                        <id>repackage</id>
                        <goals>
                            <goal>repackage</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

</project>

application.yml

# 应用名称
spring:
  application:
    name: springboot-rabbitmq-consumer
  rabbitmq:
    host: 192.168.0.105
    port: 5672
    virtual-host: /
    username: admin
    password: admin
server:
  port: 8001

监听队列类

package com.example.springbootrabbitmqconsumer.listener;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @Author:humorchen
 * @Date 2021/9/26 18:15
 */
@Component
public class RabbitmqListener {
    public static final String QUEUE_NAME = "springboot_queue";

    /**
     * 定义对队列进行监听
     * @param message
     */
    @RabbitListener(queues = QUEUE_NAME)
    public void listenerQueue(Message message){
        System.out.println("监听方法收到了消息：");
        System.out.println(message);
    }
}

测试运行

监听方法收到了消息：
(Body:'hello,springboot rabbitmq!' MessageProperties [headers={}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=true, receivedExchange=springboot_topic_exchange, receivedRoutingKey=springboot.hello, deliveryTag=1, consumerTag=amq.ctag-rwLRHnLKMiETHNnk1ZAwDw, consumerQueue=springboot_queue])

RabbitMQ高级特性

消息可靠性投递

在使用RabbitMQ的时候，作为消息发送方希望杜绝任何消息丢失或者投递失败场景。RabbitMQ为我们提供了两种方式用来控制消息的投递可靠性模式。

• confirm 确认模式
• return 退回模式

rabbitmq整个消息投递的路径为:

producer—>rabbitmq broker—>exchange- -->queue—>consumer

消息从producer 到 exchange 则会返回一个confirmCallback .

消息从exchange–>queue投递失败则会返回一个returnCallback 。我们将利用这两个callback 控制消息的可靠性投递

• 设置ConnectionFactory的publisher-confirms="true”开启确认模式。

• 使用rabbitTemplate.setConfirmCallback设置回调函数。当消息发送到exchange后回调confirm方法。在方法中判断ack，如果为true，则发送成功，如果为false，则发送失败，需要处理。

• 设置ConnectionFactory的publisher-returns="true"开启退回模式。

• 使用rabbitTemplate.setReturnCallback设置退回函数，当消息从exchange路由到queue失败后，如果设置了rabbitTemplate.setMandatory(true)参数，则会将消息退回给producer。并执行回调函数returnedMessage.

• 在RabbitMQ中也提供了事务机制，但是性能较差，此处不做讲解。使用channel下列方法，完成事务控制:

  • txSelect(),用于将当前channel设置成transaction模式
  • txCommit()，用于提交事务
  • txRollback(),用于回滚事务

confirm确认模式

# 旧版配置
publisher-confirms: true
# 新版配置
publisher-confirm-type: correlated

创建一个子项目springboot-rabbitmq-producer-confirm,复制一份之前项目的应用配置

pom.xml

<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.example</groupId>
    <artifactId>springboot-rabbitmq-producer-confirm</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>springboot-rabbitmq-producer-confirm</name>
    <description>Demo project for Spring Boot</description>

    <properties>
        <java.version>1.8</java.version>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <spring-boot.version>2.3.7.RELEASE</spring-boot.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-configuration-processor</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-dependencies</artifactId>
                <version>${spring-boot.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <version>2.3.7.RELEASE</version>
                <configuration>
                    <mainClass>
                        com.example.springbootrabbitmqproducerconfirm.SpringbootRabbitmqProducerConfirmApplication
                    </mainClass>
                </configuration>
                <executions>
                    <execution>
                        <id>repackage</id>
                        <goals>
                            <goal>repackage</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

</project>

application.yml

spring:
  application:
    name: springboot-rabbitmq-producer
  rabbitmq:
    host: 192.168.0.105
    port: 5672
    virtual-host: /
    username: admin
    password: admin
    # 启用消息确认
    publisher-confirm-type: correlated

测试类

package com.example.springbootrabbitmqproducerconfirm;

import com.example.springbootrabbitmqproducerconfirm.config.RabbitmqConfig;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest

class SpringbootRabbitmqProducerConfirmApplicationTests {


    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testConfirmSend(){
        //声明确认的回调函数
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
             * 确认函数
             * @param correlationData  相关数据（自己可以设置）
             * @param ack 交换机是否收到
             * @param s 错误原因
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String s) {
                System.out.println("收到投递反馈");
                System.out.println(correlationData);
                if (ack){
                    //交换机收到了
                    System.out.println("消息投递成功");
                }else {
                    System.out.println("消息投递失败，错误原因：");
                    System.out.println(s);
                }
            }
        });
        String msgId = "1";
        String msg = "hello,this message is a test for sending  message efficiently,i expect receive a confirm from exchange.";
        rabbitTemplate.convertAndSend(RabbitmqConfig.QUEUE_NAME,msg.getBytes(),new CorrelationData(msgId));
    }
}

运行结果

return退回模式

跟上面一样再次创建一个测试用的项目

application.yml

spring:
  application:
    name: springboot-rabbitmq-producer
  rabbitmq:
    host: 192.168.0.105
    port: 5672
    virtual-host: /
    username: admin
    password: admin
    # 启用消息投递失败回退
    publisher-returns: true

测试类

package com.example.springbootrabbitmqproducerreturn;

import com.example.springbootrabbitmqproducerreturn.config.RabbitmqConfig;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
class SpringbootRabbitmqProducerReturnApplicationTests {


    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testReturnSend() {
        //也可以通过代码设置交换机模式为return模式 投递失败退回
        //rabbitTemplate.setMandatory(true);
        //设置消息回退函数
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            /**
             *
             * @param message 消息体
             * @param replyCode 回复码
             * @param replyText 回复文本
             * @param exchange 交换机
             * @param routingKey 路由键
             */
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                System.out.println("消息投递失败，调用了回退函数 replyCode：" + replyText + " replyText：" + replyText + " exchange：" + exchange + " routingKey：" + routingKey);
                System.out.println(message);
            }
        });
        String msg = "hello,this message is a test for sending  message efficiently,if message send failed,i will receive a return callback.";
        //故意给队列名后面加个not，这个队列我们是没有定义的
        rabbitTemplate.convertAndSend(RabbitmqConfig.QUEUE_NAME + "not", msg);
    }

}

运行结果

消费者Ack消费可靠性

自动确认(默认)

手动确认

根据异常情况确认（不常用）

ack指Acknowledge，确认。表示消费端收到消息后的确认方式。有三种确认方式:

• 自动确认: acknowledge=“none”

• 手动确认: acknowledge=“manual”

• 根据异常情况确认: acknowledge=“auto”，(这种方式使用麻烦，不作讲解)

其中自动确认是指，当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从 RabbitMQ的消息缓存中移除。但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么该消息就会丢失。

如果设置了手动确认方式，则需要在业务处理成功后，调用channel.basicAck)，手动签收，

如果出现异常，则调用channel.basicNack)方法，让其自动重新发送消息。

创建项目springboot-rabbitmq-consumer-ack

pom.xml

<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.example</groupId>
    <artifactId>springboot-rabbitmq-consumer-ack</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>springboot-rabbitmq-consumer-ack</name>
    <description>Demo project for Spring Boot</description>

    <properties>
        <java.version>1.8</java.version>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <spring-boot.version>2.3.7.RELEASE</spring-boot.version>
    </properties>

    <dependencies>
        <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-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-configuration-processor</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-dependencies</artifactId>
                <version>${spring-boot.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <version>2.3.7.RELEASE</version>
                <configuration>
                    <mainClass>com.example.springbootrabbitmqconsumerack.SpringbootRabbitmqConsumerAckApplication
                    </mainClass>
                </configuration>
                <executions>
                    <execution>
                        <id>repackage</id>
                        <goals>
                            <goal>repackage</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

</project>

application.yml

spring:
  application:
    name: springboot-rabbitmq-producer
  rabbitmq:
    host: 192.168.0.105
    port: 5672
    virtual-host: /
    username: admin
    password: admin
server:
  port: 8080

RabbitmqListener.java

package com.example.springbootrabbitmqconsumerack.listener;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

/**
 * @Author:humorchen
 * @Date 2021/9/26 18:15
 */
@Component
//ChannelAwareMessageListener这个接口不实现也是可以的，实现只是为了写下面方法的时候参数确定
public class RabbitmqListener implements ChannelAwareMessageListener {
    public static final String QUEUE_NAME = "springboot_queue";

    /**
     * 定义对队列进行监听
     * 设置手动签收 ackMode填写的要是大写，否则匹配不到枚举类AcknowledgeMode
     * public enum AcknowledgeMode {
     * NONE,
     * MANUAL,
     * AUTO;
     *
     * @param message
     */
    @RabbitListener(queues = RabbitmqListener.QUEUE_NAME, ackMode = "MANUAL")
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        System.out.println("收到消息");
        System.out.println(message);
        boolean dealSuccess = true;
        try {
            // ... your business code
            Thread.sleep(1500);
            //故意放个错误，处理失败
            int i = 1 / 0;
        } catch (Exception e) {
            e.printStackTrace();
            //出了异常，消费失败
            dealSuccess = false;
        }
        //如果处理成功手动签收
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        if (dealSuccess) {
            //第一个是delivery tag ，后面的b是是否签收多条消息  multiple
            channel.basicAck(deliveryTag, false);
            System.out.println("手动签收");
        } else {
            //处理失败，拒签     deliveryTag消息标签   multiple多条消息签掉  requeue重回队列
            channel.basicNack(deliveryTag, false, true);
            //也可以使用之前的一个方法  deliveryTag requeue重新入队  建议使用前面一个
            // channel.basicReject(deliveryTag,true);
            System.out.println("拒绝签收");
        }
    }
}

往这个队列里放一条消息，然后我们看我们的代码故意报错拒绝签收，让队列重新入队。

运行后我们可以看到一直在收到这个消息然后因为报错，拒收了消息，重新入队

我们注释掉错误试下，成功签收

去rabbitmq控制台手动获取下消息看看，发现消息的确被消费了

消费端限流

步骤：

• 设置每次最多拉取多少条消息
• 让消费端ack方式为手动确认方式(manual)

我们往刚才队列里填充10个消息（1-10，手动在控制台添加的），然后对刚才项目，注释掉故意设置的报错，注释掉睡眠，然后启动我们看rabbitmq控制台，我们可以看到消息速率是2/s，同时可以点开consumers看看有哪些消费者在消费，也可以看到我们设置的每次抓取消息数目限制为2

# 设置消费端限流 同时最多拉2个消息回来
    listener:
      simple:
        prefetch: 2

application.yml

spring:
  application:
    name: springboot-rabbitmq-producer
  rabbitmq:
    host: 192.168.0.105
    port: 5672
    virtual-host: /
    username: admin
    password: admin
    # 设置消费端限流 同时最多拉2个消息回来
    listener:
      simple:
        prefetch: 2
server:
  port: 8080

队列长度限制MaxLength

队列创建的时候设置参数"x-max-length"即可设置队列的最大长度

/**
 * 配置一个持久化的队列并命名
 * @return
 */
@Bean(QUEUE_NAME)
public Queue queue(){
    //设置队列最大长度为1000条消息
    return QueueBuilder.durable(QUEUE_NAME).maxLength(1000).build();
}

消息设置TTL（time to live）过期时间

队列创建的时候设置参数"x-message-ttl"即可设置消息的存活时间，也可以发送消息的时候单独设置过期时间，如果两个都设置了，取最小值。

/**
 * 配置一个持久化的队列并命名
 * @return
 */
@Bean("queue_name")
public Queue queue(){
    //设置消息存活时间为10秒
    return QueueBuilder.durable(QUEUE_NAME).ttl(10000).build();
}

//发送时手动设置过期时间示范
@SpringBootTest
class SpringbootRabbitmqProducerApplicationTests {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 测试使用rabbitmq发送消息
     */
    @Test
    public void testSend(){
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //手动设置消息的过期时间为5秒
                message.getMessageProperties().setExpiration("5000");
                return message;
            }
        };
        rabbitTemplate.convertAndSend(RabbitmqConfig.EXCHANGE_NAME,"springboot.hello","hello,springboot rabbitmq!",messagePostProcessor);
    }

}

我测试每隔一两秒时间发一个消息进队列，然后停下来不发了。看到的监控图如下，是生效的。

死信队列

消息成为死信的三种情况:

• 队列消息长度到达限制;
• 消费者拒接消费消息，basicNack/basicReject,并且不把消息重新放入原目标队列,requeue=false;
• 原队列存在消息过期设置，消息到达超时时间未被消费;

给队列绑定死信交换机：

给队列设置参数：x-dead-letter-exchange和x-dead-letter-routing-key

创建交换机1，队列1，队列1绑定到交换机1，并将队列1的死信队列绑定到交换机2去

创建交换机2，队列2，队列2绑定到交换机2

给交换机1发消息送入队列1，由一个消费者消费队列1的消息，并拒绝掉消息，这样这个消息就被送到了死信队列里去了，其实就是按照配置好的routing key 送到了交换机2，从而送到了队列2当中去了

创建项目springboot-rabbitmq-demo-dead-letter-exchange

pom.xml

<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.example</groupId>
    <artifactId>springboot-rabbitmq-demo-dead-letter-exchange</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>springboot-rabbitmq-demo-dead-letter-exchange</name>
    <description>Demo project for Spring Boot</description>

    <properties>
        <java.version>1.8</java.version>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <spring-boot.version>2.3.7.RELEASE</spring-boot.version>
    </properties>

    <dependencies>
        <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-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-configuration-processor</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-dependencies</artifactId>
                <version>${spring-boot.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <version>2.3.7.RELEASE</version>
                <configuration>
                    <mainClass>
                        com.example.springbootrabbitmqdemodeadletterexchange.SpringbootRabbitmqDemoDeadLetterExchangeApplication
                    </mainClass>
                </configuration>
                <executions>
                    <execution>
                        <id>repackage</id>
                        <goals>
                            <goal>repackage</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

</project>