03_RabbitMQ快速入门案例

RabbitMQ快速入门案例

新建一个maven过程

导入依赖

 
        <dependency>
            <groupId>com.rabbitmqgroupId>
            <artifactId>amqp-clientartifactId>
            <version>5.10.0version>
        dependency>
1
2
3
4
5
6

producer

• 打断点一步一步查看

package com.tian.rabbitmq.simple;

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

public class Producer {
    public static void main(String[] args) {
        //所有的中间件技术都是基于tcp/ip协议基础之上构建新型的协议规范，只不过rabbitmq遵循的是amcp
        // ip port


        //1:创建连接工程
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.1.150");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        connectionFactory.setVirtualHost("/");
        Connection connection = null;
        Channel channel = null;
        try {
            //2:创建连接connection
            connection = connectionFactory.newConnection("生产者");

            //3:通过连接获取通道Channel
            channel = connection.createChannel();

            //4:通过创建交换机，声明队列，绑定关系，路由key, 发送消息，和接收消息
            String queueName = "queue1";
            /**
             * @params1 队列的名称
             * @params2 是否持久化
             * @params3 排他性
             * @params4 是否自动删除
             * @params5 携带附属参数
             */
            channel.queueDeclare(queueName, false, false, false, null);
            //5:准备消息内容
            String message = "hello rabbitmq";
            //6:发送消息给队列queue
            channel.basicPublish("", queueName, null, message.getBytes());
            System.out.println("消息发送成功");

        } catch (Exception ex) {
            ex.printStackTrace();
        }finally {
            //7:关闭连接
            if(channel!=null&&channel.isOpen()){
                try {
                    channel.close();
                }catch (Exception ex){
                    ex.printStackTrace();
                }
            }
            //8:关闭通道
            if(connection!=null&&connection.isOpen()){
                try {
                    connection.close();
                }catch (Exception ex){
                    ex.printStackTrace();
                }
            }
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

consumer

package com.tian.rabbitmq.simple;

import com.rabbitmq.client.*;

import java.io.IOException;

public class Consumer {
    public static void main(String[] args) {
        //1:创建连接工程
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.1.150");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        connectionFactory.setVirtualHost("/");
        Connection connection = null;
        Channel channel = null;
        try {
            //2:创建连接connection
            connection = connectionFactory.newConnection("生产者");

            //3:通过连接获取通道Channel
            channel = connection.createChannel();

            channel.basicConsume("queue1", true, new DeliverCallback() {
                public void handle(String s, Delivery delivery) throws IOException {
                    System.out.println("收到的消息是" + new String(delivery.getBody(), "UTF-8"));
                }
            }, new CancelCallback() {
                public void handle(String s) throws IOException {
                    System.out.println("接收失败了");
                }
            });

            System.out.println("开始接收消息");
            System.in.read();

        } catch (Exception ex) {
            ex.printStackTrace();
        }finally {
            //7:关闭连接
            if(channel!=null&&channel.isOpen()){
                try {
                    channel.close();
                }catch (Exception ex){
                    ex.printStackTrace();
                }
            }
            //8:关闭通道
            if(connection!=null&&connection.isOpen()){
                try {
                    connection.close();
                }catch (Exception ex){
                    ex.printStackTrace();
                }
            }
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

什么是AMQP

AMQP全称: Advanced Message Queuing Protocol(高级消息队列协议)。是应用层协议的一个开发标准，为面向消息的中间件设计。

AMQP生产者流传过程

RabbitMQ的核心组成部分

核心概念

Server : 又称Broker ,接受客户端的连接，实现AMQP实体服务。安装rabbitmq-server

Connection: 连接，应用程序与Broker的网络连接TCP/IP/三次握手和四次挥手

Channel: 网络信道，几乎所有的操作都在Channel中进行，Channel是进行消息读写的通道，客户端可以建立对各Channel，每个Channel代表一个会话任务。

Message : 消息:服务与应用程序之间传送的数据，由Properties和body组成，Properties可是对消息进行修饰，比如消息的优先级，延迟等高级特性，Body则就是消息体的内容。

Virtual Host 虚拟地址，用于进行逻辑隔离，最上层的消息路由，一个虚拟主机理由可以有若干个Exhange和Queueu，同一个虚拟主机里面不能有相同名字的Exchange

Exchange :交换机，接受消息，根据路由键发送消息到绑定的队列。(不具备消息存储的能力)

Bindings : Exchange和Queue之间的虚拟连接，binding中可以保护多个routing key.

Routing key :是一个路由规则，虚拟机可以用它来确定如何路由一个特定消息。

Queue : 队列:也成为Message Queue,消总队列，保存消息并将它们转发给消费者。

//@params1:交换机 @params2 队列 路由key @params3 消息的控制状态 @params4 消息主题
//面试题 可以存在没有交换机的队列嘛？不可能，虽然没有指定交换机但是一定会存在一个默认的交换机
channel.basicPublish("", queueName, null, message.getBytes());
         
1
2
3
4

RabbitMQ整体架构图

RabbitMQ的运行流程