RabbitMQ实战

- 日理万妓 2023-02-16 00:49 57阅读 0赞

# 1.什么是RabbitMQ #

RabbitMQ是一个免费的，开源的，可扩展的消息队列解决方案。本质是一个消息代理，可以理解AMQP（高级消息队列协议），但也可以与其他流行的消息解决方案一起使用。它具有高可用性，容错性和可扩展性。  
它是一个中间件层，使您的应用程序中的不同服务可以彼此通信，而不必担心消息丢失。本质是使不同的进程或程序进行解耦。

# 2.RabbitMQ环境搭建 #

参考：https://www.cnblogs.com/saryli/p/9729591.html

# 3.学习5种队列 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1OTMzMjQ5_size_16_color_FFFFFF_t_70]

## 3.1创建RabbitMq项目 ##

<?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>
        <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>2.1.14.RELEASE</version>
            <relativePath/> 
        </parent>
        <groupId>com.lpp</groupId>
        <artifactId>rabbitmq</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <name>rabbitmq</name>
        <description>Demo project for Spring Boot</description>
    
    
        <properties>
            <java.version>1.8</java.version>
        </properties>
    
        <dependencies>
          <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
            <dependency>
                <groupId>com.rabbitmq</groupId>
                <artifactId>amqp-client</artifactId>
                <version>3.4.1</version>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
            </dependency>
        </dependencies>
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                </plugin>
            </plugins>
        </build>
    
    </project>

### 3.1 简单队列 ###

在下图中，“ P”是我们的生产者，“ C”是我们的消费者。中间的框是一个队列-RabbitMQ代表使用者保留的消息缓冲区。  
![在这里插入图片描述][20200606214007331.png]  
生产者P将生产的消息“Hello”发送给队列，消费者再从队列中获取。

**建立连接**

package com;
    
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    public class ConnectionUtil { 
    
        public static Connection getConnection() throws Exception{ 
            //1.定义连接工程
            ConnectionFactory factory = new ConnectionFactory();
            //2.设置服务地址
            factory.setHost("localhost");
            //3.端口
            factory.setPort(5672);
            //4.设置虚拟信息
            factory.setVirtualHost("admin");
            factory.setUsername("admin");
            factory.setPassword("123456");
            Connection connection = factory.newConnection();
            return connection;
        }
    }

**生产者发送消息到队列**

package com.lpp.rabbitmq.queue;
    
    import com.ConnectionUtil;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    public class Send { 
    
        private final static String QUEUE_NAME = "q_test_01";
    
        public static void main(String[] args)throws Exception { 
    
            //获取连接
            Connection connection = ConnectionUtil.getConnection();
    
            //从连接中创建管道
            Channel channel = connection.createChannel();
    
            //管道中创建队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            //消息内容
            String message = "Hello World";
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");
    
            //关闭管道和连接
            channel.close();
            connection.close();
    
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1OTMzMjQ5_size_16_color_FFFFFF_t_70 1]  
点击队列查询消息  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1OTMzMjQ5_size_16_color_FFFFFF_t_70 2]  
会看见一个消息待消费  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1OTMzMjQ5_size_16_color_FFFFFF_t_70 3]  
查看消息

**消费者从队列中获取消息**

package com.lpp.rabbitmq.queue;
    
    import com.ConnectionUtil;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    
    public class Recv { 
    
        private final static String QUEUE_NAME = "q_test_01";
    
        public static void main(String[] args) throws Exception{ 
            //获取连接
            Connection connection = ConnectionUtil.getConnection();
    
            //从连接中创建管道
            Channel channel = connection.createChannel();
    
            //管道中创建队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            //定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            
            //监听队列
            channel.basicConsume(QUEUE_NAME, true, consumer);
            
            //获取消息
            while (true)
            { 
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                byte[] body = delivery.getBody();
                String message = new String(body);
                System.out.println(" [x] Received '" + message + "'");
            }
    
        }
    }

## 3.2 工作队列（work模式又叫竞争消费模式） ##

![在这里插入图片描述][20200606221229384.png]  
生产者  
发送100条消息

package com.lpp.rabbitmq.work;
    
    import com.ConnectionUtil;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    
    public class WorkSend { 
    
        private final static String QUEUE_NAME = "test_queue_work";
    
        public static void main(String[] args) throws Exception { 
    
            Connection connection = ConnectionUtil.getConnection();
    
            Channel channel = connection.createChannel();
    
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            for (int i = 0; i < 100; i++) { 
    
                String message = ""+i;
    
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
                System.out.println(" [x] Sent '" + message + "'");
                Thread.sleep(i * 10);
            }
    
            channel.close();
            connection.close();
    
        }
    }

消费者X

package com.lpp.rabbitmq.work;
    
    import com.ConnectionUtil;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class WorkRevX { 
    
        private final static String QUEUE_NAME = "test_queue_work";
    
        public static void main(String[] args) throws Exception { 
    
            //创建连接及通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            //声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            //同一时刻服务器只会发一条消息给消费者
            channel.basicQos(1);
    
            //定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
    
            channel.basicConsume(QUEUE_NAME, true, consumer);
            int i = 0;
            while (true) { 
    
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
    
                String message = new String(delivery.getBody());
    
                System.out.println(" [x] Received '" + message + ","+i++);
                Thread.sleep(1000);
                //channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
            }
        }
    }

消费者Y

package com.lpp.rabbitmq.work;
    
    import com.ConnectionUtil;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class WorkRecvY { 
    
        private final static String QUEUE_NAME = "test_queue_work";
    
        public static void main(String[] args) throws Exception { 
    
            //创建连接及通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            //声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            //同一时刻服务器只会发一条消息给消费者
            channel.basicQos(1);
    
            //定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列，false表示手动返回完成状态，true表示自动
            channel.basicConsume(QUEUE_NAME, true, consumer);
            int i = 0;
            while (true) { 
    
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
    
                String message = new String(delivery.getBody());
    
                System.out.println(" [y] Received '" + message + ","+i++);
                Thread.sleep(10);
                //channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
            }
        }
    }

**测试结果**

*  消费者X和消费者Y消费的内容的是不同，同一个消息只能被消费一次。
 *  消费者X和消费者Y消费的消息内容相同，一个是消费奇数号消息，一个是偶数。
 *  其实，这样是不合理的，因为消费者Y线程停顿的时间短。应该是消费者Y要比消费者X获取到的消息多才对。  
    RabbitMQ 默认将消息顺序发送给下一个消费者，这样，每个消费者会得到相同数量的消息。即轮询（round-robin）分发消息。
 *  怎样才能根据消费者消费能力进行消息分配呢？使用Qos和Acknowlege就可以做到。basicQos方法设置当前信道最大预获取（prefetch）消息数量。消息从队列异步推送给消费者，消费者的ack确认机制响应也是异步发送给队列，从队列角度看，当Qos的prefetch(预载入）设置为1时，只用消费者进行了ack消费确认，队列才会发送下一条消息。Qos默认为0,即没有限制，也无需确认。所以必须Qos和ack配置使用。

**队列消息策略**

*  轮询分发：队列默认的分发策略是轮询分发。即逐个发送消息给序列中的消费者，进而达到并行工作，易于扩展、伸缩。
 *  公平分发：即按照消费能力进行消息分发，能者多劳。  
    上面的消费者X和消费者Y，明显消费者Y单位时间内消费者能力更强。为此我们使用basicQos( prefetchCount = 1)方法，来限制RabbitMQ只发不超过1条的消息给同一个消费者。当消息处理完毕后，有了反馈，才会进行第二次发送。  
    还有一点需要注意，使用公平分发，必须关闭自动应答，改为手动应答。  
    消费者可以设置

// 同一时刻服务器只会发一条消息给消费者
    channel.basicQos(1);

//开启这行 表示使用手动确认模式
    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

// 监听队列，false表示手动返回完成状态，true表示自动
    channel.basicConsume(QUEUE_NAME, false, consumer);

## 3.3 几种交换机模式 ##

RabbitMQ的消息发送的核心思想是生产者P从不将消息发送到任何队列，实际上生产者常常甚至不知道是否将消息已经发送到队列上。由此就有交换机的概念，生产者将消息发送给交换机，交换机将消息推入指定的队列。交换机确切知道从生产者是否获取了消息，以及如何处理消息即是否将其将入特定队列和是否应该丢弃。

一个生产者也交换机绑定，由交换管理消息，可以由多个队列和多个消费者消费。  
交换机的类型主要有direct、topic、headers、fanout四种类型。

### 3.3.1 Fanout Exchange（扇形交换机） ###

**fanout 型交换机实现的其实就是发布订阅模式。**  
生产者  
![在这里插入图片描述][20200607150123334.png]

package com.lpp.rabbitmq.ps;
    
    import com.ConnectionUtil;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    public class PsSend { 
    
        private final static String EXCHANGE_NAME = "test_exchange_fanout";
    
        public static void main(String[] args) throws Exception { 
    
            //获取连接及通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            //声明交换机
            channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
    
            //消息内容
            String message = "Hello World";
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
    
            System.out.println(" [x] Sent '" + message + "'");
    
            channel.close();
            connection.close();
    
        }
    }

注意：消息发送到没有队列绑定的交换机时，消息将丢失，因为，交换机没有存储消息的能力，消息只能存在在队列中。

消费者X

package com.lpp.rabbitmq.ps;
    
    import com.ConnectionUtil;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class PsRecvX { 
    
        private final static String QUEUE_NAME = "test_queue_work1";
    
        private final static String EXCHANGE_NAME = "test_exchange_fanout";
    
        public static void main(String[] args) throws Exception { 
    
            //获取连接和通道
            Connection connection = ConnectionUtil.getConnection();
    
            Channel channel = connection.createChannel();
    
            //声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            //将队列绑定到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
    
            //同一时刻服务器只会发送一条消息给消费者
            channel.basicQos(1);
    
            //定义消费者队列
            QueueingConsumer consumer = new QueueingConsumer(channel);
    
            //监听队列，手动返回完成
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            while (true) { 
    
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" [X] Received '" + message + "'");
                Thread.sleep(10);
    
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    
            }
    
    
        }
    }

消费者Y

package com.lpp.rabbitmq.ps;
    
    import com.ConnectionUtil;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class PsRecvY { 
    
        private final static String QUEUE_NAME = "test_queue_work2";
    
        private final static String EXCHANGE_NAME = "test_exchange_fanout";
    
        public static void main(String[] args) throws Exception { 
    
            //1.获取连接和通道
            Connection connection = ConnectionUtil.getConnection();
    
            Channel channel = connection.createChannel();
    
            //2.声明队令
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            //3.绑定交换寄
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
    
            channel.basicQos(1);
            //4.创建消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
    
            //5.监听队列，手动确认
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            while (true) { 
    
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" [Y] Received '" + message + "'");
                Thread.sleep(10);
    
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    
            }
    
    
        }
    }

测试结果：  
同一个消息被多个消费者获取。一个消费者队列可以有多个消费者实例，只有其中一个消费者实例会消费到消息。

### 3.3.2 Direct Exchange(直接交换机) ###

Direct Exchange的实现是路由模式。  
扇型交换，它并没有给我们带来太大的灵活性-它只能进行无意识的广播。  
直接交换背后的路由算法很简单-消息进入其绑定键与消息的路由键全匹配的队列 。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1OTMzMjQ5_size_16_color_FFFFFF_t_70 4]  
我们可以看到绑定了两个队列的直接交换机X。第一个队列由绑定键orange绑定，第二个队列有两个绑定，一个绑定键为black，另一个绑定为green。

在这种设置中，使用路由键orange发布到交换机的消息 将被路由到队列Q1。路由键为black 或greed的消息将转到Q2。所有其他消息将被丢弃。  
示例：  
生产者

package com.lpp.rabbitmq.route;
    
    import com.ConnectionUtil;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    public class RouteSend { 
    
        public static final String EXCHANGE_NAME = "test_exchange_direct";
    
        public static void main(String[] args) throws Exception { 
            //1.获取连接和通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            //2.声明exchange
            channel.exchangeDeclare(EXCHANGE_NAME, "direct");
    
            //3.消息内容
            String message = "删除商品";
    
            channel.basicPublish(EXCHANGE_NAME, "delete", null, message.getBytes());
            System.out.println("X send "+message);
    
            channel.close();
            connection.close();
    
        }
    }

消费者X

package com.lpp.rabbitmq.route;
    
    import com.ConnectionUtil;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class RouteRecvX { 
    
        public static final String EXCHANGE_NAME = "test_exchange_direct";
        public static final String QUEUE_NAME = "test_queue_direct_1";
    
        public static void main(String[] args)throws Exception { 
    
            //1.获取连接和通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            //2.声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            //3.绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");
    
            //4.同一时刻服务器只会发送一条消息给消费者
            channel.basicQos(1);
    
            //5.定义消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
    
            while (true) { 
    
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
    
                System.out.println("X Recv:" + message);
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

消费者Y

package com.lpp.rabbitmq.route;
    
    import com.ConnectionUtil;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class RouteRecvY { 
    
        public static final String EXCHANGE_NAME = "test_exchange_direct";
        public static final String QUEUE_NAME = "test_queue_direct_2";
    
        public static void main(String[] args) throws Exception { 
            //1.获取连接和通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            //2.申明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            //3.绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert");
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
            //channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");
    
            //4.同一时刻服务器只会发送一条消息给消费者
            channel.basicQos(1);
    
            //5.定义队列消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
    
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            while (true) { 
    
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println("Y Recv:" + message);
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
    
        }
    
    }

测试结果：生产者发送的消息被消费者X所消费。

### 3.3.3 Topic Exchage(主题交换机) ###

topic exchange对应是主题模式

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1OTMzMjQ5_size_16_color_FFFFFF_t_70 5]

发送到主题交换机的消息不能具有任意的 routing\_key-它必须是单词列表，以点分隔。这些词可以是任何，但是通常它们指定与消息相关的某些功能。一些有效的路由关键示例：“ stock.usd.nyse ”，“ nyse.vmw ”，“ quick.orange.rabbit ”。路由键中可以包含任意多个单词，最多255个字节。

绑定键也必须采用相同的形式。主题交换机背后的逻辑类似于直接交换机的逻辑 -使用特定路由键发送的消息将被传递到所有使用匹配绑定键绑定的队列。但是，绑定键有两个重要的特殊情况：

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

示例：  
生产者

package com.lpp.rabbitmq.topic;
    
    import com.ConnectionUtil;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    public class TopicSend { 
    
        public static final String EXCHANGE_NAME = "test_exchange_topic";
    
        public static void main(String[] args) throws Exception { 
    
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
    
            channel.exchangeDeclare(EXCHANGE_NAME, "topic");
    
            String message = "Hello World";
    
            channel.basicPublish(EXCHANGE_NAME, "routekey.kk", null, message.getBytes());
    
            System.out.println(" [x] Sent '" + message + "'");
    
            channel.close();
            connection.close();
        }
    }

消费者X

package com.lpp.rabbitmq.topic;
    
    import com.ConnectionUtil;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class TopicRecvX { 
    
        private final static String QUEUE_NAME = "test_queue_topic_work_1";
    
        private final static String EXCHANGE_NAME = "test_exchange_topic";
    
        public static void main(String[] args) throws Exception { 
            //1.获取连接和通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
            //2.声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            //3.绑定队列和交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "routekey.kk");
    
            //4.服务器发送信息数
            channel.basicQos(1);
    
            //5.定义消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            //6.监听队列
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
    
            while (true) { 
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" [x] Received '" + message + "'");
                Thread.sleep(10);
    
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

消费者Y

package com.lpp.rabbitmq.topic;
    
    import com.ConnectionUtil;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class TopicRecvY { 
    
        private final static String QUEUE_NAME = "test_queue_topic_work_2";
    
        private final static String EXCHANGE_NAME = "test_exchange_topic";
    
        public static void main(String[] args) throws Exception { 
    
            //1.获取连接和通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            //2.声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            //3.绑定队列和交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*.*");
    
            //4.定义服务器统一时刻发送消息数
            channel.basicQos(1);
    
            //5.定义消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
    
            //6.监听队列
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            while (true) { 
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" [Y] Received '" + message + "'");
                Thread.sleep(10);
    
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    
            }
        }
    }

测试结果：消费者X消费了消息。

# 4.SpringBoot整合RabbitMQ #

感觉这篇写的比较好  
https://blog.csdn.net/qq\_35387940/article/details/100514134

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1OTMzMjQ5_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200606212653192.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1OTMzMjQ5,size_16,color_FFFFFF,t_70
[20200606214007331.png]: https://img-blog.csdnimg.cn/20200606214007331.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1OTMzMjQ5_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200606220410471.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1OTMzMjQ5,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1OTMzMjQ5_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200606220445440.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1OTMzMjQ5,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1OTMzMjQ5_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200606220518792.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1OTMzMjQ5,size_16,color_FFFFFF,t_70
[20200606221229384.png]: https://img-blog.csdnimg.cn/20200606221229384.png
[20200607150123334.png]: https://img-blog.csdnimg.cn/20200607150123334.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1OTMzMjQ5_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200607153132706.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1OTMzMjQ5,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1OTMzMjQ5_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20200607154307382.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1OTMzMjQ5,size_16,color_FFFFFF,t_70