热门标签 | HotTags
当前位置:  开发笔记 > 编程语言 > 正文

RabbitMQ(二):入门案例

一、功能功能描述:生产者将消息发送到队列(队列的名字为hello)中,消费者从队列中获取消息。二、生产者定义队列名publicfinalstatic

一、功能

功能描述:生产者将消息发送到队列(队列的名字为hello)中,消费者从队列中获取消息。

这里写图片描述

二、生产者
    // 定义队列名
    public final static String QUEUE_NAME="rabbitMQ.zhxb";
    public void rabbitMQ (){
        try {
            // 创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            // 设置RabbitMQ参数
            factory.setHost("IP");
            factory.setUsername("*");
            factory.setPassword("*");
            factory.setPort(*);
            // 创建一个连接
            Connection cOnnection= factory.newConnection();
            // 创建一个通道(大连接中的一根线)
            Channel channel = connection.createChannel();
            // 声明一个队列
            // 第一个参数表示队列名称、第二个参数为是否持久化、第三个参数为是否是独占队列、第四个参数为当所有消费者客户端连接断开时是否自动删除队列、第五个参数为队列的其他参数
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            // 定义一个消息
            String message = "Hello RabbitMQ";
            // 发送消息到队列
       // 第一个参数为交换机名称、第二个参数为队列映射的路由key、第三个参数为消息的其他属性、第四个参数为发送信息的主体
channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8")); System.out.println("我向"+QUEUE_NAME+"发送一个消息:+'" + message); // 关闭通道和连接 channel.close(); connection.close(); } catch (IOException e) { e.printStackTrace(); } catch (TimeoutException e) { e.printStackTrace(); } }
步骤说明:

连接工厂:生产Connection的的工厂

连接:RabbitMQ的socket链接,它封装了socket协议

通道:把Connection比作一条光纤电缆的话,那么Channel信道就比作成光纤电缆中的其中一束光纤

声明队列:rabbitMQ.zhxb,储存消息的容器(非持久化)

发送消息:将消息发送到交换机,然后交换机通过路由策略(规则)将消息路由到匹配的队列中去

步骤详解:

1.声明一个队列

// 声明一个队列

// 第一个参数表示队列名称、第二个参数为是否持久化、第三个参数为是否是独占队列、第四个参数为当所有消费者客户端连接断开时是否自动删除队列、第五个参数为队列的其他参数

channel.queueDeclare(QUEUE_NAME, false, false, false, null);

参数详解
queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments);
  • queue: 队列名称
  • durable:是否持久化(true表示是,队列将在服务器重启时生存), 队列的声明默认是存放到内存中的,如果rabbitmq重启会丢失,如果想重启之后还存在就要使队列持久化,保存到Erlang自带的Mnesia数据库中,当rabbitmq重启之后会读取该数据库
  • exclusive:是否排外的/是否是独占队列,有两个作用(创建者可以使连接断开后自动删除,可使其变为某一消费者的私有队列)一:当连接关闭时connection.close()该队列是否会自动删除;二:该队列是否是私有的private,如果不是排外的,可以使用两个消费者都访问同一个队列,没有任何问题,如果是排外的,会对当前队列加锁,其他通道channel是不能访问的,如果强制访问会报异常:com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method(reply-code=405, reply-text=RESOURCE_LOCKED - cannot obtain exclusive access to locked queue 'queue_name' in vhost '/', class-id=50, method-id=20)一般等于true的话用于一个队列只能有一个消费者来消费的场景*/
  • autoDelete:是当所有消费者客户端连接断开时是否自动删除队列,当最后一个消费者断开连接之后队列是否自动被删除,可以通过RabbitMQ Management,查看某个队列的消费者数量,当cOnsumers= 0时队列就会自动删除
  • arguments:其他属性(消息什么时候会自动被删除、自动过期、最长长度、最大长度字节、死信交换、死信路由键、最高优先级、懒惰模式)
  • 注意:关于队列的声明,如果使用同一套参数进行声明了,就不能再使用其他参数来声明,要么删除该队列重新删除,可以使用命令行删除也可以在RabbitMQ Management上删除,要么给队列重新起一个名字。
arguments具体说明
  • Message TTL:队列中的消息什么时候会自动被删除,(x-message-ttl)
  • Auto expire:自动过期,(x-expires):
  • Max length:最长长度,(x-max-length)
  • Max length bytes:最大长度字节,(x-max-length-bytes)
  • Dead letter exchange:死信交换,(x-dead-letter-exchange)
  • Dead letter routing key:死信路由键,(x-dead-letter-routing-key)
  • Maximum priority:最高优先级,(x-max-priority)
  • Lazy mode:懒惰模式,(x-queue-mode=lazy)
  • Master locator(x-queue-master-locator)
官方解释:
  • 发布到队列的消息在丢弃之前可以存活多长时间(毫秒)。(设置“ x-message-ttl ”参数。)
  • 在自动删除队列(毫秒)之前,队列可以使用多长时间。(设置“ x-expires ”参数。)
  • 在队列开始从队列中删除之前,队列可以包含多少(就绪)消息。(设置“ x-max-length ”参数。)
  • 队列在开始从头部删除之前可以包含的就绪消息的总体大小。(设置“ x-max-length-bytes ”参数。)
  • 如果邮件被拒绝或过期,将重新发布邮件的交换的可选名称。(设置“ x-dead-letter-exchange ”参数。)
  • 邮件是无效的时候使用的可选替换路由密钥。如果未设置,将使用消息的原始路由密钥。(设置“ x-dead-letter-routing-key ”参数。)
  • 要支持的队列的最大优先级数; 如果未设置,则队列将不支持消息优先级。(设置“ x-max-priority ”参数。)
  • 将队列设置为延迟模式,在磁盘上保留尽可能多的消息以减少RAM使用; 如果未设置,队列将保留内存缓存以尽快传递消息。(设置“ x-queue-mode ”参数。)
增加解释:
  • 设置队列中的所有消息的生存周期(统一为整个队列的所有消息设置生命周期),单位毫秒, 类似于redis中的ttl,生存时间到了,消息会被从队里中删除,注意是消息被删除,而不是队列被删除, 特性Features=TTL,也可以在发布消息的时候单独为某个消息指定剩余生存时间(但在消息发送的设置里)
  • 当队列在指定的时间没有被访问(consume, basicGet, queueDeclare…)就会被删除,Features=Exp
  • 限定队列的消息的最大值长度,超过指定长度将会把最早的几条删除掉, 类似于mongodb中的固定集合,例如保存最新的100条消息, Feature=Lim
  • 限定队列最大占用的空间大小, 一般受限于内存、磁盘的大小, Features=Lim B
  • 当队列消息长度大于最大长度、或者过期的等,将从队列中删除的消息推送到指定的交换机中去而不是丢弃掉,Features=DLX
  • 将删除的消息推送到指定交换机的指定路由键的队列中去, Feature=DLK
  • 优先级队列,声明队列时先定义最大优先级值(定义最大值一般不要太大),在发布消息的时候指定该消息的优先级, 优先级更高(数值更大的)的消息先被消费,
  • 先将消息保存到磁盘上,不放在内存中,当消费者开始消费的时候才加载到内存中

2.发送消息到队列

// 发送消息到队列
// 第一个参数为交换机名称、第二个参数为队列映射的路由key、第三个参数为消息的其他属性、第四个参数为发送信息的主体
channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
void basicPublish(String neme, String routingKey, boolean mandatory, boolean immediate, BasicProperties properties, byte[] message) throws IOException;
  • neme:为交换机名称
  • routingKey:路由键(队列映射的路由key),#匹配0个或多个单词,*匹配一个单词,在topic exchange做消息转发用
  • mandatory:true:如果exchange根据自身类型和消息routeKey无法找到一个符合条件的queue,那么会调用basic.return方法将消息返还给生产者。false:出现上述情形broker会直接将消息扔掉
  • immediate:true:如果exchange在将消息route到queue(s)时发现对应的queue上没有消费者,那么这条消息不会放入队列中。当与消息routeKey关联的所有queue(一个或多个)都没有消费者时,该消息会通过basic.return方法返还给生产者。(简单来说:mandatory标志告诉服务器至少将该消息route到一个队列中,否则将消息返还给生产者;immediate标志告诉服务器如果该消息关联的queue上有消费者,则马上将消息投递给它,如果所有queue都没有消费者,直接把消息返还给生产者,不用将消息入队列等待消费者了。)
  • properties :需要注意的是BasicProperties.deliveryMode,0:不持久化 1:持久化 这里指的是消息的持久化,配合channel(durable=true),queue(durable)可以实现,即使服务器宕机,消息仍然保留
  • message:消息主体
// 给某个消息设置存在时间
AMQP.BasicProperties.Builder properties = new AMQP.BasicProperties().builder().expiration("6000");
channel.basicPublish("", QUEUE_NAME, properties.build(), message.getBytes("UTF-8"));

3.强调一个概念:

  • 持久化
  • 队列持久化:重启RabbitMQ服务器队列还在
  • 消息持久化:重启RabbitMQ服务器消息还在(设置消息持久化必须先设置队列持久化,要不然队列不持久化,队列都不存在了,消息存在还有什么意义。消息持久化需要将交换机持久化、队列持久化、消息持久化,才能最终达到持久化的目的)
三、消费者
    // 定义队列名
    private final static String QUEUE_NAME = "rabbitMQ.zhxb";
    public void rabbitMQCustomer() {
        try {
            // 创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            // 设置RabbitMQ参数
            factory.setHost("IP");
            factory.setUsername("*");
            factory.setPassword("*");
            factory.setPort(*);
            // 创建一个连接
            Connection cOnnection= factory.newConnection();
            // 创建一个通道
            Channel channel = connection.createChannel();
            // 声明一个队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            System.out.println("消费者一,等待消息产生");
            // DefaultConsumer类实现了Consumer接口,通过传入一个频道,
            // 告诉服务器我们需要那个频道的消息,如果频道中有消息,就会执行回调函数handleDelivery
            // 消费一个消息
            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);
                    try {
                        Thread thread = Thread.currentThread();
                        thread.sleep(5000);//暂停1.5秒后程序继续执行
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
            // 自动应答回复队列(RabbitMQ中的消息确认机制)
            channel.basicConsume(QUEUE_NAME, true, consumer);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
步骤详解:

1.声明一个队列

// 声明一个队列
// 第一个参数表示队列名称、第二个参数为是否持久化、第三个参数为是否是独占队列、第四个参数为当所有消费者客户端连接断开时是否自动删除队列、第五个参数为队列的其他参数
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 规则一单指定,必须保持一致

2.消费一个消息(订阅方式)

 
 
// 消费一个消息
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);
}
};

订阅方式:
  • 订阅方式其实是向queue注册consumer,通过rpc向queue server发送注册consumer的消息,
  • rabbitMQ Server在收到消息后,根据消息的内容类型判断这是一个订阅消息,
  • 当MQ 中queue有消息时,会自动把消息通过该socket(长连接)通道发送出去。
消费原理:
  • DefaultConsumer类实现了Consumer接口,
  • consumer通过建立到queue的连接connection,创建channel对象,
  • 通过传入这个频道,告诉服务器我们需要那个频道的消息,如果频道中有消息,就会执行回调函数handleDelivery来消费这个消息

3.RabbitMQ中的消息确认机制

// 自动应答回复队列(RabbitMQ中的消息确认机制)
channel.basicConsume(QUEUE_NAME, true, consumer);

原理:
  • true:自动应答,即消费者获取到消息,该消息就会从队列中删除掉,
  • false:手动应答,当从队列中取出消息后,需要程序员手动调用方法应答,如果没有应答,该消息还会再放进队列中,就会出现该消息一直没有被消费掉的现象

  


推荐阅读
  • 阿,里,云,物,联网,net,core,客户端,czgl,aliiotclient, ... [详细]
  • Spring特性实现接口多类的动态调用详解
    本文详细介绍了如何使用Spring特性实现接口多类的动态调用。通过对Spring IoC容器的基础类BeanFactory和ApplicationContext的介绍,以及getBeansOfType方法的应用,解决了在实际工作中遇到的接口及多个实现类的问题。同时,文章还提到了SPI使用的不便之处,并介绍了借助ApplicationContext实现需求的方法。阅读本文,你将了解到Spring特性的实现原理和实际应用方式。 ... [详细]
  • 本文分享了一个关于在C#中使用异步代码的问题,作者在控制台中运行时代码正常工作,但在Windows窗体中却无法正常工作。作者尝试搜索局域网上的主机,但在窗体中计数器没有减少。文章提供了相关的代码和解决思路。 ... [详细]
  • 本文讨论了在openwrt-17.01版本中,mt7628设备上初始化启动时eth0的mac地址总是随机生成的问题。每次随机生成的eth0的mac地址都会写到/sys/class/net/eth0/address目录下,而openwrt-17.01原版的SDK会根据随机生成的eth0的mac地址再生成eth0.1、eth0.2等,生成后的mac地址会保存在/etc/config/network下。 ... [详细]
  • 在Android开发中,使用Picasso库可以实现对网络图片的等比例缩放。本文介绍了使用Picasso库进行图片缩放的方法,并提供了具体的代码实现。通过获取图片的宽高,计算目标宽度和高度,并创建新图实现等比例缩放。 ... [详细]
  • 本文介绍了在rhel5.5操作系统下搭建网关+LAMP+postfix+dhcp的步骤和配置方法。通过配置dhcp自动分配ip、实现外网访问公司网站、内网收发邮件、内网上网以及SNAT转换等功能。详细介绍了安装dhcp和配置相关文件的步骤,并提供了相关的命令和配置示例。 ... [详细]
  • 本文介绍了Redis的基础数据结构string的应用场景,并以面试的形式进行问答讲解,帮助读者更好地理解和应用Redis。同时,描述了一位面试者的心理状态和面试官的行为。 ... [详细]
  • 计算机存储系统的层次结构及其优势
    本文介绍了计算机存储系统的层次结构,包括高速缓存、主存储器和辅助存储器三个层次。通过分层存储数据可以提高程序的执行效率。计算机存储系统的层次结构将各种不同存储容量、存取速度和价格的存储器有机组合成整体,形成可寻址存储空间比主存储器空间大得多的存储整体。由于辅助存储器容量大、价格低,使得整体存储系统的平均价格降低。同时,高速缓存的存取速度可以和CPU的工作速度相匹配,进一步提高程序执行效率。 ... [详细]
  • 本文介绍了计算机网络的定义和通信流程,包括客户端编译文件、二进制转换、三层路由设备等。同时,还介绍了计算机网络中常用的关键词,如MAC地址和IP地址。 ... [详细]
  • Tomcat/Jetty为何选择扩展线程池而不是使用JDK原生线程池?
    本文探讨了Tomcat和Jetty选择扩展线程池而不是使用JDK原生线程池的原因。通过比较IO密集型任务和CPU密集型任务的特点,解释了为何Tomcat和Jetty需要扩展线程池来提高并发度和任务处理速度。同时,介绍了JDK原生线程池的工作流程。 ... [详细]
  • 开发笔记:Java是如何读取和写入浏览器Cookies的
    篇首语:本文由编程笔记#小编为大家整理,主要介绍了Java是如何读取和写入浏览器Cookies的相关的知识,希望对你有一定的参考价值。首先我 ... [详细]
  • 本文介绍了在Linux下安装和配置Kafka的方法,包括安装JDK、下载和解压Kafka、配置Kafka的参数,以及配置Kafka的日志目录、服务器IP和日志存放路径等。同时还提供了单机配置部署的方法和zookeeper地址和端口的配置。通过实操成功的案例,帮助读者快速完成Kafka的安装和配置。 ... [详细]
  • Python SQLAlchemy库的使用方法详解
    本文详细介绍了Python中使用SQLAlchemy库的方法。首先对SQLAlchemy进行了简介,包括其定义、适用的数据库类型等。然后讨论了SQLAlchemy提供的两种主要使用模式,即SQL表达式语言和ORM。针对不同的需求,给出了选择哪种模式的建议。最后,介绍了连接数据库的方法,包括创建SQLAlchemy引擎和执行SQL语句的接口。 ... [详细]
  • 海马s5近光灯能否直接更换为H7?
    本文主要介绍了海马s5车型的近光灯是否可以直接更换为H7灯泡,并提供了完整的教程下载地址。此外,还详细讲解了DSP功能函数中的数据拷贝、数据填充和浮点数转换为定点数的相关内容。 ... [详细]
  • RouterOS 5.16软路由安装图解教程
    本文介绍了如何安装RouterOS 5.16软路由系统,包括系统要求、安装步骤和登录方式。同时提供了详细的图解教程,方便读者进行操作。 ... [详细]
author-avatar
kiki俏佳人2502909673
这个家伙很懒,什么也没留下!
PHP1.CN | 中国最专业的PHP中文社区 | DevBox开发工具箱 | json解析格式化 |PHP资讯 | PHP教程 | 数据库技术 | 服务器技术 | 前端开发技术 | PHP框架 | 开发工具 | 在线工具
Copyright © 1998 - 2020 PHP1.CN. All Rights Reserved | 京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有