RabbitMQ - （二）RabbitMQ之Hello World

向右看齐 2022-05-26 12:29 197阅读 0赞

## 基础概念 ##

![这里写图片描述][70]

**producer生产者**  
消息的生产者，可以理解为发送消息的一方

**consumer消费者**  
消息的消费者，可以理解为接受消息的一方

**connection连接**  
Connection 网络连接，比如一个TCP连接。

**channel信道**  
Channel 信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内地虚拟连接，AMQP 命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接。

**exchange交换机**  
用来接收生产者发送的消息并将这些消息路由给服务器中的队列。

所有消息先送达到这里，然后他再按Queue的定义规则，将消息分类发送给对应的queue

**queue队列**  
消息队列申明自己所要消息的规则，然后从exchange获取对应的消息。另一端则将该消息推送给对应的consumer 。queue可以设置durable,这样可以保证RabbitMQ挂后，不丢数据，但是有一定时间窗口，没落盘的数据，可能还是会丢。

**message消息**  
字节数组

**routing key路由键**  
用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换机理解成一个由绑定构成的路由表。

**Virtual Host**  
虚拟主机，表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器，拥有自己的队列、交换器、绑定和权限机制。vhost 是 AMQP 概念的基础，必须在连接时指定，RabbitMQ 默认的 vhost 是 / 。

**Broker**  
表示消息队列服务器实体。

## 添加依赖 ##

<dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>5.2.0</version>
    </dependency>

## 生产者 ##

**创建步骤：**

*  创建连接工厂
 *  设置连接工厂的host、port、用户名、密码
 *  创建连接
 *  创建信道
 *  声明交换机
 *  声明队列
 *  绑定交换机和队列
 *  创建消息
 *  发送消息
 *  关闭信道和连接

@Test
        public void testSend1() throws IOException, TimeoutException {
            // 1、创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            // 2、设置连接工厂的host、port、用户名、密码
            factory.setHost("localhost");
            factory.setPort(5672);
            factory.setUsername("guest");
            factory.setPassword("guest");
            // 3、创建连接
            Connection connection = factory.newConnection();
            // 4、创建信道
            Channel channel = connection.createChannel();
            // 5、声明交换机
            String exchangeName = "exchange-direct";
            String exchangeType = "direct";
            channel.exchangeDeclare(exchangeName, exchangeType, true);
            // 6、声明队列
            String queueNameA = "queue-a";
            channel.queueDeclare(queueNameA, true, false, false, null);
            String queueNameB = "queue-b";
            channel.queueDeclare(queueNameB, true, false, false, null);
            // 7、绑定队列和交换机
            String routingKey = "routing-key";
            String routingKeyA = "routing-key-a";
            String routingKeyB = "routing-key-b";
            channel.queueBind(queueNameA, exchangeName, routingKey);
            channel.queueBind(queueNameA, exchangeName, routingKeyA);
            channel.queueBind(queueNameB, exchangeName, routingKeyB);
            // 8、创建消息
            String message = "hello rabbitmq";
            // 9、发送消息
            channel.basicPublish(exchangeName, routingKey, null, message.getBytes());
            channel.basicPublish(exchangeName, routingKeyA, null, message.getBytes());
            channel.basicPublish(exchangeName, routingKeyB, null, message.getBytes());
            System.out.println("发送的消息:" + message);
            // 10、关闭信道和连接
            channel.close();
            connection.close();
        }

## 消费者 ##

**创建步骤：**

*  创建连接工厂
 *  设置连接工厂的host、post、用户名、密码
 *  创建连接
 *  创建信道
 *  声明交换机
 *  声明队列
 *  绑定交换机和队列
 *  创建消费者并和信道进行绑定
 *  绑定消费者和队列

@Test
        public void testReceive2() throws Exception {
            // 1、创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            // 2、设置连接工厂的host、port、用户名、密码
            factory.setHost("localhost");
            factory.setPort(5672);
            factory.setUsername("guest");
            factory.setPassword("guest");
            // 3、创建连接
            Connection connection = factory.newConnection();
            // 4、创建信道
            Channel channel = connection.createChannel();
            // 5、声明交换机
            String exchangeName = "exchange-direct";
            String exchangeType = "direct";
            channel.exchangeDeclare(exchangeName, exchangeType, true);
            // 6、声明队列
            String queueNameA = "queue-a";
            channel.queueDeclare(queueNameA, true, false, false, null);
            String queueNameB = "queue-b";
            channel.queueDeclare(queueNameB, true, false, false, null);
            // 7、绑定队列和交换机
            String routingKey = "routing-key";
            String routingKeyA = "routing-key-a";
            String routingKeyB = "routing-key-b";
            channel.queueBind(queueNameA, exchangeName, routingKey);
            channel.queueBind(queueNameA, exchangeName, routingKeyA);
            channel.queueBind(queueNameB, exchangeName, routingKeyB);
            // 8、创建消费者并和信道进行绑定
            Consumer consumer = new DefaultConsumer(channel) {
    
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String message = new String(body, "UTF-8");
                    System.out.println("接收到消息:" + message);
                }
    
            };
            // 9、绑定消费者和队列
            channel.basicConsume(queueNameA, true, consumer);
            channel.basicConsume(queueNameB, true, consumer);
            while (true) {
    
            }
        }

[70]: /images/20220526/e77eaf1c72a94525804c42063b6b4e2d.png