RabbitMQ 从入门到精通（一）

程序员文章站 2022-06-19 17:07:24

[TOC] 1. 初识RabbitMQ RabbitMQ 是一个开源的消息代理和队列服务器，用来通过普通协议在完全不同的应用之间共享数据，RabbitMQ是使用 Erlang语言来编写的，并且RabbitMQ是基于AMQP协议的 RabbitMQ的优点: 开源、性能优秀、稳定性保障提供可靠性消息投 ......

1. 初识rabbitmq
2. amqp
3.rabbitmq的极速入门
4. exchange(交换机)详解
5. message 消息

1. 初识rabbitmq

rabbitmq 是一个开源的消息代理和队列服务器，用来通过普通协议在完全不同的应用之间共享数据，rabbitmq是使用 erlang语言来编写的，并且rabbitmq是基于amqp协议的

rabbitmq的优点:

开源、性能优秀、稳定性保障
提供可靠性消息投递模式(confirm)、返回模式(return)
与springamqp完美的整合、api丰富
集群模式丰富，表达式配置，ha模式，镜像队列模型
保证数据不丢失的前提下做到高可靠性、可用性

rabbitmq官网

rabbitmq的整体架构:

RabbitMQ 从入门到精通（一）

rabbitmq的消息流转:

RabbitMQ 从入门到精通（一）

2. amqp

amqp全称: advanced message queuing protocol

amqp翻译: 高级消息队列协议

amqp定义: 是具有现代特征的二进制协议。是一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计

RabbitMQ 从入门到精通（一）

amqp核心概念：

server:又称broker，接受客户端的连接，实现amqp实体服务
connection:连接，应用程序与broker的网络连接
channel:网络信道，几乎所有的操作都在channel中进行，channel是进行消息读写的通道。客户端可建立多个channel，每个channel代表一个会话任务
message:消息，服务器和应用程序之间传送的数据，由properties和body组成。properties可以对消息进行修饰，比如消息的优先级、延迟等高级特性；body则是消息体的内容
virtual host:虚拟地址，用于进行逻辑隔离，最上层的消息路由。同一个virtual host里面不能有相同名称的exchange或queue
exchange:交换机，接收消息，根据路由键转发消息到绑定的队列
binding:exchange和queue之间的虚拟连接,binding中可以包含routing key
routing key:一个路由规则，虚拟机可用它确定如何路由一个特定消息
queue:也称为message queue，消息队列，保存消息并将它们转发给消费者

3.rabbitmq的极速入门

后台启动: ./rabbitmq start &

关闭: ./rabbitmqctl stop

节点状态: ./rabbitmqctl status

管控台:

rabbitmq生产消费快速入门:

环境： springboot+jdk1.7+rabbitmq3.6.5 (maven依赖配置)

 <parent>
        <groupid>org.springframework.boot</groupid>
        <artifactid>spring-boot-starter-parent</artifactid>
        <version>1.5.9.release</version>
  </parent>
    <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.6.5</version>
        </dependency>
    </dependencies>

public class procuder {
    public static void main(string[] args) throws exception {
        
        //1.创建一个connectionfactory 并进行配置
        connectionfactory connectionfactory = new connectionfactory();
        connectionfactory.sethost("192.168.244.11");
        connectionfactory.setport(5672);
        connectionfactory.setvirtualhost("/");
        connectionfactory.sethandshaketimeout(20000);
        //2.通过连接工厂创建连接
        connection connection = connectionfactory.newconnection();
        
        //3.通过connection 创建一个 channel
        channel channel = connection.createchannel();
    
        /**
         * basicpublish(string exchange, string routingkey, basicproperties props, byte[] body)
         * exchange:指定交换机 不指定 则默认 （amqp default交换机） 通过routingkey进行匹配 
         * props 消息属性
         * body 消息体
         */
        //4.通过channel发送数据
        for(int i = 0; i < 5; i++){
          system.out.println("生产消息:" + i);
          string msg = "hello rabbitmq" + i;
          channel.basicpublish("", "test", null, msg.getbytes());
        }
        
        
        //5.记得关闭相关的连接
        channel.close();
        connection.close();
    }
}

public class consumer {
    public static void main(string[] args) throws exception{
                //1.创建一个connectionfactory 并进行配置
                connectionfactory connectionfactory = new connectionfactory();
                connectionfactory.sethost("192.168.244.11");
                connectionfactory.setport(5672);
                connectionfactory.setvirtualhost("/");
                connectionfactory.sethandshaketimeout(20000);
                //2.通过连接工厂创建连接
                connection connection = connectionfactory.newconnection();
                
                //3.通过connection 创建一个 channel
                channel channel = connection.createchannel();
                
                //4. 声明创建一个队列
                string queuename = "test";
                /**
                 * durable 是否持久化
                 * exclusive 独占的  相当于加了一把锁
                 */
                channel.queuedeclare(queuename,true,false,false,null);
                
                //5.创建消费者
                queueingconsumer queueingconsumer = new queueingconsumer(channel);
                
                //6.设置channel
                /**
                 * ack: 当一条消息从生产端发到消费端，消费端接收到消息后会马上回送一个ack信息给broker,告诉它这条消息收到了
                 * autoack: 
                 * true  自动签收 当消费者一收到消息就表示消费者收到了消息，消费者收到了消息就会立即从队列中删除。
                 * false 手动签收 当消费者收到消息在合适的时候来显示的进行确认，说我已经接收到了该消息了，rabbitmq可以从队列中删除该消息了
                 * 
                 */
                channel.basicconsume(queuename, true, queueingconsumer);
                
                //7.获取消息
                while(true){
                    delivery delivery = queueingconsumer.nextdelivery();
                    string msg = new string(delivery.getbody());
                    system.err.println("消费端:" + msg);
                    //envelope envelope = delivery.getenvelope();
                }
    }
}

4. exchange(交换机)详解

exchange: 接收消息，并根据路由键转发消息所绑定的队列

RabbitMQ 从入门到精通（一）

交换机属性:

name: 交换机名称
type: 交换机类型 diect、topic、fanout、headers
durability: 是否需要持久化，true为持久化
autodelete: 当最后一个绑定到exchange的队列删除后，自动删除该exchange
internal: 当前exchange是否用于rabbitmq内部使用，默认为false (百分之99的情况默认为false 除非对erlang语言较了解，做一些扩展)
arguments: 扩展参数, 用于扩展amqp协议可自定化使用

4.1 direct exchange

所有发送到direct exchange的消息被转发到routekey指定的queue

注意：direct模式可以使用rabbitmq自带的exchange: default exchange，所以不需要将exchange进行任何绑定(binding)操作，消息传递时，routingkey必须完全匹配才会被队列接收，否则该消息会被抛弃

RabbitMQ 从入门到精通（一）

public class producerdirectexchange {
    public static void main(string[] args) throws exception {
        //1.创建connectionfactory
        connectionfactory connectionfactory = new connectionfactory();
        connectionfactory.sethost("192.168.244.11");
        connectionfactory.setport(5672);
        connectionfactory.setvirtualhost("/");
        
        //2.创建connection
        connection connection = connectionfactory.newconnection();
        //3.创建channel
        channel channel = connection.createchannel();
        //4.声明
        string exchangename = "test_direct_exchange";
        string routingkey = "test.direct";
        //5.发送
        string msg = "hello world rabbitmq4 direct exchange message";
        channel.basicpublish(exchangename, routingkey, null, msg.getbytes());
    }
}

public class consumerdirectexchange {
    public static void main(string[] args) throws exception{
        connectionfactory connectionfactory = new connectionfactory();
        connectionfactory.sethost("192.168.244.11");
        connectionfactory.setport(5672);
        connectionfactory.setvirtualhost("/");
        connectionfactory.sethandshaketimeout(20000);
        connectionfactory.setautomaticrecoveryenabled(true);
        connectionfactory.setnetworkrecoveryinterval(3000);
        
        connection connection = connectionfactory.newconnection();
        
        channel channel = connection.createchannel();
        //声明
        string exchangename = "test_direct_exchange";
        string exchangetype = "direct";
        string queuename = "test_direct_queue";
        string routingkey = "test.direct";
        //表示声明了一个交换机
        channel.exchangedeclare(exchangename, exchangetype,true,false,false,null);
        //表示声明了一个队列
        channel.queuedeclare(queuename,false,false,false,null);
        //建立一个绑定关系
        channel.queuebind(queuename, exchangename, routingkey);
        
        //durable 是否持久化消息
        queueingconsumer consumer = new queueingconsumer(channel);
        //参数：队列名称,是否自动ack,consumer
        channel.basicconsume(queuename, true, consumer);
        
        //循环获取消息
        while(true){
            //获取消息,如果没有消息,这一步将会一直阻塞
            delivery delivery = consumer.nextdelivery();
            string msg = new string(delivery.getbody());
            system.out.println("收到消息:" + msg);
        }
    }
}

4.2 topic exchange

所有发送到topic exchange的消息被转发到所有关心routekey中指定topic的queue上

exchange将routekey和某topic进行模糊匹配，此时队列需要绑定一个topic

RabbitMQ 从入门到精通（一）

注意：可以使用通配符进行匹配

符号 # 匹配一个或多个词

符号 * 匹配不多不少一个词

例如: "log.#" 能够匹配到 “log.info.oa”

"log.*" 只会匹配到 "log.err"

public class producertopicexchange {
    public static void main(string[] args) throws exception {
        //1.创建connectionfactory
        connectionfactory connectionfactory = new connectionfactory();
        connectionfactory.sethost("192.168.244.11");
        connectionfactory.setport(5672);
        connectionfactory.setvirtualhost("/");
        connectionfactory.sethandshaketimeout(20000);

        //2.创建connection
        connection connection = connectionfactory.newconnection();
        //3.创建channel
        channel channel = connection.createchannel();
        //4.声明
        string exchangename = "test_topic_exchange";
        string routingkey1 = "user.save";
        string routingkey2 = "user.update";
        string routingkey3 = "user.delete.abc";
        //5.发送
        string msg = "hello world rabbitmq4 direct exchange message";
        channel.basicpublish(exchangename, routingkey1, null, msg.getbytes());
        channel.basicpublish(exchangename, routingkey2, null, msg.getbytes());
        channel.basicpublish(exchangename, routingkey3, null, msg.getbytes());
    }
}

public class consumertopicexchange {
    public static void main(string[] args) throws exception{
        connectionfactory connectionfactory = new connectionfactory();
        connectionfactory.sethost("192.168.244.11");
        connectionfactory.setport(5672);
        connectionfactory.setvirtualhost("/");
        connectionfactory.sethandshaketimeout(20000);
        connectionfactory.setautomaticrecoveryenabled(true);
        connectionfactory.setnetworkrecoveryinterval(3000);
        
        connection connection = connectionfactory.newconnection();
        
        channel channel = connection.createchannel();
        //声明
        string exchangename = "test_topic_exchange";
        string exchangetype = "topic";
        string queuename = "test_topic_queue";
        string routingkey = "user.#";
        //表示声明了一个交换机
        channel.exchangedeclare(exchangename, exchangetype,true,false,false,null);
        //表示声明了一个队列
        channel.queuedeclare(queuename,false,false,false,null);
        //建立一个绑定关系
        channel.queuebind(queuename, exchangename, routingkey);
        
        //durable 是否持久化消息
        queueingconsumer consumer = new queueingconsumer(channel);
        //参数：队列名称,是否自动ack,consumer
        channel.basicconsume(queuename, true, consumer);
        
        //循环获取消息
        while(true){
            //获取消息,如果没有消息,这一步将会一直阻塞
            delivery delivery = consumer.nextdelivery();
            string msg = new string(delivery.getbody());
            system.out.println("收到消息:" + msg);
        }
    }
}

4.3 fanout exchange

不处理路由键，只需要简单的将队列绑定到交换机上
发送到交换机的消息都会被转发到与该交换机绑定的所有队列上
所以fanout交换机转发消息是最快的

RabbitMQ 从入门到精通（一）

public class producerfanoutexchange {
    public static void main(string[] args) throws exception {
        //1.创建connectionfactory
        connectionfactory connectionfactory = new connectionfactory();
        connectionfactory.sethost("192.168.244.11");
        connectionfactory.setport(5672);
        connectionfactory.setvirtualhost("/");
        connectionfactory.sethandshaketimeout(20000);

        //2.创建connection
        connection connection = connectionfactory.newconnection();
        //3.创建channel
        channel channel = connection.createchannel();
        //4.声明
        string exchangename = "test_fanout_exchange";
        //5.发送
        for(int i = 0; i < 10 ; i++){
            string msg = "hello world rabbitmq4 direct exchange message";
            channel.basicpublish(exchangename, "", null, msg.getbytes());
        }
        channel.close();
        connection.close();
    }
}

public class consumerfanoutexchange {
    public static void main(string[] args) throws exception{
        connectionfactory connectionfactory = new connectionfactory();
        connectionfactory.sethost("192.168.244.11");
        connectionfactory.setport(5672);
        connectionfactory.setvirtualhost("/");
        connectionfactory.sethandshaketimeout(20000);
        connectionfactory.setautomaticrecoveryenabled(true);
        connectionfactory.setnetworkrecoveryinterval(3000);
        
        connection connection = connectionfactory.newconnection();
        
        channel channel = connection.createchannel();
        //声明
        string exchangename = "test_fanout_exchange";
        string exchangetype = "fanout";
        string queuename = "test_topic_queue";
        //无需指定路由key 
        string routingkey = "";
        //表示声明了一个交换机
        channel.exchangedeclare(exchangename, exchangetype,true,false,false,null);
        //表示声明了一个队列
        channel.queuedeclare(queuename,false,false,false,null);
        //建立一个绑定关系
        channel.queuebind(queuename, exchangename, routingkey);
        
        //durable 是否持久化消息
        queueingconsumer consumer = new queueingconsumer(channel);
        //参数：队列名称,是否自动ack,consumer
        channel.basicconsume(queuename, true, consumer);
        
        //循环获取消息
        while(true){
            //获取消息,如果没有消息,这一步将会一直阻塞
            delivery delivery = consumer.nextdelivery();
            string msg = new string(delivery.getbody());
            system.out.println("收到消息:" + msg);
        }
    }
}

5. message 消息

服务器与应用程序之间传递的数据，本质上就是一段数据，由properties和body组成

常用属性：delivery mode、headers (自定义属性)

其他属性：content_type、content_encoding、priority、expiration

消息的properties属性用法示例:

public class procuder {
    public static void main(string[] args) throws exception {
        
        //1.创建一个connectionfactory 并进行配置
        connectionfactory connectionfactory = new connectionfactory();
        connectionfactory.sethost("192.168.244.11");
        connectionfactory.setport(5672);
        connectionfactory.setvirtualhost("/");
        connectionfactory.sethandshaketimeout(20000);
        //2.通过连接工厂创建连接
        connection connection = connectionfactory.newconnection();
        
        //3.通过connection 创建一个 channel
        channel channel = connection.createchannel();
    
        map<string,object> headers = new hashmap<>();
        headers.put("my1", "111");
        headers.put("my2", "222");
        
        //10秒不消费 消息过期移除消息队列
        amqp.basicproperties properties = new amqp.basicproperties().builder()
                .deliverymode(2)
                .contentencoding("utf-8")
                .expiration("10000")
                .headers(headers)
                .build();
        
        //4.通过channel发送数据
        for(int i = 0; i < 5; i++){
          system.out.println("生产消息:" + i);
          string msg = "hello rabbitmq" + i;
          channel.basicpublish("", "test", properties, msg.getbytes());
        }
        
        
        //5.记得关闭相关的连接
        channel.close();
        connection.close();
    }
}

public class consumer {
    public static void main(string[] args) throws exception{
                //1.创建一个connectionfactory 并进行配置
                connectionfactory connectionfactory = new connectionfactory();
                connectionfactory.sethost("192.168.244.11");
                connectionfactory.setport(5672);
                connectionfactory.setvirtualhost("/");
                connectionfactory.sethandshaketimeout(20000);
                //2.通过连接工厂创建连接
                connection connection = connectionfactory.newconnection();
                
                //3.通过connection 创建一个 channel
                channel channel = connection.createchannel();
                
                //4. 声明创建一个队列
                string queuename = "test";
                channel.queuedeclare(queuename,true,false,false,null);
                
                //5.创建消费者
                queueingconsumer queueingconsumer = new queueingconsumer(channel);
                
                //6.设置channel
                channel.basicconsume(queuename, true, queueingconsumer);
                
                //7.获取消息
                while(true){
                    delivery delivery = queueingconsumer.nextdelivery();
                    string msg = new string(delivery.getbody());
                    system.err.println("消费端:" + msg);
                    
                    map<string, object> headers = delivery.getproperties().getheaders();
                    system.err.println("headers value:" + headers.get("my1"));
                }
    }
}

上一篇： Haproxy+keepalive-群集架构实验

下一篇： scss混合(mixins)使用

RabbitMQ 从入门到精通（一）

1. 初识rabbitmq

2. amqp

3.rabbitmq的极速入门

4. exchange(交换机)详解

4.1 direct exchange

4.2 topic exchange

4.3 fanout exchange

5. message 消息

PHP 怎么计算从某一天到某一天一共的天数然后增加？

PHP之文件 php7 php环境搭建 php从入门到精通

入门到精通python要多久

从.NET CORE2.2升级到3.0过程及遇到的一些问题

Linux下安装Mysql多实例作为数据备份服务器实现多主到一从多实例的备份

从零到一详聊创建Vue工程及遇到的常见问题

Linux下安装Mysql多实例作为数据备份服务器实现多主到一从多实例的备份

详解React服务端渲染从入门到精通

Mysql 流增量写入 Hdfs（一） --从 mysql 到 kafka

Javascript实现一朵从含苞到绽放的玫瑰

RabbitMQ 从入门到精通 （一）

1. 初识rabbitmq

2. amqp

3.rabbitmq的极速入门

4. exchange(交换机)详解

4.1 direct exchange

4.2 topic exchange

4.3 fanout exchange

5. message 消息

PHP 怎么计算从某一天到某一天一共的天数然后增加？

PHP之文件 php7 php环境搭建 php从入门到精通

入门到精通python要多久

从.NET CORE2.2升级到3.0过程及遇到的一些问题

Linux下安装Mysql多实例作为数据备份服务器实现多主到一从多实例的备份

从零到一详聊创建Vue工程及遇到的常见问题

Linux下安装Mysql多实例作为数据备份服务器实现多主到一从多实例的备份

详解React服务端渲染从入门到精通

Mysql 流增量写入 Hdfs（一） --从 mysql 到 kafka

Javascript实现一朵从含苞到绽放的玫瑰

RabbitMQ 从入门到精通（一）