[RabbitMQ] 7种工作模式详细介绍

🌸个人主页:https://blog.csdn.net/2301_80050796?spm=1000.2115.3001.5343
🏵️热门专栏:
🧊 Java基本语法(97平均质量分)https://blog.csdn.net/2301_80050796/category_12615970.html?spm=1001.2014.3001.5482
🍕 Collection与数据结构 (92平均质量分)https://blog.csdn.net/2301_80050796/category_12621348.html?spm=1001.2014.3001.5482
🧀线程与网络(96平均质量分) https://blog.csdn.net/2301_80050796/category_12643370.html?spm=1001.2014.3001.5482
🍭MySql数据库(93平均质量分)https://blog.csdn.net/2301_80050796/category_12629890.html?spm=1001.2014.3001.5482
🍬算法(97平均质量分)https://blog.csdn.net/2301_80050796/category_12676091.html?spm=1001.2014.3001.5482
🍃 Spring(97平均质量分)https://blog.csdn.net/2301_80050796/category_12724152.html?spm=1001.2014.3001.5482
🎃Redis(97平均质量分)https://blog.csdn.net/2301_80050796/category_12777129.html?spm=1001.2014.3001.5482
🐰RabbitMQ(97平均质量分) https://blog.csdn.net/2301_80050796/category_12792900.html?spm=1001.2014.3001.5482
感谢点赞与关注~~~
在这里插入图片描述

目录

  • 1. 7种工作模式介绍
    • 1.1 简单模式(simple)
    • 1.2 工作队列(Work Queue)
    • 1.3 对交换机和路由键的解释
    • 1.4 广播模式/发布订阅模式(Publish/Subscribe)
    • 1.5 路由模式(Routing)
    • 1.6 通配符模式(Topics)
    • 1.7 RPC通信(RPC)
    • 1.8 发布确认(Publish Confirms)
  • 2. 工作模式的代码实现
    • 2.1 简单模式
    • 2.2 工作队列
    • 2.3 广播模式/发布订阅模式(Publish/Subscribe)
    • 2.4 路由模式(Routing)
    • 2.5 通配符模式
    • 2.6 RPC通信(RPC)
    • 2.7 发布确认(Publisher Confirms/消息可靠性保证)
      • 2.7.1 概述
      • 2.7.2 单独确认
      • 2.7.3 批量确认
      • 2.7.4 异步确认

1. 7种工作模式介绍

在这里插入图片描述

1.1 简单模式(simple)

在这里插入图片描述
P: producer生产者,也就是要发送消息的程序.
C: consumer消费者,消息的接收者.
Queue: 消息队列,其中可以缓存信息,生产者可以向其中投递信息,消费者从其中获取信息.
特点: 一个生产者,一个消费者,一个消息只能被消费一次,也称为点对点模式.

1.2 工作队列(Work Queue)

在这里插入图片描述
一个生产者P,多个消费者C1,C2,在队列中有多个消息的情况下,work Queue会将消息分派给不同的消费者,每个消费者都会接收到不同的消息.
特点: 消息不会重复,分配给不同的消费者.

1.3 对交换机和路由键的解释

在这里插入图片描述

  • X: 代表的是交换机(exchange),作用就是生产者在将消息发送到交换机之后,交换机会按照生产者指定的RoutingKey,也就是路由规则把消息路由到一个或者多个队列中.
  • RabbitMQ中的交换机有四种类型: fanout,direct,topic,headers,不同的类型有不同的路由策略.
    1. fanout: 广播模式,将消息发送给所有绑定交换机的队列.(Publish/Subscribe模式)
    2. direct: 定向模式,将消息交给符合指定的routingKey的队列.(Routing模式)
    3. topic: 通配符模式,将消息交给符合指定的Routing pattern的队列.(Topics模式).
    4. header: headers类型的交换器性能会很差,而且也不实用,基本上不会看到它的存在.
      交换机只是负责转发消息,不具有存储消息的功能,因此,如果没有任何队列和交换机绑定,或者是消息没有任何符合路由规则的队列,那么信息就会丢失.
  • RoutingKey: 路由键,生产者在消息转发给交换机的时候,指定的一个字符串,用来让交换机知道该如何处理这条消息.
  • BindingKey: 绑定键,BindingKey是把交换机和消息队列绑定的字符串,这样交换机就会知道如何根据路由键将对应的消息转发到指定的队列了.
    在这里插入图片描述
    下面我们就举一个具体的例子来对RoutingKey和BindingKey具体说明.

在发送消息的时候,生产者设置了RoutingKey为orange,消息就会路由到Q1.
在这里插入图片描述

BindingKey其实也属于RoutingKey的一种,我们为了把这两个键混淆,我们可以这样理解:
在交换机和队列绑定的时候,需要的路由键是BindingKey.在发送消息的时候,需要的路由键是RoutingKey.

1.4 广播模式/发布订阅模式(Publish/Subscribe)

在这里插入图片描述
一个生产者P,多个消费者C1,C2,X代表交换机,交换机会将消息复制多份,发送给所有和交换机绑定的队列,每个消费者接收相同的消息.生产者发送⼀条消息，经过交换机转发到多个不同的队列，多个不同的队列就有多个不同的消费者.
适用场景: 消息需要被多个消费者同时接收的场景.比如: 实时通知或者广播消息.

比如国家地震局发送地震预警,地震发生的时候,需要把预警消息发送给可能有震感地区的所有电子设备.

1.5 路由模式(Routing)

在这里插入图片描述
路由模式是发布订阅模式的变种,在发布订阅模式的基础上,增加了路由key.
相比广播模式,路由模式是Exchange根据RoutingKey的规则,将数据筛选后发给对应的消费者队列.
适合场景: 需要根据特定的规则分发消息的场景.
比如我们在Spring中学习的日志,日志等级分为error,warning,info,debug,就可以通过这种模式,把不同的日志发送到不同的队列.

1.6 通配符模式(Topics)

在这里插入图片描述
路由模式的升级版,在routingKey的基础上,增加了通配符的功能,使之更加灵活.其中,一个.是一个节,使用*代表的是一个节,使用#代表的是多个节.Topics和Routing的基本原理相同.不同之处是：routingKey的匹配方式不同，Routing模式是相等匹配，topics模式是通配符匹配.
适用场景: 需要灵活匹配和过滤消息的场景.

1.7 RPC通信(RPC)

在RPC模式中没有生产者和消费者,大概就是通过两个队列实现了一个消息回调的过程.有点类似与我们在网络中学习的"请求和响应",这个功能是MQ的额外功能.
在这里插入图片描述

  1. 客户端发送消息到一个指定的队列,并在消息属性中设置replyTo字段,这个字段指定了回调队列,用于接收服务端返回的响应.
  2. 服务端接收到请求后,处理请求并发送响应消息到replyTo指定的回调队列.
  3. ⼀旦收到响应，客户端会检查消息的correlationId属性，以确保它是所期望的响应.因为队列中不仅仅有一条消息,保证发送的请求和收到的响应通过correlationId对应的上.

1.8 发布确认(Publish Confirms)

Publish Confirms模式是RabbitMQ提供的一种确保消息可靠发送到RabbitMQ服务器的机制.在这种模式之下,生产者可以等待RabbitMQ服务器确认,可以确保消息已经被服务器接收并处理.

  1. 生产者将Channel设置为Confirm模式,发布的每一条消息都会获得一个唯一的ID,生产者可以将这些序号与消息关联起来,以便跟踪消息的状态.
  2. 当消息被RabbitMQ服务器接收并处理之后,服务器会一步地向生产者发送一个(ACK)给生产者(其中包含消息的唯一ID),表名消息已经送达.
    通过该模式,生产者可以确保消息被RabbitMQ服务器成功接收,从而避免消息丢失的问题.
    适用场景: 对数据安全性较高的场景,比如金融交易,订单处理.
    在这里插入图片描述

2. 工作模式的代码实现

前面我们对这几种工作模式有了简单的了解,接下来我们学习他们的写法.

2.1 简单模式

就是快速上手中的程序,此处忽略.

2.2 工作队列

就是简单模式的增强版,和简单模式下最大的区别就是,工作队列模式支持多个消费者接收消息,消费者之间是竞争关系,每个消息只能被⼀个消费者接收.
在这里插入图片描述

  • 引入依赖
<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.7.3</version>
</dependency>
  • 编写生产者
    工作队列模式的代码和简单模式的代码没有多大的出入.我们只是发送了10次消息.我们把发送消息的地方改为10次发送.
    有一些配置相关的东西是固定的,所以我们可以把他们单独提出来一个Constant类.
public class Constant {public static String HOST = "39.105.137.64";public static int PORT = 5672;public static String USER_NAME = "jiangruijia";public static String PASSWORD = "qwe123524";public static String QUEUE_NAME = "work";public static String VIRTUAL_HOST = "/";
}
public class Producer {public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);connectionFactory.setUsername(Constant.USER_NAME);connectionFactory.setPassword(Constant.PASSWORD);connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();channel.queueDeclare(Constant.QUEUE_NAME,true,false,false,null);for (int i = 0; i < 10 ; i++){//发送10次消息String message = "hello work~" + i;channel.basicPublish("",Constant.QUEUE_NAME,null,message.getBytes());}channel.close();connection.close();}
}
  • 消费者代码
public class Consumer2 {//两个消费者是竞争关系public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);connectionFactory.setUsername(Constant.USER_NAME);connectionFactory.setPassword(Constant.PASSWORD);connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();channel.queueDeclare(Constant.QUEUE_NAME,true,false,false,null);DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String message = new String(body);System.out.println("接收到消息" + message);}};channel.basicConsume(Constant.QUEUE_NAME,true,consumer);//先不要关闭资源,因为需要先开启消费者,等待生产者发送消息}
}

另一个消费者和这个消费者相同,直接复制粘贴一份.

  • 运行程序,观察结果
    先启动两个消费者(而且两个消费者不可以关闭资源),再运行生产者,如果先启动生产者,在启动第一个消费者的时候,消息会被瞬间消费完.
    在这里插入图片描述
    在这里插入图片描述

2.3 广播模式/发布订阅模式(Publish/Subscribe)

在这里插入图片描述

  • 编写生产者
public class Producer {public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);connectionFactory.setUsername(Constant.USER_NAME);connectionFactory.setPassword(Constant.PASSWORD);connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();//声明交换机.channel.exchangeDeclare(Constant.FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT,true,false,false,null);//声明队列channel.queueDeclare(Constant.FANOUT_QUEUE1,true,false,false,null);channel.queueDeclare(Constant.FANOUT_QUEUE2,true,false,false,null);//绑定队列与交换机channel.queueBind(Constant.FANOUT_QUEUE1,Constant.FANOUT_EXCHANGE,"");channel.queueBind(Constant.FANOUT_QUEUE2,Constant.FANOUT_EXCHANGE,"");String message = "发送广播消息";channel.basicPublish(Constant.FANOUT_EXCHANGE,"",null,message.getBytes());channel.close();connection.close();}
}

参数解释:
- exchangeDeclare: 第一个参数是交换机名称,第二个参数是交换机的路由规则(这里指定为FANOUT广播类型),第三个参数是是否持久化,如果设置持久化,那么在重启服务之后,交换机不会被释放,第四个参数是是否自动删除,当没有队列与其绑定的时候,它就会被删除.第五个参数是是否是内部使用的,一般情况下为false,第六个参数是指定相关参数.
- queueDeclare: 声明队列,我们在之前解释过,这里不再赘述.
- queueBind: 绑定队列与交换机,第一个参数是队列名称,第二个参数是交换机名称,第三个参数是交换机和队列之间的路由规则,在这里我们是广播模式,所以我们没有指定路由规则,指定为默认的"".
- basicPublish: 第一个参数是发送消息的交换机,第二个参数是发送消息时的路由关键字,这里为广播模式,所以我们指定为"",第三个参数是一些相关配置,第四个参数是发送的消息.

  • 编写消费者代码
    虽然在消费者中可以不用声明队列,但是为有时候生产者和消费者不会在一个主机上,我们还是加上队列的声明比较符合逻辑,队列存在的时候,不会重复创建队列.
public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);connectionFactory.setUsername(Constant.USER_NAME);connectionFactory.setPassword(Constant.PASSWORD);connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();//队列声明可以省略,如果队列已经存在,则不会创建队列channel.queueDeclare(Constant.FANOUT_QUEUE1,true,false,false,null);DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String message = new String(body);System.out.println("获取到广播消息:"+message);}};channel.basicConsume(Constant.FANOUT_QUEUE1,true,consumer);}
}

第二个消费者直接复制一份,改掉消费者于队列的绑定和队列的声明即可.

  • 运行程序
    首先启动两个消费者,再运行生产者.
    我们看到生产者生产出两条消息之后,迅速被消费者消费掉.
    在这里插入图片描述
    两个消费者接收到了消息:
    在这里插入图片描述
    在这里插入图片描述

2.4 路由模式(Routing)

相比于发布订阅模式,交换机和队列不可以是任意绑定了==,而是需要指定一个BindingKey(RoutingKey的一种)==.生产者在向交换机发送消息的时候,也需要指定RoutingKey.这时,Exchange也不再把消息交给每⼀个绑定的key,而是根据消息RoutingKey进行判断,只有队列绑定时的BindingKey和发送消息的RoutingKey完全⼀致,才会接收到消息.
在这里插入图片描述

  • 编写生产者
    与发布订阅模式的不同是,交换机类型不同,而且绑定队列的bindingKey不同.
public class Producer {public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);connectionFactory.setUsername(Constant.USER_NAME);connectionFactory.setPassword(Constant.PASSWORD);connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();channel.exchangeDeclare(Constant.ROUTING_EXCHANGE, BuiltinExchangeType.DIRECT,true,false,false,null);channel.queueDeclare(Constant.ROUTING_QUEUE1,true,false,false,null);channel.queueDeclare(Constant.ROUTING_QUEUE2,true,false,false,null);channel.queueBind(Constant.ROUTING_QUEUE1,Constant.ROUTING_EXCHANGE,"a");channel.queueBind(Constant.ROUTING_QUEUE2,Constant.ROUTING_EXCHANGE,"a");channel.queueBind(Constant.ROUTING_QUEUE2,Constant.ROUTING_EXCHANGE,"b");channel.queueBind(Constant.ROUTING_QUEUE2,Constant.ROUTING_EXCHANGE,"c");String message1 = "routingKey_a";String message2 = "routingKey_b";String message3 = "routingKey_c";channel.basicPublish(Constant.ROUTING_EXCHANGE,"a",null,message1.getBytes());channel.basicPublish(Constant.ROUTING_EXCHANGE,"b",null,message2.getBytes());channel.basicPublish(Constant.ROUTING_EXCHANGE,"c",null,message3.getBytes());connection.close();channel.close();}
}

和上面广播模式不同的是,在绑定队列与交换机的时候,需要指定bindingKey.channel.queueBind(Constant.ROUTING_QUEUE1,Constant.ROUTING_EXCHANGE,"a");比如这一行指定了bindingKey为a.
在发送消息的时候,需要指定消息的RoutingKey,比如:channel.basicPublish(Constant.ROUTING_EXCHANGE,"a",null,message1.getBytes());.

  • 编写消费者
    Routing模式的消费者代码和Publish/Subscribe代码⼀样,同样复制出来两份.修改消费的队列名称就可以.
public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);connectionFactory.setUsername(Constant.USER_NAME);connectionFactory.setPassword(Constant.PASSWORD);connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();channel.queueDeclare(Constant.ROUTING_QUEUE1,true,false,false,null);DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String message = new String(body);System.out.println("接收到消息:"+message);}};channel.basicConsume(Constant.ROUTING_QUEUE1,true,consumer);}
}
  • 运行代码
    在这里插入图片描述
    队列中的消息被消费.
    队列2中的消息:
    在这里插入图片描述
    队列1中的消息:
    在这里插入图片描述

2.5 通配符模式

Topic和Routing模式的区别就是:
1. Topic模式使用的交换机类型为topic(Routing模式使⽤的交换机类型为direct).
2. Topic类型的交换机在匹配规则上进行了扩展,bindingKey支持通配符匹配.
在这里插入图片描述
在Topic类型的交换机在匹配的规则上,有一些要求:
1. RoutingKey是一系列由.分割的单词,比如"a.b.c".
2. BindingKey和RoutingKey一样,也是点.分割的字符串.
3. BindingKey中可以存在两种特殊字符串,用于模糊匹配.其中*表示一个单词,#表示多个单词.
比如:
• BindingKey为"d.a.b"会同时路由到Q1和Q2.
• BindingKey为"d.a.f"会路由到Q1.
• BindingKey为"c.e.f"会路由到Q2.
• BindingKey为"d.b.f"会被丢弃,或者返回给⽣产者(需要设置mandatory参数).
接下来我们就来实现Topic模式:

  • 编写生产者
    和路由模式最大的区别就是: 交换机类型不同,绑定队列的RoutingKey不同.
public class Producer {public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);connectionFactory.setUsername(Constant.USER_NAME);connectionFactory.setPassword(Constant.PASSWORD);connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();channel.queueDeclare(Constant.TOPIC_QUEUE1,true,false,false,null);channel.queueDeclare(Constant.TOPIC_QUEUE2,true,false,false,null);channel.exchangeDeclare(Constant.TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC,true,false,false,null);//交换机类型不同channel.queueBind(Constant.TOPIC_QUEUE1,Constant.TOPIC_EXCHANGE,"*.a.*");channel.queueBind(Constant.TOPIC_QUEUE2,Constant.TOPIC_EXCHANGE,"*.*.b");channel.queueBind(Constant.TOPIC_QUEUE2,Constant.TOPIC_EXCHANGE,"c.#");//bindingKey不同String message1 = "hello_b.a.c";String message2 = "hello_a.c.b";String message3 = "hello_c.a.b";channel.basicPublish(Constant.TOPIC_EXCHANGE,"b.a.c",null,message1.getBytes());channel.basicPublish(Constant.TOPIC_EXCHANGE,"c.a.b",null,message3.getBytes());channel.basicPublish(Constant.TOPIC_EXCHANGE,"a.c.b",null,message2.getBytes());channel.basicPublish(Constant.TOPIC_EXCHANGE,"c.a.b",null,message3.getBytes());channel.close();connection.close();}
}
  • 编写消费者
public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);connectionFactory.setUsername(Constant.USER_NAME);connectionFactory.setPassword(Constant.PASSWORD);connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();channel.queueDeclare(Constant.TOPIC_QUEUE1,true,false,false,null);DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String s = new String(body);System.out.println("收到消息: "+s);}};channel.basicConsume(Constant.TOPIC_QUEUE1,true,consumer);}
}

第二个消费者直接修改一下队列名称就可以.

  • 运行结果
    队列产生消息,并消费
    在这里插入图片描述
    消费者根据RoutingKey收到了关键字:
    消费者2:
    在这里插入图片描述
    消费者1:
    在这里插入图片描述

2.6 RPC通信(RPC)

RPC通信,是远程过程调用,它是一种发送请求,得到响应的模式,有点类似与我们之前学习网络时候的http协议.
RabbitMQ实现RPC通信的过程,大概是通过两个队列实现⼀个可回调的过程.
在这里插入图片描述
在这个模式中,没有明确的生产者和消费者,在发送请求的时候,客户端是生产者,服务端是消费者,在接收响应的时候,服务端是生产者,客户端是消费者.
大概的流程如下:

  1. 客户端发送消息到⼀个指定的队列,并在消息属性中设置replyTo字段,这个字段指定了⼀个回调队列,在消息中还会设置correlate_id,这个字段用于保证接收到的响应是与请求对应的响应,服务端处理后,会把响应结果发送到这个队列.
  2. 服务端接收到消息之后,处理请求并发送响应消息到replyTo指定的回调队列.
  3. 客户端在回调队列上等待响应消息.⼀旦收到响应，客户端会检查消息的correlationId属性，以确保它是所期望的响应.

接下来我们来实现PCR模式:

  • 客户端代码
    客户端代码的主要流程如下:
    首先声明两个队列,包含回调队列replyQueueName,和发送请求的队列.并声明本次请求的唯一标志corrld.之后将replyQueueName和corrld配置到要发送的消息队列中.之后使用阻塞队列来阻塞当前的进程,监听回调队列中的消息,把请求放到阻塞队列中.阻塞队列中有消息之后,主线程被唤醒,打印返回的内容.
public class Client {public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);connectionFactory.setUsername(Constant.USER_NAME);connectionFactory.setPassword(Constant.PASSWORD);connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();channel.queueDeclare(Constant.SRC_REQUEST_QUEUE,true,false,false,null);channel.queueDeclare(Constant.SRC_RESPONSE_QUEUE,true,false,false,null);//发送请求String corrId = UUID.randomUUID().toString();AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().correlationId(corrId).replyTo(Constant.SRC_RESPONSE_QUEUE)//指定相关属性.build();String request = "发送请求";//如果没有交换机的时候,RoutingKey就是队列的名称channel.basicPublish("",Constant.SRC_REQUEST_QUEUE,properties,request.getBytes());//接收请求BlockingQueue<String> queue = new ArrayBlockingQueue<>(1);DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {if(corrId.equals(properties.getCorrelationId())){//判断corrId是否相等queue.offer(new String(body));}}};channel.basicConsume(Constant.SRC_RESPONSE_QUEUE,true,consumer);String ret = queue.take();System.out.println("收到请求:" + ret);channel.close();connection.close();}
}
  • 服务端代码
    服务端需要做的就是:接收消息,根据消息内容进行响应处理,把应答结果返回到回调队列中.
public class Server {public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);connectionFactory.setUsername(Constant.USER_NAME);connectionFactory.setPassword(Constant.PASSWORD);connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();channel.queueDeclare(Constant.SRC_REQUEST_QUEUE,true,false,false,null);DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("成功接收到请求:"+new String(body));AMQP.BasicProperties basicProperties = new AMQP.BasicProperties().builder().correlationId(properties.getCorrelationId()).build();String message = "返回响应";channel.basicPublish("",Constant.SRC_RESPONSE_QUEUE,basicProperties,message.getBytes());channel.basicAck(envelope.getDeliveryTag(),false);}};channel.basicConsume(Constant.SRC_REQUEST_QUEUE,false,consumer);//设置收到消息之后不自动应答,在发送响应之后手动应答}
}

RabbitMQ消息确定的机制:
在RabbitMQ中,basicConsumer方法的autoAck参数用于指定消费者是否应该自动向消息队列确认消息.
自动确认（autoAck=true）:消息队列在将消息发送给消费者后,会立即从内存中删除该消息.这意味着,如果消费者处理消息失败，消息将丢失，因为消息队列认为消息已经被成功消费
手动确认（autoAck=false）:消息队列在将消息发送给消费者后，需要消费者显式地调用basicAck方法来确认消息.手动确认提供了更高的可靠性,确保消息不会被意外丢失,适用于消息处理重要且需要确保每个消息都被正确处理的场景.

  • 运行程序
    在这里插入图片描述
    在这里插入图片描述

2.7 发布确认(Publisher Confirms/消息可靠性保证)

2.7.1 概述

作为消息中间件,都会面临丢失的问题.
消息丢失大概分为三种类型:

  1. 生产者问题.因为应用程序故障,网络都用等原因,生产者没有成功向Broker发送消息.
  2. 消息中间件自身问题,生产者成功发送给了Broker,但是Broker没有吧消息保存好,导致了信息丢失.
  3. 消费者问题,Broker发送消息到消费者,消费者在消费消息的时候,没有处理好,导致Broker将消费失败的消息从队列中删除了.
    在这里插入图片描述
    上面的这几个问题都有对应的解决方式,问题2可以通过持久化的机制来解决,问题3可以通过消息应答机制来解决,问题1,可以采用发布确认的机制来实现.
    发布确认属于RabbitMQ的七大工作模式之一.
    生产者会将Channel设置为Confirm模式,一旦信道进入Confirm模式,所有在该信道上面的消息都会被指派为一个唯一的Id.一旦消息被投递到所有匹配的队列之后,RabbitMQ就会发送一个确认给生产者(包含消息的唯一ID),这就使得生产者知道消息已经确认到达目的队列了.Broker回传给生产者的确认消息中包含了确认消息的序号,此外Broker也可以设置channel.basicAck方法中的multiple参数,true表示在deliveryTag序号之前的消息都已经收到了,如果为false,那么则有消息没有收到,消息确认出了一些问题.
    在这里插入图片描述
    使用发送机制的时候,必须要将信道设置为Confirm模式.发布确认有3种策略,单独确认,批量确认,异步确认,接下来我们就来学习着三种策略.
    下面是开启信道确认模式的方法.
Channel channel = connection.createChannel();
channel.confirmSelect();//开启信道确认模式

2.7.2 单独确认

首先我们需要建立连接,建立连接需要放入try语句中,所以我们可以把建立连接单独提出一个静态方法.

private static Connection createConnection() throws IOException, TimeoutException {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);connectionFactory.setUsername(Constant.USER_NAME);connectionFactory.setPassword(Constant.PASSWORD);connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);return connectionFactory.newConnection();
}

之后在主方法中调用三种策略,

public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {publishingMessagesIndividually();publishMessagesInBatch();handlePublishConfirmsAsynchronously();
}

之后我们来编写单条确认模式:

    /*** 单条确认*/
private static void publishingMessagesIndividually() throws IOException, TimeoutException, InterruptedException {try (Connection connection = createConnection()){//首先建立连接Channel channel = connection.createChannel();channel.confirmSelect();//开启信道确认模式channel.queueDeclare(Constant.CONFIRM_QUEUE1,true,false,false,null);//创建队列String message = "发送信息";for (int i = 0; i < 200; i++){channel.basicPublish("",Constant.CONFIRM_QUEUE1,null,(message+i).getBytes());channel.waitForConfirms(5000);//等待消息确认,如果超过规定的等待时间还没有确认,则抛出异常}}
}

在编写代码之前,我们首先要使用try语句与服务器建立连接.
之后在创建Channel之后,需要开启信道的确认模式.channel.confirmSelect().
这里我们在发送消息之后,需要做的最重要的一件事就是等待消息的确认channel.waitForConfirms(5000),在这个方法中可以指定阻塞时间,如果在指定的时间内消息被确认,这个方法就会立即返回,如果在指定时间之内没有确认消息,则会抛出异常.

2.7.3 批量确认

/*** 批量确认*/
private static void publishMessagesInBatch() throws IOException, TimeoutException, InterruptedException {try (Connection connection = createConnection()){Channel channel = connection.createChannel();channel.confirmSelect();//设置为确认模式channel.queueDeclare(Constant.CONFIRM_QUEUE2,true,false,false,null);String message = "批量确认发送消息";int batchSize = 100;//每次批量发送的消息条数int outstandingMessageCount = 0;//记录已经发送的条数for (int i = 0 ; i < 200 ; i++){channel.basicPublish("",Constant.CONFIRM_QUEUE2,null,(message+i).getBytes());outstandingMessageCount++;//每发送一条消息,参数就进行++if(outstandingMessageCount == batchSize){//达到了一次性批量发送的指定数量,等待确认,确认完成之后,将参数清零channel.waitForConfirms(5000);outstandingMessageCount = 0;}}//如果发送的消息不是100的倍数,就还有消息没有确认if (outstandingMessageCount > 0){channel.waitForConfirms(5000);}}
}

这里需要注意的几点就是,在发送的消息达到一次性最大的批量数量的时候,就要确认,如果确认成功之后,需要把记录的发送数量清零.
之后,就是在出循环之后,如果发送的消息条数不是batchSize的整数倍的时候,这时候不满足循环之内的if条件,还是有一些消息没有确认完成,就需要在循环之外再次进行确认.

2.7.4 异步确认

异步确认就是,生产者在发送消息的同时,还可以确认消息是否收到.
Channel接口中为我们提供了一个方法,addConfirmListener.这个方法可以添加ConfirmListener回调接口.
ConfirmListener中包含两个方法: handleAck(long deliveryTag, boolean multiple) 和handleNack(long deliveryTag, boolean multiple),分别对应处理的是MQ发送给生产者的ack和nack.其中ack代表的是消息确认成功,即消息都收到的情况下,nack指的是消息在确认的时候出现了一些问题.deliveryTag表示的是发送消息的序号,multiple表示是否批量确认.
这里我们还需要一个有序集合来存储为确认的消息.

/*** 异步确认*/
private static void handlePublishConfirmsAsynchronously() throws IOException, TimeoutException, InterruptedException {try (Connection connection = createConnection()){Channel channel = connection.createChannel();channel.confirmSelect();channel.queueDeclare(Constant.CONFIRM_QUEUE3,true,false,false,null);SortedSet<Long> set = Collections.synchronizedSortedSet(new TreeSet<>());//设置一个有序集合,用来存储未确认消息的序号channel.addConfirmListener(new ConfirmListener() {//为信道添加监听器,监听消息的确认情况@Overridepublic void handleAck(long deliveryTag, boolean multiple) throws IOException {//表示消息被成功确认if (multiple){//判断消息的处理是否批量set.headSet(deliveryTag+1).clear();//将小于deliveryTag的消息全部清除,证明这批消息已经被ack了}else {set.remove(set.last());//如果不是批量,清除最后一个即可}}@Overridepublic void handleNack(long deliveryTag, boolean multiple) throws IOException {if (multiple){//判断消息的处理是否批量set.headSet(deliveryTag+1).clear();//将小于deliveryTag的消息全部清除,证明这批消息已经被ack了}else {set.remove(set.last());//如果不是批量,清除最后一个即可}//这里在消息确认不成功的时候,需要重传,这里省略}});String message = "异步确认发送消息";for (int i = 0 ;i < 200 ;i++){long nextPublishSeqNo = channel.getNextPublishSeqNo();//获取到消息发送的序号channel.basicPublish("",Constant.CONFIRM_QUEUE3,null,(message+i).getBytes());set.add(nextPublishSeqNo);//把这些消息都添加到集合中}while (!set.isEmpty()){//等待集合中的消息都被确认完成Thread.sleep(1000);}}
}

这里我们在消息确认成功之后,即handleAck方法被调用的时候,需要把这些消息都从集合中清除掉.一种是批量的情况,直接清除掉deliveryTag之前所有的消息,另一种是没有批量的情况,直接清除掉最后一个元素即可.当然在没有确认成功的情况下,我们需要根据具体的业务逻辑进行消息的重发.在给队列中发送消息的时候,我们需要从Channel中获取到下次发送消息开始的序号,之后我们把开始的序号放入set中,代表这些消息还没有被处理过.最后我们需要等待消息确认完成,只要存放未确认消息的set中不为空,就证明还有消息没有被确认,我们就进行阻塞等待.
上面三种方式中,假如发送的消息较多,这三种策略的执行时间:单个确认>批量确认>异步确认.

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.rhkb.cn/news/435376.html

如若内容造成侵权/违法违规/事实不符，请联系长河编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

软件测试|数据库常见面试题

软件测试|数据库常见面试题

在软件测试数据库的面试中&#xff0c;面试官通常会考察应聘者对数据库的理解、SQL语言的应用、数据库性能优化、以及数据库相关的技术栈和工具等方面的知识。以下是一些可能的面试问题及建议的回答思路&#xff1a; 1、什么是关系型数据库&#xff0c;主键&#xff0c;外键&am…
阅读更多...
【C++算法】6.双指针_有效三角形的个数

【C++算法】6.双指针_有效三角形的个数

文章目录 题目链接&#xff1a;题目描述&#xff1a;解法C 算法代码&#xff1a;图解&#xff1a; 题目链接&#xff1a; 611.有效三角形的个数 题目描述&#xff1a; 解法 数学知识&#xff1a; 给我们3个数&#xff0c;判断是否能够构成三角形。 平时&#xff1a;ab>c&am…
阅读更多...
12、echarts 没有显示折线图

12、echarts 没有显示折线图

一、问题描述 echarts 没有显示折线图&#xff0c;但是&#xff0c;有数据显示&#xff1a; 看图表展示&#xff0c;y轴数据全部没有显示&#xff0c;直接可以判定是数据结构出问题了。 检查 series.data[] 数据结构&#xff1a; dataList [{"dateStr":"202…
阅读更多...
SpringMVC4-SpringMVC获取请求参数

SpringMVC4-SpringMVC获取请求参数

目录 通过ServletAPI获取&#xff08;不常用&#xff09; 通过控制器方法的形参获取请求参数 RequestParam RequestHeader CookieValue 通过POJO获取请求参数 解决获取请求参数的乱码问题 test_param.html&#xff1a; <!DOCTYPE html> <html lang"en&qu…
阅读更多...
vue2中使用tailwindCss 详细教程

vue2中使用tailwindCss 详细教程

1、先看官方文档&#xff1a;https://www.tailwindcss.cn/ 2、先安装&#xff1a;npm install -D tailwindcss ---------------通过 npm 安装 tailwindcss&#xff0c;然后创建你自己的 create your tailwind.config.js 配置文件。 npm install -D tailwindcss 3、初始化文件…
阅读更多...
vue 同一个页面第二次跳转路由内容不更新

vue 同一个页面第二次跳转路由内容不更新

问题出现原因 在vue中相同路由之间跳转&#xff0c;默认在跳转路由时会采用缓存策略,并不会刷新当前路由组件。导致mounted&#xff08;初始化&#xff09;,beforeDestory&#xff08;销毁&#xff09;等生命周期钩子函数并不会触发&#xff0c;从而产生路由跳转了&#xff0c;…
阅读更多...
SpringMVC源码-SpringMVC源码请求执行流程及重点方法doDispatch讲解

SpringMVC源码-SpringMVC源码请求执行流程及重点方法doDispatch讲解

一、开始请求 在浏览器访问http://localhost:8080/spring_mymvc/userlist这个接口&#xff0c;是个get请求。 FrameworkServlet类的service方法会被请求到: 调用路径如下&#xff1a; service:945, FrameworkServlet (org.springframework.web.servlet) service:764, HttpSer…
阅读更多...
？？实验——完全使用Ansible部署多台服务器的服务

？？实验——完全使用Ansible部署多台服务器的服务

文章目录 需求“NTP时间同步服务”“Zabix”“ELK”gitlab、Jenkis集群架构自动上线代码两台Web服务器部署同一Web应用WeCenter&#xff0c;且两台服务器的用户上传的数据目录挂载到共享存储服务器中&#xff0c;总数据保存在一台数据库服务器中使用sersync简单实现两台共享存储…
阅读更多...
「系列投研｜01」建立自己的移动比特币银行——赛道概况

「系列投研｜01」建立自己的移动比特币银行——赛道概况

ZJUBCA 09/28/2024 01&#xff5c;BTCfi赛道概况 BTCFi&#xff1a;建立自己的移动比特币银行 ——从Lending到Staking的全面解读 作者&#xff1a; Freya & Knight& Ausdin from ZJUBCA Elaine & Youyu from Satoshi Lab 关键词 BTCFi&#xff0c;稳定币&#xff…
阅读更多...
4. 将pycharm本地项目同步到(Linux)服务器上——深度学习·科研实践·从0到1

4. 将pycharm本地项目同步到(Linux)服务器上——深度学习·科研实践·从0到1

目录 前序工作 1. 服务器项目名和本地一致 2. pycharm连接服务器 3. 本地项目对应到服务器项目 4. 简单测试一下同步效果 同步成功 前序工作 在同步到服务器之前&#xff0c;得确保已经完成以下几个前置步骤&#xff1a; 1. 租一个云服务器&#xff0c;可参考&#xff1a…
阅读更多...
使用transformers中的pipeline调用huggingface中模型过程中可能遇到的问题和修改建议

使用transformers中的pipeline调用huggingface中模型过程中可能遇到的问题和修改建议

使用transformers中的pipeline调用huggingface中模型过程 前言管线使用中的问题和解决huggingface的连接问题使用huggingface示例修改源继续使用pipeline No module named keras.engine 前言 HuggingFace有一个巨大的模型库&#xff0c;其中包括很多的比较成熟的经典模型&…
阅读更多...
利用ChatGPT实现的生成式人工智能自动化控制系统

利用ChatGPT实现的生成式人工智能自动化控制系统

一、引言 随着信息化与智能化时代的到来&#xff0c;人工智能&#xff08;AI&#xff09;技术迅猛发展&#xff0c;正在深刻地重塑各行业的运营模式。在这一背景下&#xff0c;生成式人工智能&#xff08;Generative AI&#xff09;以其卓越的创造力和广泛的应用潜力&#xff…
阅读更多...
Webpack 特性探讨：CDN、分包、Tree Shaking 与热更新

Webpack 特性探讨：CDN、分包、Tree Shaking 与热更新

文章目录 前言包准备CDN 集成代码分包Tree Shaking原理实现条件&#xff1a;解决 treeShaking 无效方案&#xff1a;示例代码&#xff1a; 热更新&#xff08;HMR&#xff09; 前言 Webpack 作为现代前端开发中的核心构建工具&#xff0c;提供了丰富的特性来帮助开发者优化和打…
阅读更多...
java基础（4）类和对象

java基础（4）类和对象

目录 1.前言 2.正文 2.1类的定义与使用 2.1.1类的定义 2.1.2类的实例化 2.1.3this引用 2.1.3.1 访问当前对象的成员变量 2.1.3.2调用当前对象的成员方法 2.1.3.3构造函数中的 this 2.1.3.4归纳this 2.2封装 2.2.1封装的定义 2.2.2访问修饰符 2.3static 2.3.1sta…
阅读更多...
初学Vue

初学Vue

文章目录 简介特点 初识Vue模板语法两大类插值语法指令语法 两种数据绑定方式单项绑定&#xff08;v-bind&#xff09;双向绑定&#xff08;v-model&#xff09; 数据代理事件处理基本使用事件修饰符 键盘事件计算属性 computed 简介 一套用于构建用户界面的渐进式JavaScript框…
阅读更多...
【面试题】软件测试实习（含答案）

【面试题】软件测试实习（含答案）

软件测试实习常见面试题&#xff0c;主要是功能测试相关的基础问题 目录 一、软件测试基础 1、介绍一下你最近的项目&#xff0c;以及工作职责 2、软件项目的测试流程? 3、黑盒测试与白盒测试的区别? 4、黑盒测试常见的设计方法?怎么理解等价类方法和边界值方法 1&…
阅读更多...
服务器几核几G几M是什么意思？如何选择？

服务器几核几G几M是什么意思？如何选择？

服务器几核几G几M是什么意思&#xff1f;我们建站、搭建网络平台都要用到云服务器&#xff0c;不管在腾讯云、阿里云还是别的云服务平台选购&#xff0c;都会接触到服务器配置。云服务器就是把物理服务器&#xff08;俗称“母鸡”&#xff09;&#xff0c;用虚拟机技术虚拟出多…
阅读更多...
FreeRTOS学习笔记一——FreeRTOS介绍

FreeRTOS学习笔记一——FreeRTOS介绍

RTOS学习笔记&#xff0c;主要参考正点原子教程 目录 FreeRTOS特点任务调度方式抢占式调度时间片调度 任务状态状态转换任务列表 FreeRTOS特点 实现多个任务功能划分延时函数实现任务调度高优先级抢占低优先级每个任务都有自己的栈空间 注意&#xff1a; 中断可以打断任意任务…
阅读更多...
设计模式之享元（Flyweight）模式

设计模式之享元（Flyweight）模式

前言 面向对象很好地解决了 “抽象” 的问题&#xff0c;但是不可避免的要付出一定的代价。对于通常情况来讲&#xff0c;面向对象的成本大都可以忽略不计。但是某些情况&#xff0c;面向对象所带来的成本必须谨慎处理 具体需要自己根据需求去评估 定义 “对象性能” 模式。运用…
阅读更多...
探索基于知识图谱和 ChatGPT 结合制造服务推荐前沿

探索基于知识图谱和 ChatGPT 结合制造服务推荐前沿

0.概述 论文地址&#xff1a;https://arxiv.org/abs/2404.06571 本研究探讨了制造系统集成商如何构建知识图谱来识别新的制造合作伙伴&#xff0c;并通过供应链多样化来降低风险。它提出了一种使用制造服务知识图谱&#xff08;MSKG&#xff09;提高 ChatGPT 响应准确性和完整…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023