热门标签 | 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:手动应答,当从队列中取出消息后,需要程序员手动调用方法应答,如果没有应答,该消息还会再放进队列中,就会出现该消息一直没有被消费掉的现象

  


推荐阅读
  • 深入解析Java枚举及其高级特性
    本文详细介绍了Java枚举的概念、语法、使用规则和应用场景,并探讨了其在实际编程中的高级应用。所有相关内容已收录于GitHub仓库[JavaLearningmanual](https://github.com/Ziphtracks/JavaLearningmanual),欢迎Star并持续关注。 ... [详细]
  • 本文介绍了如何使用JavaScript的Fetch API与Express服务器进行交互,涵盖了GET、POST、PUT和DELETE请求的实现,并展示了如何处理JSON响应。 ... [详细]
  • 深入解析Spring启动过程
    本文详细介绍了Spring框架的启动流程,帮助开发者理解其内部机制。通过具体示例和代码片段,解释了Bean定义、工厂类、读取器以及条件评估等关键概念,使读者能够更全面地掌握Spring的初始化过程。 ... [详细]
  • 并发编程 12—— 任务取消与关闭 之 shutdownNow 的局限性
    Java并发编程实践目录并发编程01——ThreadLocal并发编程02——ConcurrentHashMap并发编程03——阻塞队列和生产者-消费者模式并发编程04——闭锁Co ... [详细]
  • 深入解析SpringMVC核心组件:DispatcherServlet的工作原理
    本文详细探讨了SpringMVC的核心组件——DispatcherServlet的运作机制,旨在帮助有一定Java和Spring基础的开发人员理解HTTP请求是如何被映射到Controller并执行的。文章将解答以下问题:1. HTTP请求如何映射到Controller;2. Controller是如何被执行的。 ... [详细]
  • 在高并发需求的C++项目中,我们最初选择了JsonCpp进行JSON解析和序列化。然而,在处理大数据量时,JsonCpp频繁抛出异常,尤其是在多线程环境下问题更为突出。通过分析发现,旧版本的JsonCpp存在多线程安全性和性能瓶颈。经过评估,我们最终选择了RapidJSON作为替代方案,并实现了显著的性能提升。 ... [详细]
  • 深入解析 Android IPC 中的 Messenger 机制
    本文详细介绍了 Android 中基于消息传递的进程间通信(IPC)机制——Messenger。通过实例和源码分析,帮助开发者更好地理解和使用这一高效的通信工具。 ... [详细]
  • Spring Boot单元测试中Redis连接失败的解决方案
    本文探讨了在Spring Boot项目中进行单元测试时遇到Redis连接问题的原因及解决方法,详细分析了配置文件加载路径不当导致的问题,并提供了有效的解决方案。 ... [详细]
  • Java 实现二维极点算法
    本文介绍了一种使用 Java 编程语言实现的二维极点算法。该算法用于从一组二维坐标中筛选出极点,适用于需要处理几何图形和空间数据的应用场景。文章不仅详细解释了算法的工作原理,还提供了完整的代码示例。 ... [详细]
  • 本题来自WC2014,题目编号为BZOJ3435、洛谷P3920和UOJ55。该问题描述了一棵不断生长的带权树及其节点上小精灵之间的友谊关系,要求实时计算每次新增节点后树上所有可能的朋友对数。 ... [详细]
  • 本文详细介绍了 Java 中 org.geotools.data.shapefile.ShapefileDataStore 类的 getCurrentTypeName() 方法,并提供了多个代码示例,帮助开发者更好地理解和使用该方法。 ... [详细]
  • 深入解析动态代理模式:23种设计模式之三
    在设计模式中,动态代理模式是应用最为广泛的一种代理模式。它允许我们在运行时动态创建代理对象,并在调用方法时进行增强处理。本文将详细介绍动态代理的实现机制及其应用场景。 ... [详细]
  • 深入解析ArrayList与LinkedList的差异
    本文详细对比了Java中ArrayList和LinkedList两种常用集合类的特性、性能及适用场景,通过代码示例进行测试,并结合实际应用场景分析其优缺点。 ... [详细]
  • 为了解决不同服务器间共享图片的需求,我们最初考虑建立一个FTP图片服务器。然而,考虑到项目是一个简单的CMS系统,为了简化流程,团队决定探索七牛云存储的解决方案。本文将详细介绍使用七牛云存储的过程和心得。 ... [详细]
  • 由二叉树到贪心算法
    二叉树很重要树是数据结构中的重中之重,尤其以各类二叉树为学习的难点。单就面试而言,在 ... [详细]
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社区 版权所有