消息队列RabbitMQ - RabbitMQ概念与工作模式 - 《中间件》

1.1MQ概述
1.2 MQ的优势和劣势
1.3 常见的 MQ 产品
1.4 RabbitMQ 简介
2.1 RabbitMQ快速入门：简单模式
2.2 RabbitMQ 的工作模式

1.1MQ概述

MQ简称 Message Queue（消息队列），是在消息的传输过程中保存消息的容器。多用于分布式系统之间进行通信。

小结：

MQ，消息队列，存储消息的中间件
分布式系统通信的两种方式：直接远程调用和借助第三方完成间接通信
发送方称为生产者，接收方称为消费者

1.2 MQ的优势和劣势

优势：

应用解耦：提高系统的容错性和可维护性

异步提速：提升用户体验和系统吞吐量

削峰填谷：提高系统稳定性

劣势：

系统可用性降低
系统复杂性提高
一致性问题

小结

1.3 常见的 MQ 产品

1.4 RabbitMQ 简介

AMQP协议

RabbitMQ

JMS

小结

2.1 RabbitMQ快速入门：简单模式

简单模式

创建一个工程rabbitmq，在这个工程下面创建2个模块：rabbitmq-producer，rabbitmq-consumer。
修改两个pom.xml文件 ```xml com.rabbitmq amqp-client 5.14.2

org.apache.maven.plugins maven-compiler-plugin 3.10.1 1.8 1.8


3. 生产者代码
```java
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
 * 发送消息
 */
public class Producer_HelloWorld {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 2.设置参数
        factory.setHost("192.168.133.128"); // ip 默认值为 localhost
        factory.setPort(5672); // 端口 默认值 5672
        factory.setVirtualHost("/"); // 虚拟机 默认值/
        factory.setUsername("xg"); // 用户名 默认值 guest
        factory.setPassword("167219"); // 密码 默认值 guest
        // 3.创建连接 Connection
        Connection connection = factory.newConnection();
        // 4.创建Channel
        Channel channel = connection.createChannel();
        // 5.HelloWorld方式，不需要交换机。创建队列Queue
        // 声明一个队列
        /**
         * queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String,Object> arguments</>)
         * 参数：
         *      1. queue：队列名称
         *      2. durable：是否持久化，当mq重启后，还在
         *      3. exclusive：
         *          * 是否独占。只能有一个消费者监听这个队列
         *          * 当Connection关闭时，是否删除队列
         *      4. autoDelete：是否自动删除。当没有Consumer时，自动删除掉
         *      5. arguments：参数
         */
        // 如果没有一个名字叫做hello_world的队列，则会创建该队列；如果有则不会创建
        channel.queueDeclare("hello_world", true, false, false, null);
        // 6.发送消息
        /**
         * basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
         * 参数：
         *      1. exchange：交换机名称。简单模式下交换机会使用默认的 ""
         *      2. routingKey：路由名称
         *      3. props：配置信息
         *      4. body：发送的消息数据(字节)
         */
        String body = "hello rabbitmq~~~";
        channel.basicPublish("", "hello_world", null, body.getBytes());
        // 7.释放资源
        channel.close();
        connection.close();
    }
}

运行后，访问rabbitmq客户端：

消费者代码 ```java import com.rabbitmq.client.*;

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

public class Consumer_HelloWorld { public static void main(String[] args) throws IOException, TimeoutException { // 1.创建连接工厂 ConnectionFactory factory = new ConnectionFactory();

    // 2.设置参数
    factory.setHost("192.168.133.128"); // ip 默认值为 localhost
    factory.setPort(5672); // 端口 默认值 5672
    factory.setVirtualHost("/"); // 虚拟机 默认值/
    factory.setUsername("xg"); // 用户名 默认值 guest
    factory.setPassword("167219"); // 密码 默认值 guest
    // 3.创建连接 Connection
    Connection connection = factory.newConnection();
    // 4.创建Channel
    Channel channel = connection.createChannel();
    // 5.HelloWorld方式，不需要交换机。创建队列Queue
    // 声明一个队列
    /**
     * queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String,Object> arguments</>)
     * 参数：
     *      1. queue：队列名称
     *      2. durable：是否持久化，当mq重启后，还在
     *      3. exclusive：
     *          * 是否独占。只能有一个消费者监听这个队列
     *          * 当Connection关闭时，是否删除队列
     *      4. autoDelete：是否自动删除。当没有Consumer时，自动删除掉
     *      5. arguments：参数
     */
    // 如果没有一个名字叫做hello_world的队列，则会创建该队列；如果有则不会创建
    channel.queueDeclare("hello_world", true, false, false, null);
    // 6.消费(接收)消息
    /**
     * basicConsume(String queue, boolean autoAck, Consumer callback)
     * 参数：
     *      1. queue：队列名称
     *      2. autoAck：是否自动确认
     *      3. callback：回调对象
     */
    // 接收消息(匿名内部类方式，重写方法)
    Consumer consumer = new DefaultConsumer(channel) {
        /**
         * 回调方法，当收到消息后，会自动执行该方法
         * 参数：
         *      1. consumerTag：消息标识
         *      2. envelope：获取一些信息，交换机，路由key...
         *      3. properties：配置信息
         *      4. body：真实数据
         */
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
            System.out.println("consumerTag: " + consumerTag);
            System.out.println("Exchange: " + envelope.getExchange());
            System.out.println("RoutingKey: " + envelope.getRoutingKey());
            System.out.println("properties: " + properties);
            System.out.println("body: " + new String(body));
        }
    };
    channel.basicConsume("hello_world", true, consumer);
    // 消费者要进行监听，不能关闭资源
}