入门案例

需求：使用简单模式完成消息传递

步骤：
创建工程（生成者、消费者）
分别添加依赖
编写生产者发送消息
编写消费者接收消息
在这里插入图片描述

3.1.2. 添加依赖

往heima-rabbitmq的pom.xml文件中添加如下依赖：

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

3.2. 编写生产者

package com.itheima.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;

/**
 * 发送消息
 */
public class Producer_HelloWorld {
    public static void main(String[] args) throws IOException, TimeoutException {

        ///建立连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 设置参数
        factory.setHost("localhost"); //默认值为localhost
        factory.setPort(5672); //端口 默认值 5672
        factory.setVirtualHost("/chengpeng");// 虚拟机 默认 /
        factory.setUsername("chengpeng"); // 用户名 默认值 guest
        factory.setPassword("chengpeng"); // 密码 默认值 guest
        /// 创建连接 connection
        Connection connection = factory.newConnection();
        /// 创建Channel
        Channel channel = connection.createChannel();
        ///创建队列 声明队列
        /**
         * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
         * queue:队列名称
         * durable:是否持久化，当mq重启之后还在
         * exclusive：
         *           一：是否独占，只能有一个消费者监听队列
         *           二：当Connection关闭时，是否删除队列
         * autoDelete：是否自动删除 当没有Consumer时，自动删除
         * arguments：参数
         */
        /// 如果没有一个名字叫helloworld的队列，则会创建
        channel.queueDeclare("hello_world",true,false,false,null);
        //发送消息
        /**
         * String exchange, String routingKey,  AMQP.BasicProperties props, byte[] body
         *
         * exchange:交换机名称，简单模式下交换机会使用默认的""
         * routingKey:路由名称 如果给默认的交换机需要给队列名称一样
         * props:配置信息
         * body: 发送消息数据
         */

        String body = "hello rabbitmq~~~~";
        channel.basicPublish("","hello_world",null,body.getBytes());

        /// 释放资源
        channel.close();
        connection.close();
    }
}

在执行上述的消息发送之后；可以登录rabbitMQ的管理控制台，可以发现队列和其消息：
在这里插入图片描述

3.3. 编写消费者

抽取创建connection的工具类com.itheima.rabbitmq.util.ConnectionUtil；
package com.itheima.consumer;

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 {
        ///建立连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 设置参数
        factory.setHost("localhost"); //默认值为localhost
        factory.setPort(5672); //端口 默认值 5672
        factory.setVirtualHost("/chengpeng");// 虚拟机 默认 /
        factory.setUsername("chengpeng"); // 用户名 默认值 guest
        factory.setPassword("chengpeng"); // 密码 默认值 guest
        /// 创建连接 connection
        Connection connection = factory.newConnection();
        /// 创建Channel
        Channel channel = connection.createChannel();
        ///创建队列 声明队列
        /**
         * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
         * queue:队列名称
         * durable:是否持久化，当mq重启之后还在
         * exclusive：
         *           一：是否独占，只能有一个消费者监听队列
         *           二：当Connection关闭时，是否删除队列
         * autoDelete：是否自动删除 当没有Consumer时，自动删除
         * arguments：参数
         */
        /// 如果没有一个名字叫helloworld的队列，则会创建
        channel.queueDeclare("hello_world",true,false,false,null);
        //接收消息
        /**
         * String queue, boolean autoAck, Consumer callback
         * queue: 队列名称
         * autoAck: 是否自动确认
         * 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("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);
        /// 释放资源
        //channel.close();
        //connection.close();
    }
}

3.4. 小结

在这里插入图片描述

在上图的模型中，有以下概念：

P：生产者，也就是要发送消息的程序
C：消费者：消息的接受者，会一直等待消息到来。
queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。

4. AMQP

4.1. 相关概念介绍

AMQP 一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。

AMQP是一个二进制协议，拥有一些现代化特点：多信道、协商式，异步，安全，扩平台，中立，高效。

RabbitMQ是AMQP协议的Erlang的实现。

概念	说明
连接Connection	一个网络连接，比如TCP/IP套接字连接。
会话Session	端点之间的命名对话。在一个会话上下文中，保证“恰好传递一次”。
信道Channel	多路复用连接中的一条独立的双向数据流通道。为会话提供物理传输介质。
客户端Client	AMQP连接或者会话的发起者。AMQP是非对称的，客户端生产和消费消息，服务器存储和路由这些消息。
服务节点Broker	消息中间件的服务节点；一般情况下可以将一个RabbitMQ Broker看作一台RabbitMQ 服务器。
端点	AMQP对话的任意一方。一个AMQP连接包括两个端点（一个是客户端，一个是服务器）。
消费者Consumer	一个从消息队列里请求消息的客户端程序。
生产者Producer	一个向交换机发布消息的客户端应用程序。

4.2. RabbitMQ运转流程

在入门案例中：

生产者发送消息
1. 生产者创建连接（Connection），开启一个信道（Channel），连接到RabbitMQ Broker；
2. 声明队列并设置属性；如是否排它，是否持久化，是否自动删除；
3. 将路由键（空字符串）与队列绑定起来；
4. 发送消息至RabbitMQ Broker；
5. 关闭信道；
6. 关闭连接；
消费者接收消息
1. 消费者创建连接（Connection），开启一个信道（Channel），连接到RabbitMQ Broker
2. 向Broker 请求消费相应队列中的消息，设置相应的回调函数；
3. 等待Broker回应闭关投递响应队列中的消息，消费者接收消息；
4. 确认（ack，自动确认）接收到的消息；
5. RabbitMQ从队列中删除相应已经被确认的消息；
6. 关闭信道；
7. 关闭连接；

在这里插入图片描述

4.3. 生产者流转过程说明

客户端与代理服务器Broker建立连接。会调用newConnection() 方法,这个方法会进一步封装Protocol Header 0-9-1 的报文头发送给Broker ，以此通知Broker 本次交互采用的是AMQPO-9-1 协议，紧接着Broker 返回Connection.Start 来建立连接，在连接的过程中涉及Connection.Start/.Start-OK 、Connection.Tune/.Tune-Ok ，Connection.Open/ .Open-Ok 这6 个命令的交互。
客户端调用connection.createChannel方法。此方法开启信道，其包装的channel.open命令发送给Broker,等待channel.basicPublish方法，对应的AMQP命令为Basic.Publish,这个命令包含了content Header 和content Body()。content Header 包含了消息体的属性，例如:投递模式，优先级等，content Body 包含了消息体本身。
客户端发送完消息需要关闭资源时，涉及到Channel.Close和Channl.Close-Ok 与Connetion.Close和Connection.Close-Ok的命令交互。

在这里插入图片描述

4.4. 消费者流转过程说明

消费者客户端与代理服务器Broker建立连接。会调用newConnection() 方法,这个方法会进一步封装Protocol Header 0-9-1 的报文头发送给Broker ，以此通知Broker 本次交互采用的是AMQPO-9-1 协议，紧接着Broker 返回Connection.Start 来建立连接，在连接的过程中涉及Connection.Start/.Start-OK 、Connection.Tune/.Tune-Ok ，Connection.Open/ .Open-Ok 这6 个命令的交互。
消费者客户端调用connection.createChannel方法。和生产者客户端一样，协议涉及Channel . Open/Open-Ok命令。
在真正消费之前，消费者客户端需要向Broker 发送Basic.Consume 命令(即调用channel.basicConsume 方法〉将Channel 置为接收模式，之后Broker 回执Basic . Consume - Ok 以告诉消费者客户端准备好消费消息。
Broker 向消费者客户端推送(Push) 消息，即Basic.Deliver 命令，这个命令和Basic.Publish 命令一样会携带Content Header 和Content Body。
消费者接收到消息并正确消费之后，向Broker 发送确认，即Basic.Ack 命令。
客户端发送完消息需要关闭资源时，涉及到Channel.Close和Channl.Close-Ok 与Connetion.Close和Connection.Close-Ok的命令交互。

在这里插入图片描述