【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.mzph.cn/news/119551.shtml

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

相关文章

jvm摘要

jvm摘要

第 2 章 Java 内存区域与内存溢出异常 2.2 运行时数据区域 程序计数器-线程私有:是一块较小的内存空间&#xff0c;它可以看作是当前线程所执行的字节码的行号指示器。 程序计数器是唯一一个没有规定任何OutOfMemoryError 情况的区域。 Java 虚拟机栈-线程私有:用于执行Java …
阅读更多...
悟空crm安装搭建 报错[0] RedisException in Redis.php line 56问题处理办法

悟空crm安装搭建 报错[0] RedisException in Redis.php line 56问题处理办法

相信很多朋友进行安装悟空crm的时候 提示错误&#xff1a; [0] RedisException in Redis.php line 56 Connection refused 不知道怎么样处理是吧~~~ $this->options array_merge($this->options, $options);}# redis 密码$password config(cache.password);if (!empty…
阅读更多...
（二开）Flink 修改源码拓展 SQL 语法

（二开）Flink 修改源码拓展 SQL 语法

1、Flink 扩展 calcite 中的语法解析 1&#xff09;定义需要的 SqlNode 节点类-以 SqlShowCatalogs 为例 a&#xff09;类位置 flink/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/dql/SqlShowCatalogs.java 核心方法&#xff1a; Override pu…
阅读更多...
【C++】priority_queue仿函数

【C++】priority_queue仿函数

今天我们来学习C中另一个容器适配器&#xff1a;优先级队列——priority_queue&#xff1b;和C一个重要组件仿函数&#xff1a; 目录 一、priority_queue 1.1 priority_queue是什么 1.2 priority_queue的接口 1.2.1 priority_queue使用举例 二、仿函数 三、关于priority…
阅读更多...
Docker GitLab-Runner安装

Docker GitLab-Runner安装

Docker GitLab-Runner安装 GitLab-Runner安装 问题合集GitLab 域名的配置修改Runner容器内注册失败&#xff0c;提示 dial tcp: lookup home.zsl0.com on 192.168.254.2:53: no such host GitLab-Runner 安装 拉去gitlab/gitlab-runner镜像 docker pull gitlab/gitlab-runne…
阅读更多...
适用于 Windows 10 和 Windows 11 设备的笔记本电脑管理软件

适用于 Windows 10 和 Windows 11 设备的笔记本电脑管理软件

便携式计算机管理软件使 IT 管理员能够简化企业中使用的便携式计算机的部署和管理&#xff0c;当今大多数员工使用Windows 笔记本电脑作为他们的主要工作机器&#xff0c;他们确实已成为几乎每个组织不可或缺的一部分。由于与台式机相比&#xff0c;笔记本电脑足够便携&#xf…
阅读更多...
Git的远程仓库

Git的远程仓库

Git的远程仓库 添加远程仓库从远程库克隆 添加远程仓库 你在本地创建了一个Git仓库后&#xff0c;又想在GitHub创建一个Git仓库&#xff0c;并且让这两个仓库进行远程同步&#xff0c;这样&#xff0c;GitHub上的仓库既可以作为备份&#xff0c;又可以让其他人通过该仓库来协作…
阅读更多...
JoySSL：免费SSL证书的新选择

JoySSL：免费SSL证书的新选择

阿里云曾经提供了一年期的免费SSL证书&#xff0c;然而从下个月14号开始&#xff0c;阿里云不再提供免费的一年期SSL证书&#xff0c;而是改为68元/张/年&#xff0c;这对于很多网站建设公司需要大量基本证书来说&#xff0c;无疑是一种负担&#xff0c;又不知道该去哪里获取可…
阅读更多...
jmeter如何测试websocket接口？

jmeter如何测试websocket接口？

jmeter做接口测试&#xff0c;很多人都是做http协议的接口&#xff0c;就有很多人问websocket的接口怎么测试啊&#xff1f; 首先&#xff0c;我们要明白&#xff0c;websocket接口是什么接口。 然后&#xff0c;我们怎么用jmeter测试&#xff1f; jmeter要测试websocket接口…
阅读更多...
fastjson对象序列化的问题

fastjson对象序列化的问题

今天偶然遇到一个fastjson将字符串反序列化为一个对象的时候的问题&#xff0c;就是简单的通过com.alibaba.fastjson.JSON将对象转为字符串&#xff0c;然后再从字符串转换为原类型的对象。 涉及的代码也非常简单 package cn.edu.sgu.www.mhxysy.service.role.impl;import cn…
阅读更多...
Linux文件I/O

Linux文件I/O

下面的内容需要了解系统调用&#xff0c;可看下面的链接&#xff1a; 系统调用来龙去脉-CSDN博客 1.底层文件IO和标准IO 这里指的是操作系统提供的IO服务&#xff0c;不同于ANSI建立的标准IO。 底层IO和标准IO各自所使用的函数&#xff1a; 区别&#xff1a; 1.底层文件IO不…
阅读更多...
Android 13 Framework 裁剪

Android 13 Framework 裁剪

裁剪应用 1. 修改 build/core/product.mk 添加PRODUCT_DEL_PACKAGES变量的声明 新增一行_product_single_value_vars PRODUCT_DEL_PACKAGES # The first API level this product shipped with _product_single_value_vars PRODUCT_SHIPPING_API_LEVEL _product_single_val…
阅读更多...
本地存储 sessionStoragelocalStorage

本地存储 sessionStoragelocalStorage

随着互联网的快速发展&#xff0c;基于网页的应用越来越普遍&#xff0c;同时也变的越来越复杂&#xff0c;为了满足各种各样的需求&#xff0c;会经常性在本地存储大量的数据&#xff0c;HTML5规范提出了相关解决方案。 本地存储特性 数据存储在用户浏览器中 设置、读取方便、…
阅读更多...
使用字节流读取文件中的数据的几种方式

使用字节流读取文件中的数据的几种方式

public class FileReader02_ {public static void main(String[] args) {}Testpublic void m1() {String filePath "e:\\hello.txt";FileReader fileReader null;int date0;try {fileReader new FileReader(filePath);//循环读取 使用readwhile ((datefileReader.…
阅读更多...
【opencv】【CPU】windows10下opencv4.8.0-cuda C++版本源码编译教程

【opencv】【CPU】windows10下opencv4.8.0-cuda C++版本源码编译教程

【opencv】【CPU】windows10下opencv4.8.0-cuda C版本源码编译教程 提示:博主取舍了很多大佬的博文并亲测有效,分享笔记邀大家共同学习讨论 文章目录 【opencv】【CPU】windows10下opencv4.8.0-cuda C版本源码编译教程前言准备工具cmakeopencv4.8.0opencv_contrib CMake编译VS2…
阅读更多...
uniapp: 本应用使用HBuilderX x.x.xx 或对应的cli版本编译，而手机端SDK版本是 x.x.xx。不匹配的版本可能造成应用异常。

uniapp: 本应用使用HBuilderX x.x.xx 或对应的cli版本编译，而手机端SDK版本是 x.x.xx。不匹配的版本可能造成应用异常。

文章目录 前言一、原因分析二、解决方案2.1、方案一&#xff1a;更新HbuilderX版本2.2、方案二&#xff1a;设置固定的版本2.3、方案三&#xff1a;忽略版本&#xff08;不推荐&#xff09; 三、总结四、感谢 前言 项目场景&#xff1a;示例&#xff1a;通过使用HbuilderX打包…
阅读更多...
Apache JMeter 安装教程

Apache JMeter 安装教程

下载&#xff1a; 注意事项&#xff1a;使用JMeter前需要配置JDK环境 下载地址 下载安装以后&#xff0c;打开安装的bin目录 D:\software\apache-jmeter-5.4.1\apache-jmeter-5.4.1\bin&#xff0c;找到jmeter.bat&#xff0c;双击打开 打开后的样子 语言设置&#xff1a; 1…
阅读更多...
【高效开发工具系列】Postman

【高效开发工具系列】Postman

&#x1f49d;&#x1f49d;&#x1f49d;欢迎来到我的博客&#xff0c;很高兴能够在这里和您见面&#xff01;希望您在这里可以感受到一份轻松愉快的氛围&#xff0c;不仅可以获得有趣的内容和知识&#xff0c;也可以畅所欲言、分享您的想法和见解。 推荐:kuan 的首页,持续学…
阅读更多...
“人类高质量数据”如何训练计算机视觉模型？

“人类高质量数据”如何训练计算机视觉模型？

人类的视觉系统可以复制吗&#xff1f; 答案是肯定的。 计算机视觉 (Computer Vision) 技术的不断普及&#xff0c;让机器识别和处理图像就像人的大脑一样&#xff0c;且速度更快、更准确。 机器像人类一样去“思考” 计算机视觉 (Computer Vision) 是近年来人工智能增长最快…
阅读更多...
Python遍历删除列表元素的一个奇怪bug

Python遍历删除列表元素的一个奇怪bug

假定有一个Python列表&#xff0c;比如[CFFEX.IF, CFFEX.TS,SHFE.FU]&#xff0c;现在需要将其中带‘CFFEX’前缀的所有元素都删除。在使用列表推导式一行代码搞定之前&#xff0c;用了一种最朴素的遍历删除方法&#xff0c;结果出现了意想不到的的问题。复盘了下&#xff0c;结…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023