RabbitMQ系列(三)RabbitMQ交换器Exchange介绍与实践

时间:2022-09-09 20:30:41

RabbitMQ交换器Exchange介绍与实践

RabbitMQ系列文章

  1. RabbitMQ在Ubuntu上的环境搭建
  2. 深入了解RabbitMQ工作原理及简单使用
  3. RabbitMQ交换器Exchange介绍与实践
  4. RabbitMQ事务和Confirm发送方消息确认——深入解读
  5. 使用Docker部署RabbitMQ集群
  6. 你不知道的RabbitMQ集群架构全解

导读

有了Rabbit的基础知识之后(基础知识详见:深入解读RabbitMQ工作原理及简单使用),本章我们重点学习一下Rabbit里面的exchange(交换器)的知识。

交换器分类

RabbitMQ的Exchange(交换器)分为四类:

  • direct(默认)
  • headers
  • fanout
  • topic

其中headers交换器允许你匹配AMQP消息的header而非路由键,除此之外headers交换器和direct交换器完全一致,但性能却很差,几乎用不到,所以我们本文也不做讲解。

注意:fanout、topic交换器是没有历史数据的,也就是说对于中途创建的队列,获取不到之前的消息。

1、direct交换器

direct为默认的交换器类型,也非常的简单,如果路由键匹配的话,消息就投递到相应的队列,如图:

RabbitMQ系列(三)RabbitMQ交换器Exchange介绍与实践

使用代码:channel.basicPublish("", QueueName, null, message)推送direct交换器消息到对于的队列,空字符为默认的direct交换器,用队列名称当做路由键。

direct交换器代码示例

发送端:

Connection conn = connectionFactoryUtil.GetRabbitConnection();
Channel channel = conn.createChannel();
// 声明队列【参数说明:参数一:队列名称,参数二:是否持久化;参数三:是否独占模式;参数四:消费者断开连接时是否删除队列;参数五:消息其他参数】
channel.queueDeclare(config.QueueName, false, false, false, null);
String message = String.format("当前时间:%s", new Date().getTime());
// 推送内容【参数说明:参数一:交换机名称;参数二:队列名称,参数三:消息的其他属性-路由的headers信息;参数四:消息主体】
channel.basicPublish("", config.QueueName, null, message.getBytes("UTF-8"));

接收端,持续接收消息:

Connection conn = connectionFactoryUtil.GetRabbitConnection();
Channel channel = conn.createChannel();
// 声明队列【参数说明:参数一:队列名称,参数二:是否持久化;参数三:是否独占模式;参数四:消费者断开连接时是否删除队列;参数五:消息其他参数】
channel.queueDeclare(config.QueueName, false, false, false, null);
Consumer defaultConsumer = 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);
channel.basicAck(envelope.getDeliveryTag(), false); // 手动确认消息【参数说明:参数一:该消息的index;参数二:是否批量应答,true批量确认小于当前id的消息】
}
};
channel.basicConsume(config.QueueName, false, "", defaultConsumer);

接收端,获取单条消息

Connection conn = connectionFactoryUtil.GetRabbitConnection();
Channel channel = conn.createChannel();
channel.queueDeclare(config.QueueName, false, false, false, null);
GetResponse resp = channel.basicGet(config.QueueName, false);
String message = new String(resp.getBody(), "UTF-8");
channel.basicAck(resp.getEnvelope().getDeliveryTag(), false); // 消息确认

持续消息获取使用:basic.consume;单个消息获取使用:basic.get。

注意:不能使用for循环单个消息消费来替代持续消息消费,因为这样性能很低;

公平调度

当接收端订阅者有多个的时候,direct会轮询公平的分发给每个订阅者(订阅者消息确认正常),如图:

RabbitMQ系列(三)RabbitMQ交换器Exchange介绍与实践

消息的发后既忘特性

发后既忘模式是指接受者不知道消息的来源,如果想要指定消息的发送者,需要包含在发送内容里面,这点就像我们在信件里面注明自己的姓名一样,只有这样才能知道发送者是谁。

消息确认

看了上面的代码我们可以知道,消息接收到之后必须使用channel.basicAck()方法手动确认(非自动确认删除模式下),那么问题来了。

消息收到未确认会怎么样?

如果应用程序接收了消息,因为bug忘记确认接收的话,消息在队列的状态会从“Ready”变为“Unacked”,如图:

RabbitMQ系列(三)RabbitMQ交换器Exchange介绍与实践

如果消息收到却未确认,Rabbit将不会再给这个应用程序发送更多的消息了,这是因为Rabbit认为你没有准备好接收下一条消息。

此条消息会一直保持Unacked的状态,直到你确认了消息,或者断开与Rabbit的连接,Rabbit会自动把消息改完Ready状态,分发给其他订阅者。

当然你可以利用这一点,让你的程序延迟确认该消息,直到你的程序处理完相应的业务逻辑,这样可以有效的防治Rabbit给你过多的消息,导致程序崩溃。

消息确认Demo:

Connection conn = connectionFactoryUtil.GetRabbitConnection();
Channel channel = conn.createChannel();
channel.queueDeclare(config.QueueName, false, false, false, null);
GetResponse resp = channel.basicGet(config.QueueName, false);
String message = new String(resp.getBody(), "UTF-8");
channel.basicAck(resp.getEnvelope().getDeliveryTag(), false);

channel.basicAck(long deliveryTag, boolean multiple)为消息确认,参数1:消息的id;参数2:是否批量应答,true批量确认小于次id的消息。

总结:消费者消费的每条消息都必须确认。

消息拒绝

消息在确认之前,可以有两个选择:

选择1:断开与Rabbit的连接,这样Rabbit会重新把消息分派给另一个消费者;

选择2:拒绝Rabbit发送的消息使用channel.basicReject(long deliveryTag, boolean requeue),参数1:消息的id;参数2:处理消息的方式,如果是true,Rabbib会重新分配这个消息给其他订阅者,如果设置成false的话,Rabbit会把消息发送到一个特殊的“死信”队列,用来存放被拒绝而不重新放入队列的消息。

消息拒绝Demo:

Connection conn = connectionFactoryUtil.GetRabbitConnection();
Channel channel = conn.createChannel();
channel.queueDeclare(config.QueueName, false, false, false, null);
GetResponse resp = channel.basicGet(config.QueueName, false);
String message = new String(resp.getBody(), "UTF-8");
channel.basicReject(resp.getEnvelope().getDeliveryTag(), true); //消息拒绝

2、fanout交换器——发布/订阅模式

fanout有别于direct交换器,fanout是一种发布/订阅模式的交换器,当你发送一条消息的时候,交换器会把消息广播到所有附加到这个交换器的队列上。

比如用户上传了自己的头像,这个时候图片需要清除缓存,同时用户应该得到积分奖励,你可以把这两个队列绑定到图片上传的交换器上,这样当有第三个、第四个上传完图片需要处理的需求的时候,原来的代码可以不变,只需要添加一个订阅消息即可,这样发送方和消费者的代码完全解耦,并可以轻而易举的添加新功能了。

和direct交换器不同,我们在发送消息的时候新增channel.exchangeDeclare(ExchangeName, "fanout"),这行代码声明fanout交换器。

发送端:

final String ExchangeName = "fanoutec"; // 交换器名称
Connection conn = connectionFactoryUtil.GetRabbitConnection();
Channel channel = conn.createChannel();
channel.exchangeDeclare(ExchangeName, "fanout"); // 声明fanout交换器
String message = "时间:" + new Date().getTime();
channel.basicPublish(ExchangeName, "", null, message.getBytes("UTF-8"));

接受消息不同于direct,我们需要声明fanout路由器,并使用默认的队列绑定到fanout交换器上。

接收端:

Connection conn = connectionFactoryUtil.GetRabbitConnection();
Channel channel = conn.createChannel();
channel.exchangeDeclare(ExchangeName, "fanout"); // 声明fanout交换器
String queueName = channel.queueDeclare().getQueue(); // 声明队列
channel.queueBind(queueName, ExchangeName, "");
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");
}
};
channel.basicConsume(queueName, true, consumer);

fanout和direct的区别最多的在接收端,fanout需要绑定队列到对应的交换器用于订阅消息。

其中channel.queueDeclare().getQueue()为随机队列,Rabbit会随机生成队列名称,一旦消费者断开连接,该队列会自动删除。

注意:对于fanout交换器来说routingKey(路由键)是无效的,这个参数是被忽略的。

3、topic交换器——匹配订阅模式

最后介绍的是topic交换器,topic交换器运行和fanout类似,但是可以更灵活的匹配自己想要订阅的信息,这个时候routingKey路由键就排上用场了,使用路由键进行消息(规则)匹配。

假设我们现在有一个日志系统,会把所有日志级别的日志发送到交换器,warning、log、error、fatal,但我们只想处理error以上的日志,要怎么处理?这就需要使用topic路由器了。

topic路由器的关键在于定义路由键,定义routingKey名称不能超过255字节,使用“.”作为分隔符,例如:com.mq.rabbit.error。

消费消息的时候routingKey可以使用下面字符匹配消息:

  • "*"匹配一个分段(用“.”分割)的内容;
  • "#"匹配0和多个字符;

例如发布了一个“com.mq.rabbit.error”的消息:

能匹配上的路由键:

  • cn.mq.rabbit.*
  • cn.mq.rabbit.#
  • #.error
  • cn.mq.#
  • #

不能匹配上的路由键:

  • cn.mq.*
  • *.error
  • *

所以如果想要订阅所有消息,可以使用“#”匹配。

注意:fanout、topic交换器是没有历史数据的,也就是说对于中途创建的队列,获取不到之前的消息。

发布端:

String routingKey = "com.mq.rabbit.error";
Connection conn = connectionFactoryUtil.GetRabbitConnection();
Channel channel = conn.createChannel();
channel.exchangeDeclare(ExchangeName, "topic"); // 声明topic交换器
String message = "时间:" + new Date().getTime();
channel.basicPublish(ExchangeName, routingKey, null, message.getBytes("UTF-8"));

接收端:

Connection conn = connectionFactoryUtil.GetRabbitConnection();
Channel channel = conn.createChannel();
channel.exchangeDeclare(ExchangeName, "topic"); // 声明topic交换器
String queueName = channel.queueDeclare().getQueue(); // 声明队列
String routingKey = "#.error";
channel.queueBind(queueName, ExchangeName, routingKey);
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(routingKey + "|接收消息 => " + message);
}
};
channel.basicConsume(queueName, true, consumer);

扩展部分—自定义线程池

如果需要更大的控制连接,用户可自己设置线程池,代码如下:

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; ExecutorService es = Executors.newFixedThreadPool(20);
Connection conn = factory.newConnection(es);

其实看过源码的同学可能知道,factory.newConnection本身默认也有线程池的机制,ConnectionFactory.class部分源码如下:

private ExecutorService sharedExecutor;
public Connection newConnection() throws IOException, TimeoutException {
return newConnection(this.sharedExecutor, Collections.singletonList(new Address(getHost(), getPort())));
}
public void setSharedExecutor(ExecutorService executor) {
this.sharedExecutor = executor;
}

其中this.sharedExecutor就是默认的线程池,可以通过setSharedExecutor()方法设置ConnectionFactory的线程池,如果不设置则为null。

用户如果自己设置了线程池,像本小节第一段代码写的那样,那么当连接关闭的时候,不会自动关闭用户自定义的线程池,所以用户必须自己手动关闭,通过调用shutdown()方法,否则可能会阻止JVM的终止。

官方的建议是只有在程序出现严重性能瓶颈的时候,才应该考虑使用此功能。

项目地址

GitHub:https://github.com/vipstone/rabbitmq-java.git

RabbitMQ系列(三)RabbitMQ交换器Exchange介绍与实践的更多相关文章

  1. RabbitMQ交换器Exchange介绍与实践

    RabbitMQ交换器Exchange介绍与实践 RabbitMQ系列文章 RabbitMQ在Ubuntu上的环境搭建 深入了解RabbitMQ工作原理及简单使用 RabbitMQ交换器Exchang ...

  2. RabbitMQ系列(三)--Java API

    基于java使用RabbitMQ 框架:SpringBoot1.5.14.RELEASE maven依赖: <dependency> <groupId>com.rabbitmq ...

  3. RabbitMQ系列之RabbitMQ单机安装

    安装epel源 rpm -ivh http://download.fedoraproject.org/pub/epel/6/x86_64/epel-release-6-8.noarch.rpm #ht ...

  4. rabbitmq系列三 之发布&sol;订阅

    1.发布/订阅 在上篇教程中,我们搭建了一个工作队列,每个任务只分发给一个工作者(worker).在本篇教程中,我们要做的跟之前完全不一样 —— 分发一个消息给多个消费者(consumers).这种模 ...

  5. 中间件系列三 RabbitMQ之交换机的四种类型和属性

    概述本文介绍RabbitMQ中交换机类型和属性,主要内容如下: 交换机的作用交换机的类型:Direct exchange(直连交换机).Fanout exchange(扇型交换机).Topic exc ...

  6. RabbitMQ系列之RabbitMQ集群

    本文的集群是RabbitMq 自身的Cluster搭建 . #hosts准备: vim /etc/hosts #172.16.3.79 s1 #172.16.3.105 s2 172.16.2.10 ...

  7. RabbitMQ系列(四)RabbitMQ事务和Confirm发送方消息确认——深入解读

    RabbitMQ事务和Confirm发送方消息确认--深入解读 RabbitMQ系列文章 RabbitMQ在Ubuntu上的环境搭建 深入了解RabbitMQ工作原理及简单使用 RabbitMQ交换器 ...

  8. RabbitMQ系列(二)深入了解RabbitMQ工作原理及简单使用

    深入了解RabbitMQ工作原理及简单使用 RabbitMQ系列文章 RabbitMQ在Ubuntu上的环境搭建 深入了解RabbitMQ工作原理及简单使用 RabbitMQ交换器Exchange介绍 ...

  9. Redis总结(五)缓存雪崩和缓存穿透等问题 Web API系列&lpar;三&rpar;统一异常处理 C&num;总结(一)AutoResetEvent的使用介绍(用AutoResetEvent实现同步) C&num;总结(二)事件Event 介绍总结 C&num;总结(三)DataGridView增加全选列 Web API系列&lpar;二&rpar;接口安全和参数校验 RabbitMQ学习系列(六)&colon; RabbitMQ 高可用集群

    Redis总结(五)缓存雪崩和缓存穿透等问题   前面讲过一些redis 缓存的使用和数据持久化.感兴趣的朋友可以看看之前的文章,http://www.cnblogs.com/zhangweizhon ...

随机推荐

  1. ACM模板(持续补完)

    1.KMP #include<cstring> #include<algorithm> #include<cstdio> using namespace std; ...

  2. iOS 数据存储 - 归档和解归档

    这里的归档主要是用于自定义类的归档和解档.我们这里使用NSKeyedArchiver和NSKeyedUnarchiver来归档和解档. 注意:自己定义的类需要实现<NSCoding>,如: ...

  3. Android:简单实现ViewPager&plus;TabHost&plus;TabWidget实现导航栏导航和滑动切换

    viewPager是v4包里的一个组件,可以实现滑动显示多个界面. android也为viewPager提供了一个adapter,此adapter最少要重写4个方法: public int getCo ...

  4. 【Android】碎片Fragment

    1.碎片可以让界面在平板上更好地展示. 2.碎片是一种可以嵌入到活动中的UI片段,它能让程序更加合理和充分地利用一个大屏幕的空间.有自己的生命周期,能包含布局. 3.新建碎片类继承Fragment,可 ...

  5. SQL SERVER 清空日志

    DUMP TRANSACTION [TBNAME] WITH NO_LOGBACKUP LOG [TBNAME] WITH NO_LOGDBCC SHRINKDATABASE([TBNAME]) 1. ...

  6. Java Base64 加解密

    public class base64EncryAndDecry { public static final String CODES = "ABCDEFGHIJKLMNOPQRSTUVWX ...

  7. spring命名空间

    作为一名入门级菜鸟的我,在刚开始学spring框架的时候,对于命名空间不是很理解,只是在跟着老师敲代码.随着后来学习的深入,慢慢了解到命名空间的意义.如果你没有引入相应的命名空间,就不能引用相应的标签 ...

  8. 【用例管理】用testng的groups管理用例

    一.需求: 测试时经常有两种场景,第一种是冒烟测试的小部分用例:一类是全部用例. 二.针对第一种运行部分的用例,可以用groups来管理 package com.testcases; import o ...

  9. Android ScrollView 嵌套ListView的替代方案

    概要:本例仅提供替代思路. 原需求:实现下图这个布局 要求:头部菜单固定,实现Viewpager.中间的按钮菜单,底部的listview一起能够上下滚动. 做法: 把Viewpager.中间的按钮菜单 ...

  10. 机器学习三剑客之Pandas

      pandas Pandas是基于Numpy开发出的,专门用于数据分析的开源Python库 Pandas的两大核心数据结构 Series(一维数据)   Series   创建Series的方法   ...