【RabbitMQ】常用消息模型详解

文章目录

  • AMQP协议的回顾
  • RabbitMQ支持的消息模型
  • 第一种模型(直连)
    • 开发生产者
    • 开发消费者
    • 生产者、消费者开发优化
    • API参数细节
  • 第二种模型(work quene)
    • 开发生产者
    • 开发消费者
    • 消息自动确认机制
  • 第三种模型(fanout)
    • 开发生产者
    • 开发消费者
  • 第四种模型(Routing)
    • 开发生产者
    • 开发消费者
  • 第五种模型(Topic)
    • 开发生产者
    • 开发消费者

AMQP协议的回顾

在这里插入图片描述
在RabbitMQ中有生产者、消费者的概念，生产者先与我们的RabbitMQServer建立连接，建立完连接之后，它会把消息通过连接中通道的形式去传递我们的消息。每一个生产者会对应一个专门的虚拟主机。

在我们做项目的时候，RabbitMQ希望我们每一个项目具有单独的虚拟主机，这样我们多个应用在操作同一个RabbitMQServer的时候互不影响，所以这里的虚拟机有点像关系型数据库中的库概念。

我们在访问虚拟主机的时候是需要权限的，如果需要访问到某一个具体的虚拟主机，我们需要将虚拟主机与用户进行绑定。

比如RabbitMQ默认为我们提供的guest账户，他是可以访问所有的虚拟主机的，具有至高无上的权限。在我们实际的生产环境中我们一般是一个项目访问一个虚拟主机，或者说是一个业务访问一个虚拟主机，在访问的时候我们一般为一个虚拟主机绑定特定的用户。

当我们的生产者通过通道将消息放入到虚拟机之中，因为RabbitMQ存在许多的消息模型，所以这里不一定会把消息放入到交换机之中。也就是说当生产者将消息传递给交换机或者队列之后，他的任务就告一段落了。

这个时候我们的生产者和消费者是完全解耦的，我们不需要关心生产者到底有没有运行，我只关心消费者监听的队列里面有没有对应的消息即可。

消费者在消费消息的时候也需要去连接到我们RabbitMQServer以及虚拟主机，我们才能消费到对应主机中的消息队列里面的数据。

RabbitMQ支持的消息模型

在这里插入图片描述
在这里插入图片描述

最新的版本有第七种消息模型：消息确认模型

第一种模型(直连)

在这里插入图片描述

在上图的模型中，有以下概念：

  • P：生产者，也就是要发送消息的程序
  • C：消费者：消息的接受者，会一直等待消息到来。
  • queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。

首先我们先创建一个新用户/ems，然后将一个虚拟主机与其绑定，然后给他添加超级用户权限：
在这里插入图片描述

注意：
用户名必须以/开头

开发生产者

public class Provider {//生产消息@Testpublic void testSendMessage() throws IOException, TimeoutException {//创建连接mq的连接工厂对象ConnectionFactory connectionFactory = new ConnectionFactory();//设置连接rabbitmqserver主机connectionFactory.setHost("10.15.0.9");//设置端口号connectionFactory.setPort(5672);//设置连接那个虚拟主机connectionFactory.setVirtualHost("/ems");//设置访问虚拟主机的用户名和密码connectionFactory.setUsername("ems");connectionFactory.setPassword("123");//获取连接对象Connection connection = connectionFactory.newConnection();//获取连接中通道Channel channel = connection.createChannel();//通道绑定对应消息队列//参数1:  队列名称 如果队列不存在自动创建//参数2:  用来定义队列特性是否要持久化 true 持久化队列   false 不持久化//参数3:  exclusive 是否独占队列  true 独占队列   false  不独占//参数4:  autoDelete: 是否在消费完成后自动删除队列  true 自动删除  false 不自动删除//参数5:  额外附加参数channel.queueDeclare("hello",true,false,false,null);//发布消息//参数1: 交换机名称 参数2:队列名称  参数3:传递消息额外设置  参数4:消息的具体内容channel.basicPublish("","hello", null,"hello rabbitmq".getBytes());channel.close();connection.close();}
}

开发消费者

public class Customer {public static void main(String[] args) throws IOException, TimeoutException {//创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("10.15.0.9");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/ems");connectionFactory.setUsername("ems");connectionFactory.setPassword("123");//创建连接对象Connection connection = connectionFactory.newConnection();*///创建通道Channel channel = connection.createChannel();//通道绑定对象channel.queueDeclare("hello",true,false,false,null);//消费消息//参数1: 消费那个队列的消息 队列名称//参数2: 开始消息的自动确认机制//参数3: 消费时的回调接口channel.basicConsume("hello",true,new DefaultConsumer(channel){@Override //最后一个参数: 消息队列中取出的消息public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("===================================="+new String(body));}});}}

注意：
在使用Junit测试的时候，他是不支持多线程模型的。如果我们使用@Test去运行的话，他没法让我们的消费者去监听(运行完之后直接就杀死了该进程，不会处于监听状态)，所以这里我们要换成一个main函数。
生产者则不需要注意这一点，因为它生产完消息就完事了

在这里插入图片描述
我们发现生产者生产完消息之后，会关闭通道和链接，而在消费这里我们并没有这么做。这是因为可能会导致我们的回调函数还没来得及执行，我们的通道就已经关闭。

该模型的特点：
点对点的简单消费模型。
适用于登录、注册场景

生产者、消费者开发优化

我们发现我们在开发生产者、消费者的时候前面的连接部分代码重复冗余，所以我们可以使用一个工具类对其进行封装：

public class RabbitMQUtils {private static ConnectionFactory connectionFactory;private static Properties properties;static{//重量级资源  类加载执行之执行一次connectionFactory = new ConnectionFactory();connectionFactory.setHost("10.15.0.5");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");}//定义提供连接对象的方法public static Connection getConnection() {try {return connectionFactory.newConnection();} catch (Exception e) {e.printStackTrace();}return null;}//关闭通道和关闭连接工具方法public static void closeConnectionAndChanel(Channel channel, Connection conn) {try {if(channel!=null) channel.close();if(conn!=null)   conn.close();} catch (Exception e) {e.printStackTrace();}}}

我们这里使用静态代码块是因为connectionFactory是重量级资源，所以我们决定只在类加载执行时执行一次。

我们这里稍微复习一下java的静态代码块，我们会发现在一些项目源码中经常会见到他。
静态代码块语法格式：

static{}

静态代码块的特点：随着类的加载而执行，而且只执行一次
执行优先级高于非静态的初始化块，它会在类初始化的时候执行一次，执行完成便销毁，它仅能初始化类变量，即static修饰的数据成员。
那么正好我们再来提一下非静态代码块：
非静态代码块语法格式：

{}

执行的时候如果有静态初始化块，先执行静态初始化块再执行非静态初始化块，在每个对象生成时都会被执行一次，它可以初始化类的实例变量。非静态初始化块会在构造函数执行时，在构造函数主体代码执行之前被运行。
执行顺序:
静态代码块----->非静态代码块-------->构造函数

API参数细节

生产者和消费者均有一个方法queueDeclare，就是声明操作的队列：

channel.queueDeclare("hello",true,false,false,null);
  • 参数1: 队列名称
    • 如果队列不存在自动创建
  • 参数2: 用来定义队列特性是否要持久化
    • true 持久化队列
    • false 不持久化
    • 注意这里说的是队列的持久化
    • 也就是如果开启持久化的话，我们即使重启rabbitmq服务该队列也会存在，因为其内部会把队列从内存写到硬盘中去。当重启完成之后，其又会重新将硬盘中的队列读到内存中去
  • 参数3: exclusive 是否独占队列
    • true 独占队列
    • 也就是说队列只能被当前通道所绑定
    • false 不独占
  • 参数4: autoDelete: 是否在消费完成后自动删除队列
    • true 自动删除
    • false 不自动删除
    • 这里的自动删除队列是指消费者不再监听占用队列，队列才会消失
  • 参数5: 额外附加参数

注意：直连模型下，消费者和生产者的queueDeclare中的参数要保持一致，这样才能保证操作的是同一个队列

生产者：

channel.basicPublish("","hello", MessageProperties.PERSISTENT_TEXT_PLAIN,"hello rabbitmq".getBytes());
  • 参数1: 交换机名称 (我们这里没有使用交换机所以没有指定)
  • 参数2:队列名称
  • 参数3:传递消息额外设置
    • MessageProperties.PERSISTENT_TEXT_PLAIN
    • 我们可以通过此参数设置消息在队列中的持久化
  • 参数4:消息的具体内容
    • 这里是以字节的方式进行传输

消费者：

channel.basicConsume("hello",true,new DefaultConsumer(channel){@Override //最后一个参数: 消息队列中取出的消息public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("======"+new String(body));}
});
  • 参数1: 消费哪个队列的消息 队列名称
  • 参数2: 开始消息的自动确认机制
  • 参数3: 消费时的回调接口
    • 这里我们可以传入一个consumer对象，而这个consumer是一个接口，它有一个实现类DefaultConsumer

第二种模型(work quene)

Work queues，也被称为（Task queues），任务模型。当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。此时就可以使用work 模型：让多个消费者绑定到一个队列，共同消费队列中的消息。队列中的消息一旦消费，就会消失，因此任务是不会被重复执行的。

在这里插入图片描述

角色：

  • P：生产者：任务的发布者
  • C1：消费者-1，领取任务并且完成任务，假设完成速度较慢
  • C2：消费者-2：领取任务并完成任务，假设完成速度快

开发生产者

public class Provider {public static void main(String[] args) throws IOException {//获取连接对象Connection connection = RabbitMQUtils.getConnection();//获取通道对象Channel channel = connection.createChannel();//通过通道声明队列channel.queueDeclare("work", true, false, false, null);for (int i = 1; i <=20; i++) {//生产消息channel.basicPublish("", "work", null, (i + "hello work quene").getBytes());}//关闭资源RabbitMQUtils.closeConnectionAndChanel(channel, connection);}
}

我们这里使用了前面提到的连接工具类

我们运行我们的代码：
在这里插入图片描述
这里的Unacked代表未被确认的消息

开发消费者

如果我们对两个消费者不做任何处理：

消费者-1

public class Customer1 {public static void main(String[] args) throws IOException {//获取连接Connection connection = RabbitMQUtils.getConnection();final Channel channel = connection.createChannel();channel.queueDeclare("work",true,false,false,null);//参数1:队列名称  参数2:消息自动确认 true  消费者自动向rabbitmq确认消息消费  false 不会自动确认消息channel.basicConsume("work",false,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {                System.out.println("消费者-1: "+new String(body));                }});}
}

消费者-2

public class Customer2 {public static void main(String[] args) throws IOException {//获取连接Connection connection = RabbitMQUtils.getConnection();final Channel channel = connection.createChannel();channel.queueDeclare("work",true,false,false,null);//参数1:队列名称  参数2:消息自动确认 true  消费者自动向rabbitmq确认消息消费  false 不会自动确认消息channel.basicConsume("work",false,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {                System.out.println("消费者-2: "+new String(body));                }});}
}

在这种不做任何处理的情况下，消费者1、消费者2消费的消息都是一致的：
在这里插入图片描述
在这里插入图片描述

总结:默认情况下，RabbitMQ将按顺序将每个消息发送给下一个使用者。平均而言，每个消费者都会收到相同数量的消息。这种分发消息的方式称为循环。（也就是说平均分配）

而这样的话我们不难想到一个问题：加入我们的消费者1处理的比较慢，消费者2处理的比较快。这就导致消费者1的消息会在队列中造成滞留，消费者2可能已经处理完闲着了。这样的情况下平均分配显然也会影响效率，并且导致消息再队列中的积累。

我们可以模拟一下这个情况，我们在消费者1中添加一个线程睡眠：

在这里插入图片描述

这个时候我们运行发现，在消费者2将自己的消息打印完之后，消费者1的消息只打印了一条：
在这里插入图片描述

在这里插入图片描述

那么能不能用第二种模型实现一种能者多劳的模式呢？

消息自动确认机制

Doing a task can take a few seconds. You may wonder what happens if one of the consumers starts a long task and dies with it only partly done. With our current code, once RabbitMQ delivers a message to the consumer it immediately marks it for deletion. In this case, if you kill a worker we will lose the message it was just processing. We’ll also lose all the messages that were dispatched to this particular worker but were not yet handled.

But we don’t want to lose any tasks. If a worker dies, we’d like the task to be delivered to another worker.
完成一项任务可能需要几秒钟。你可能想知道，如果其中一个消费者开始了一项长任务，但只完成了一部分任务就去世了，会发生什么。使用我们当前的代码，一旦RabbitMQ将消息传递给消费者，它会立即将其标记为删除。在这种情况下，如果你杀死一个消费者，我们将丢失它正在处理的消息。我们还将丢失发送给该特定工作人员但尚未处理的所有消息。
但我们不想失去任何任务。如果一名消费者死亡，我们希望将任务交付给另一名消费者。

自动确认是指，当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从 RabbitMQ 的消息缓存中移除。但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么该消息就会丢失。如果设置了手动确认方式，则需要在业务处理成功后，调用channel.basicAck()，手动签收，如果出现异常，则调用channel.basicNack()方法，让其自动重新发送消息。

在这里插入图片描述

我们使用能者多劳的模式需要进行两步额外的操作：

  • 设置通道一次只能消费一个消息

  • 关闭消息的自动确认,开启手动确认消息

Customer1：

public class Customer1 {public static void main(String[] args) throws IOException {//获取连接Connection connection = RabbitMQUtils.getConnection();final Channel channel = connection.createChannel();channel.basicQos(1);//每一次只能消费一个消息channel.queueDeclare("work",true,false,false,null);//参数1:队列名称  参数2:消息自动确认 true  消费者自动向rabbitmq确认消息消费  false 不会自动确认消息channel.basicConsume("work",false,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {try{Thread.sleep(2000);}catch (Exception e){e.printStackTrace();}System.out.println("消费者-1: "+new String(body));// 参数1:确认队列中那个具体消息 参数2:是否开启多个消息同时确实channel.basicAck(envelope.getDeliveryTag(),false);}});}
}

Customer2：

public class Customer2 {public static void main(String[] args) throws IOException {//获取连接Connection connection = RabbitMQUtils.getConnection();final Channel channel = connection.createChannel();channel.basicQos(1);channel.queueDeclare("work",true,false,false,null);channel.basicConsume("work",false,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者-2: "+new String(body));//手动确认  参数1:手动确认消息标识  参数2:false 每次确认一个channel.basicAck(envelope.getDeliveryTag(), false);}});    }
}

我们对上面两段代码的一些参数或方法做出解释：

basicQos（）

void basicQos(int prefetchSize, int prefetchCount, boolean global) throws IOException;

参数：

  • prefetchSize：消息的大小

  • prefetchCount：会告诉RabbitMQ不要同时给一个消费者推送多于N个消息，即一旦有N个消息还没有ack，则该consumer将block掉，直到有消息ack

  • global：是否将上面设置应用于channel，简单点说，就是上面限制是channel级别的还是consumer级别

channel.basicAck()

void basicAck(long deliveryTag, boolean multiple) throws IOException;

参数：

  • deliveryTag：该消息的index

  • multiple：是否批量处理.

    • true:将一次性ack所有小于deliveryTag的消息

envelope.getDeliveryTag()

在这里插入图片描述
这个方法是表示消息的唯一标识ID，返回的是一个正整数，是rabbitmq来自增设置的

第三种模型(fanout)

fanout 扇出 也称为广播

在这里插入图片描述

在广播模式下，消息发送流程是这样的：

  • 可以有多个消费者
  • 每个消费者有自己的queue（队列）
    • 我们这里创建的队列是临时的，用完之后就会删除
  • 每个队列都要绑定到Exchange（交换机）
  • 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。
  • 交换机把消息发送给绑定过的所有队列
  • 队列的消费者都能拿到消息。实现一条消息被多个消费者消费

使用场景：
比如我们的商品购物车，在我们结算的时候，我们可能会跟多个系统进行交互，订单系统、库存系统等等。这个时候我们的购物车信息会被多条队列给消费。

在fanout模式下，路由的相关配置没有意义，相关参数可以空着

开发生产者

public class Provider {public static void main(String[] args) throws IOException {//获取连接对象Connection connection = RabbitMQUtils.getConnection();Channel channel = connection.createChannel();//将通道声明指定交换机   //参数1: 交换机名称    参数2: 交换机类型  fanout 广播类型channel.exchangeDeclare("logs","fanout");//发送消息channel.basicPublish("logs","",null,"fanout type message".getBytes());//释放资源RabbitMQUtils.closeConnectionAndChanel(channel,connection);}
}

channel.exchangeDeclare("logs","fanout");:

  • 参数1: 交换机名称
  • 参数2: 交换机类型
    • fanout是广播类型

开发消费者

public class Customer1 {public static void main(String[] args) throws IOException {//获取连接对象Connection connection = RabbitMQUtils.getConnection();Channel channel = connection.createChannel();//通道绑定交换机channel.exchangeDeclare("logs","fanout");//临时队列String queueName = channel.queueDeclare().getQueue();//绑定交换机和队列channel.queueBind(queueName,"logs","");//消费消息channel.basicConsume(queueName,true,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者1: "+new String(body));}});}
}

其他几个消费者同理

我们创建三个消费者，把他们开启之后，再开启我们的生产者，测试结果如下：
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

第四种模型(Routing)

在这里插入图片描述
第五种模型其实是第四种模型的一个分支，如果我们叫第四种模型为路由的话，那么我们可以说第五种模型是动态路由。第四种模型我们也可以叫做direct模型(直连)

在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用direct类型的Exchange。

在Routing模型下：

  • 队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
  • 消息的发送方在 向 Exchange发送消息时，也必须指定消息的 RoutingKey。
  • Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息

流程:

在这里插入图片描述

图解：

  • P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
  • X：Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列
  • C1：消费者，其所在队列指定了需要routing key 为 error 的消息
  • C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

开发生产者

public class Provider {public static void main(String[] args) throws IOException {//获取连接对象Connection connection = RabbitMQUtils.getConnection();//获取连接通道对象Channel channel = connection.createChannel();String exchangeName = "logs_direct";//通过通道声明交换机  参数1:交换机名称  参数2:direct  路由模式channel.exchangeDeclare(exchangeName,"direct");//发送消息String routingkey = "error";channel.basicPublish(exchangeName,routingkey,null,("指定的route key"+key+"的消息").getBytes());//关闭资源RabbitMQUtils.closeConnectionAndChanel(channel,connection);}
}

这个时候我们就已经可以看到我们的交换机了：
在这里插入图片描述

开发消费者

我们这里开发两个消费者：

  • 消费者1拿到routekey为error的消息
  • 消费者2拿到routekey为info、error、warning的消息
public class Customer1 {public static void main(String[] args) throws IOException {Connection connection = RabbitMQUtils.getConnection();Channel channel = connection.createChannel();String exchangeName = "logs_direct";//通道声明交换机以及交换的类型channel.exchangeDeclare(exchangeName,"direct");//创建一个临时队列String queue = channel.queueDeclare().getQueue();//基于route key绑定队列和交换机channel.queueBind(queue,exchangeName,"error");//获取消费的消息channel.basicConsume(queue,true,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者1: "+ new String(body));}});}
}

public class Customer2 {public static void main(String[] args) throws IOException {Connection connection = RabbitMQUtils.getConnection();Channel channel = connection.createChannel();String exchangeName = "logs_direct";//声明交换机 以及交换机类型 directchannel.exchangeDeclare(exchangeName,"direct");//创建一个临时队列String queue = channel.queueDeclare().getQueue();//临时队列和交换机绑定channel.queueBind(queue,exchangeName,"info");channel.queueBind(queue,exchangeName,"error");channel.queueBind(queue,exchangeName,"warning");//消费消息channel.basicConsume(queue,true,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者2: "+new String(body));}});}
}

我们测试一下：

测试生产者发送Route key为error的消息时

在这里插入图片描述

在这里插入图片描述

测试生产者发送Route key为info的消息时

在这里插入图片描述

在这里插入图片描述

第五种模型(Topic)

Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！这种模型Routingkey 一般都是由一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

在这里插入图片描述

  • *(star) ：匹配不多不少恰好1个词
  • #(hash): 匹配0个或多个词

例如：

audit.#    匹配audit.irs.corporate或者 audit.irs 等
audit.*   只能匹配audit.irs

开发生产者

public class Provider {public static void main(String[] args) throws IOException {//获取连接对象Connection connection = RabbitMQUtils.getConnection();Channel channel = connection.createChannel();//声明交换机以及交换机类型 topicchannel.exchangeDeclare("topics","topic");//发布消息String routekey = "user";channel.basicPublish("topics",routekey,null,("这里是topic动态路由模型,routekey: ["+routekey+"]").getBytes());//关闭资源RabbitMQUtils.closeConnectionAndChanel(channel,connection);}
}

开发消费者

我们还是开发两个消费者：

消费者1Routing Key中使用*通配符方式

public class Customer1 {public static void main(String[] args) throws IOException {//获取连接Connection connection = RabbitMQUtils.getConnection();Channel channel = connection.createChannel();//声明交换机以及交换机类型channel.exchangeDeclare("topics","topic");//创建一个临时队列String queue = channel.queueDeclare().getQueue();//绑定队列和交换机  动态统配符形式route keychannel.queueBind(queue,"topics","user.*");//消费消息channel.basicConsume(queue,true,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者1: "+ new String(body));}});}
}

消费者2中Routing Key中使用#通配符方式

public class Customer2 {public static void main(String[] args) throws IOException {//获取连接Connection connection = RabbitMQUtils.getConnection();Channel channel = connection.createChannel();//声明交换机以及交换机类型channel.exchangeDeclare("topics","topic");//创建一个临时队列String queue = channel.queueDeclare().getQueue();//绑定队列和交换机  动态统配符形式route keychannel.queueBind(queue,"topics","user.#");//消费消息channel.basicConsume(queue,true,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者2: "+ new String(body));}});}
}

这个时候我们测试的结果就是：

  • 消费者2可以拿到消息
  • 消费者1拿不到消息

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

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

相关文章

3.2-Docker Image概述

3.2-Docker Image概述

常用docker命令&#xff1a; 查看docker image有哪些 docker image ls Image的获取方式
阅读更多...
docker 安装 minio (单体架构)

docker 安装 minio (单体架构)

文字归档&#xff1a;https://www.yuque.com/u27599042/coding_star/qcsmgom7basm6y64 查询 minio 镜像 docker search minio拉取镜像 docker pull minio/minio创建启动 minio 容器 用户名长度至少为 3&#xff0c;密码长度至少为 8 docker run \ -p 9000:9000 \ -p 9090:909…
阅读更多...
51单片机-中断

51单片机-中断

文章目录 前言 前言 #include <reg52.h> #include <intrins.h>sbit key_s2P3^0; sbit flagP3^7;void delay(unsigned int z){unsigned int x,y;for(xz;x>0;x--)for(y114;y>0;y--); }void int_init(){EA1;EX11;IT11;}void main(){int_init();while(1){if (key…
阅读更多...
zabbix监控安装-linux

zabbix监控安装-linux

zabbix6.4中文文档1. 简介 (zabbix.com) Zabbix 是一个企业级的开源分布式监控解决方案。 1.zabbix结构体系 Server&#xff1a; server 是存储所有配置、统计和操作数据的中央存储库。 Proxy&#xff1a; zabbix proxy可以代替 Zabbix server 收集性能和可用性数据。p…
阅读更多...
引用类型；强引用；软引用；弱引用和虚引用

引用类型；强引用；软引用；弱引用和虚引用

概述 平时在编写代码的时候内存都是由jvm管理&#xff0c;对象的回收也是jvm在管理&#xff1b; 但是有些时候jvm无法回收对象&#xff0c;最后就会抛出oom异常. 那么那些回收不了的对象肯定有区别于能回收的对象&#xff1b; 先上一波引用类型介绍 强引用 比如平常我们直…
阅读更多...
【pyspider】爬取ajax请求数据（post），如何处理python2字典的unicode编码字段？

【pyspider】爬取ajax请求数据（post），如何处理python2字典的unicode编码字段？

情景&#xff1a;传统的爬虫只需要设置fetch_typejs即可&#xff0c;因为可以获取到整个页面。但是现在ajax应用越来越广泛&#xff0c;所以有的网页不能用此种爬虫类型来获取页面的数据&#xff0c;只能用slef.crawl()来发起http请求来抓取数据。 直接上例子&#xff1a; 可以…
阅读更多...
大数据学习之Spark性能优化

大数据学习之Spark性能优化

文章目录 Spark三种任务提交模式宽依赖和窄依赖StageSpark Job的三种提交模式 Shuffle机制分析未优化的Hash Based Shuffle优化后的Hash Based ShuffleSort-Based Shuffle Spark之checkpointcheckpoint概述checkpoint与持久化的区别checkPoint的使用checkpoint源码分析 Spark程…
阅读更多...
SSM之spring注解式缓存redis

SSM之spring注解式缓存redis

&#x1f3c5;我是默&#xff0c;一个在CSDN分享笔记的博主。&#x1f4da;&#x1f4da; ​ &#x1f31f;在这里&#xff0c;我要推荐给大家我的专栏《Linux》。&#x1f3af;&#x1f3af; &#x1f680;无论你是编程小白&#xff0c;还是有一定基础的程序员&#xff0c;这…
阅读更多...
【Sql】sql server数据库提示：执行Transact-SQL语句或批处理时发生了异常。 无法打开数据库msdb，错误：926。

【Sql】sql server数据库提示：执行Transact-SQL语句或批处理时发生了异常。 无法打开数据库msdb，错误：926。

【问题描述】 打开sql server2008r2数据库的时候&#xff0c; 系统提示执行Transact-SQL语句或批处理时发生了异常。 无法打开数据库msdb&#xff0c;错误&#xff1a;926。 【概念理解】 首先MSDB数据库是的作用&#xff1a; 用于给SQL Server代理提供必要的信息来运行调度警…
阅读更多...
项目实战：组件扫描(4)-筛选带有RequestMapping注解的bean实例

项目实战：组件扫描(4)-筛选带有RequestMapping注解的bean实例

1、ControllerDefinition package com.csdn.mymvc.core; import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; import java.lang.reflect.Method; import java.util.HashMap; import java.util.Map; //假设有一个uri是&#xff1a;/fruit…
阅读更多...
PTA_乙级_1001_C++

PTA_乙级_1001_C++

思路&#xff1a;使用判断语句即可&#xff0c;使用while进行循环&#xff0c;终止条件是n不等于1&#xff0c;然后用if-else判断奇数偶数 #include <iostream> using namespace std;int main(){int n;int count0;cin>>n;while(n!1){if(n%20){n/2;}else{n3*n1;n/2…
阅读更多...
技术分享 | web自动化测试-PageObject 设计模式

技术分享 | web自动化测试-PageObject 设计模式

为 UI 页面写测试用例时&#xff08;比如 web 页面&#xff0c;移动端页面&#xff09;&#xff0c;测试用例会存在大量元素和操作细节。当 UI 变化时&#xff0c;测试用例也要跟着变化&#xff0c; PageObject 很好的解决了这个问题。 使用 UI 自动化测试工具时&#xff08;包…
阅读更多...
【python基础】时间模块的time的下面的方法使用解析

【python基础】时间模块的time的下面的方法使用解析

文章目录 一、time三种表现形式二、time模块常用的方法1.时间戳&#xff08;timestamp&#xff09;表现方式time.time()time.mktime() 2.时间元组&#xff08;struct_itme&#xff09;表现方式time.localtime()time.gmtime()time.strptime(str,format) 3.默认时间字符串&#x…
阅读更多...
『MySQL快速上手』-③-库的操作

『MySQL快速上手』-③-库的操作

文章目录 1.创建数据库2.创建数据库案例3.字符集和校验规则3.2 校验规则对数据库的影响3.2.1 进行查询3.2.2 进行排序 4.字符集和检验规则的作用5.操纵数据库5.1 查看数据库5.2 显示创建语句5.3 修改数据库5.4 数据库删除 6.备份与恢复6.1 备份6.2 还原6.3 注意事项 7.查看数据…
阅读更多...
Jmeter_逻辑控制器

Jmeter_逻辑控制器

逻辑控制器 控制取样器执行顺序的组件实现(分支 循环) 分类 1、如果(if) 控制器 分支实现 2、forEach控制器 循环往复实现 3、循环控制器 循环往复实现 如果(if) 控制器 需求1:测试计划中定义一个 http 请求访问百度&#xff0c;但是该请求不是无条件执行的&#xff0c;…
阅读更多...
前端项目导入vue和element

前端项目导入vue和element

1.安装nodejs 下载链接https://cdn.npmmirror.com/binaries/node/v18.18.0/node-v18.18.0-x64.msi 进入cmd 命令行模式 管理员身份运行 输入 &#xff08;node -v&#xff09;能看到版本号 npm config set prefix "C:\Program Files\nodejs" 默认路径 npm config…
阅读更多...
聊一聊 tcp/ip 在.NET故障分析的重要性

聊一聊 tcp/ip 在.NET故障分析的重要性

一&#xff1a;背景 1. 讲故事 这段时间分析了几个和网络故障有关的.NET程序之后&#xff0c;真的越来越体会到计算机基础课的重要&#xff0c;比如 计算机网络 课&#xff0c;如果没有对 tcpip协议 的深刻理解&#xff0c;解决这些问题真的很难&#xff0c;因为你只能在高层做…
阅读更多...
【微服务】mysql + elasticsearch数据双写设计与实现

【微服务】mysql + elasticsearch数据双写设计与实现

目录 一、前言 二、为什么使用mysqles双写 2.1 单用mysql的问题 2.2 为什么不直接使用es 2.2.1 非关系型表达 2.2.2 不支持事务 2.2.3 多字段将造成性能低下 三、mysqles双写方案设计要点 3.1 全新设计 VS 中途调整架构 3.2 全表映射 VS 关键字段存储 3.2.1 最大程度…
阅读更多...
《强化学习与机器人控制》：探索深度学习的应用宝典

《强化学习与机器人控制》：探索深度学习的应用宝典

《强化学习与机器人控制》是一本涵盖了广泛主题的深度著作&#xff0c;它不仅介绍了人机交互控制和强化学习的基本原理&#xff0c;还深入探讨了无模型强化学习控制器以及其在机器人控制中的应用。这本书对于研究生和执业工程师来说是一本极具价值的参考书&#xff0c;它为读者…
阅读更多...
Markov Chain Fingerprinting to Classify Encrypted Traffic 论文笔记

Markov Chain Fingerprinting to Classify Encrypted Traffic 论文笔记

0.Abstract 在本文中&#xff0c;提出了用于SSL/TLS会话中传输的应用程序流量的随机指纹。这个指纹基于一阶齐次马尔可夫链&#xff0c;模型识别应用程序的准确率&#xff0c;并提供了检测异常对话的可能性。 1.Introduction 通过SSL/TLS会话时的头部信息创建统计指纹&#xff…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023