【02】RabbitMQ客户端应用开发实战

news/2024/10/21 6:38:23/

1、RabbitMQ基础编程模型

  • RabbitMQ提供了很多种主流编程语言的客户端支持，这里只分析Java语言的客户端。
  • 在上一章节提供了一个简单的RabbitMQ的客户端的实现，下面就以此为基础，了解RabbitMQ客户端开发的基础流程。

1.1 Maven依赖

  • amqp是一种标准的消息驱动实现协议，RabbitMQ是对这一协议的具体实现。由于协议具有稳定性，所以，通常RabbitMQ的客户端不一定要求与服务端版本一致。

    	<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.21.0</version></dependency>

1.2 基础编程模型

  • 结合上个案例，详细拆解每个核心概念的具体实现方式以及功能扩展。
1.2.1 step1：首先创建连接，获取Channel

  • 通常情况下，我们在客户端中都只是创建一个Channel就可以了。因为一个Channel只要不关闭，是可以一直复用的。但是，如果你想要创建多个Channel，就要注意Channel冲突的问题。

  • 在创建Channel时，可以在createChannel()方法中传入一个分配Int类型的参数 channelNumber。这个参数就会作为Channel的唯一标识，而RabbitMQ为了防止ChannelNumber重复的方式是：如果对应的Channel没有创建过，就会创建一个新的Channel。如果已经创建过了，这时机会返回一个null。

    	ConnectionFactory factory = new ConnectionFactory();factory.setHost("192.168.121.134");factory.setPort(5672);factory.setUsername("admin");factory.setPassword("admin");factory.setVirtualHost("/master");//建立连接Connection connection = factory.newConnection();//创建信道Channel channel = connection.createChannel();
1.2.2 step2：声明Exchange交换机
  • 关键代码：
    channel.exchangeDeclare(String exchange,String type, boolean durable,boolean autoDelete,Map<String, Object> arguments) throws IOException;
  • api说明：
    • exchange：设置交换机的名称；
    • type：交换机的类型；
    • durable：是否持久化（交换机在服务重启后仍然有效）；
    • autoDelete：是否自动删除（如果服务器不再使用是否自动删除该交换机）；
    • arguments：交换机的其他属性（构造参数）；
  • 在声明Exchange时需要注意，如果Broker（消息队列服务器实体）上没有对应的Exchange，那么RabbitMQ会自动创建一个新的交换机。如果Broker上已经存在这个Exchange，那么你声明这个交换机的参数时必须要与Broker上的Exchange参数保持一致，如果不一致就会报错；
  • 声明时的大部分参数在管理平台是可以看到的：
  • 重点需要关注的是Exchange的类型，它共有四种类型，分别对应了四种不同的消息分发逻辑。
1.2.3 step3：声明Queue

  • 关键代码：

     channel.queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments);

  • api说明：

    • queue：设置队列的名称；
    • durable：是否持久化（服务器重启后队列仍然有效）；
    • exclusive：是否独占（仅限于此连接）；
    • autoDelete：是否自动删除（服务器不再使用时自动删除）；
    • arguments：队列的其他属性（构造参数）；

  • 与Exchange一样，如果你声明的Queue在Broker上不存在，RabbitMQ会创建一个新的队列。但是如果Broker上已经有了这个队列，那么声明的属性必须和Broker上的队列保持一致，否则也会报错。

  • 声明Queue时，同样大部分的参数是可以从管理平台看到的：

    • Durablility表示是否持久化。Durable选项表示会将队列的消息写入硬盘，这样服务重启后这些消息就不会丢失。而另外一个选项Transient表示不持久化，消息只在内存中流转。这样服务重启后这些消息就会丢失。当然这也意味着消息读写的效率会比较高。
    • 但是与Exchange不同的是，在管理控制台上，队列有个Type参数，这个参数并没有在API中体现。这里是有历史版本的原因的，也有不同类型的队列实现方式不同的原因。例如：对于Quorum和Stream类型，根本没有Durability参数，因为它们的消息默认就是必须要持久化的。

  • 客户端API中默认只能声明Classic类型的队列。如果需要声明其他类型的队列，只能通过后面的arguments参数来区分。

  • 如果要声明一个Quorum队列，则只需要在后面的arguments中传入一个参数，x-queue-type，参数值设定为quorum。

    Map<String,Object> params = new HashMap<>();
params.put("x-queue-type","quorum");
//声明Quorum队列的方式就是添加一个x-queue-type参数，指定为quorum。默认是classic
channel.queueDeclare(QUEUE_NAME, true, false, false, params);
//需要注意的是：对于Quorum类型，durable参数必须是true，设置成false的话，会报错。同样，exclusive参数必须设置为false

  • 如果要声明一个Stream队列，则 x-queue-type 参数要设置为 stream 

    Map<String,Object> params = new HashMap<>();
params.put("x-queue-type","stream");
params.put("x-max-length-bytes", 20_000_000_000L); // maximum stream size: 20 GB
params.put("x-stream-max-segment-size-bytes", 100_000_000); // size of segment files: 100 MB
channel.queueDeclare(QUEUE_NAME, true, false, false, params);
//需要注意的是：1、同样，durable参数必须是true，exclusive必须是false。 对于这两种队列，这两个参数是不可以选择的。2、x-max-length-bytes 表示日志文件的最大字节数。x-stream-max-segment-size-bytes 每一个日志文件的最大大小。这两个是可选参数，通常为了防止stream日志无限制累计，都会配合stream队列一起声明。3、stream类型的对列，并不能像之前两种对列一样使用。例如前两种对列类型，声明消费者后，可以从控制台直接发消息，消费者端就能接受到。 但是Stream类型是接收不到的

  • 实际项目中用得最多的是RabbitMQ的Classic经典队列，但是从RabbitMQ官网就能看到， 目前RabbitMQ更推荐的是使用Quorum队列。至于Stream队列目前企业用得还比较少。

1.2.4 step4：声明Exchange与Queue的绑定关系

  • 关键代码：

    channel.queueBind(String queue,String exchange,String routingKey) throws IOException;

  • api说明：

    • queue：需要绑定的队列的名称；
    • exchange：需要绑定的交换机的名称；
    • routingKey：用于绑定的路由key；

  • 如果我们声明了Exchange和Queue，那么就还需要声明它们之间的绑定关系Binding。有了Binding绑定关系，Exchange才可以知道Producer发送过来的消息将要分发到哪些Queue上。

  • 这些Binding涉及到消息的不同分发逻辑，与Exchange和Queue一样，如果Broker上没有建立绑定关系，那么RabbitMQ会按照客户端的声明，创建这些绑定关系。但是如果声明的Binding存在了，那么就需要与Broker上的保持一致。

  • 另外，在声明Binding时，还可以传入两个参数， routingKey和props。这两个参数都是跟Exchange的消息分发逻辑有关。

1.2.5 step5：Producer根据应用场景发送消息到queue

  • 关键代码：

    channel.basicPublish(
String exchange, 
String routingKey, 
BasicProperties props,
message.getBytes("UTF-8")) ;

  • api说明：

    • exchange：要发送到某个交换机的名称；
    • routingKey：路由key；
    • props：支持消息的其他属性；
    • message：消息正文；

  • 这其中Exchange如果不需要，传个空字符串就行。routingKey跟Exchange的消息分发逻辑有关。

  • 关于props参数，可以传入一些消息相关的属性。管理控制台上有明确的说明:

  • props的这些配置项，可以用RabbitMQ中提供的一个Builder对象来构建:

       AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder(); //对应页面上的Properties部分，传入一些预定的参数值。builder.deliveryMode(MessageProperties.PERSISTENT_TEXT_PLAIN.getDeliveryMode());builder.priority(MessageProperties.PERSISTENT_TEXT_PLAIN.getPriority());//builder.headers(headers);对应页面上的Headers部分。传入自定义的参数值builder.build()AMQP.BasicProperties prop = builder.build();

  • 在发送消息时要注意一下消息的持久化问题。MessageProperties.PERSISTENT_TEXT_PLAIN是RabbitMQ提供的持久化消息的默认配置。而RabbitMQ中消息是否持久化不光取决于消息，还取决于Queue。通常为了保证消息安全，会将Queue和消息同时声明为持久化。

1.2.6 step6：Consumer消费消息
  • 定义消费者，对消息进行处理，并向RabbitMQ进行消息确认。确认了之后就表明这个消息已经消费完了，否则RabbitMQ还会继续发起重试。Consumer主要有两种消费方式：
  • <1> Push 推模式
    • Consumer等待rabbitMQ 服务器将message推送过来再消费。一般是启一个一直挂起的线程来等待。
    • 关键代码：
      channel.basicConsume(String queue, boolean autoAck, Consumer callback);
    • api说明：
      • 使用服务器生成的consumerTag启动一个非本地的、非独占的消费者。
      • queue：队列名称;
      • autoAck：是否自动确认消息已被消费（如果服务器认为消息一经发送就被确认则为true；如果服务器期望显式确认，则为false）;
      • callback：回调函数（封装了消费者对消息处理的业务逻辑）；
  • <2> Pull 拉模式
    • Comsumer主动到rabbitMQ服务器上去拉取messge进行消费。
    • 关键代码：
      GetResponse response = channel.basicGet(QUEUE_NAME, boolean autoAck);
    • api说明：
      • queue：队列名称；
      • autoAck：是否自动应答；
  • 其中需要注意点的是autoAck，自动应答。Consumer处理完一条消息后，需要给Broker一个Ack应答，Broker就不会重复投递消息。如果Broker没有收到应答，就会重复投递消息。
  • 实际开发中，更建议使用Push推模式，这样消息处理能够比较及时，并且也不会给服务端带来重复查询的压力。
1.2.7 step7：完成之后关闭连接，释放资源

  • 用完之后主动释放资源。如果不主动释放的话，大部分情况下，过一段时间RabbitMQ也会将这些资源释放掉，但是这就需要额外消耗系统资源。

    channel.close(); 
conection.clouse();

  • 最后给出完整生产者发送消息的代码处理过程：

    public class CallbackProducer {private static final String EXCHANGE_NAME="testExchange1";private static final String ALTER_EXCHANGE_NAME="testAlterExchange1";private static final String QUEUE_NAME = "testQueue1";public static void main(String[] args) throws Exception {Connection connection = RabbitMQUtil.getConnection();Channel channel = connection.createChannel();//给交换机添加一个备用交换机，如果发往EXCHANGE_NAME的消息无法路由，就会转发到ALTER_EXCHANGE_NAME上。Map<String,Object> params = new HashMap<>();params.put("alternate-exchange",ALTER_EXCHANGE_NAME);channel.exchangeDeclare(EXCHANGE_NAME,BuiltinExchangeType.DIRECT,true,false,params);channel.exchangeDeclare(ALTER_EXCHANGE_NAME,BuiltinExchangeType.DIRECT,true,false,null);channel.queueDeclare(QUEUE_NAME, true, false, false, null);channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"key1");channel.confirmSelect();//如果发往Exchange的消息无法路由，就会给Producer一个通知，触发ReturnListener。channel.addConfirmListener(new ConfirmListener() {@Overridepublic void handleAck(long deliveryTag, boolean multiple) throws IOException {System.out.println("message acked; deliveryTag = " +deliveryTag);}@Overridepublic void handleNack(long deliveryTag, boolean multiple) throws IOException {System.out.println("message Nacked; deliveryTag = " +deliveryTag);}});channel.addReturnListener(new ReturnListener() {@Overridepublic void handleReturn(int replyCode, String replyText, String exchange, String routingKey,AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("message returned replayCode = "+replyCode+";replyText:"+replyText+ "; message = "+new String(body)+";props = "+properties);}});AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();builder.deliveryMode(MessageProperties.PERSISTENT_TEXT_PLAIN.getDeliveryMode());builder.priority(MessageProperties.PERSISTENT_TEXT_PLAIN.getPriority());builder.correlationId("111");String message = "correlationId=1,asdfasdfas";//发送两条消息， routingkey为key2的消息没有队列接收，就会触发return callbackchannel.basicPublish(EXCHANGE_NAME,"key1",true, builder.build(),message.getBytes(StandardCharsets.UTF_8));channel.basicPublish(EXCHANGE_NAME,"key2",true, builder.build(),message.getBytes(StandardCharsets.UTF_8));System.out.println("message sended");//不要立即结束，这样才能接收到服务端的回调。Thread.sleep(10000);channel.close();connection.close();}
}

1.3 关于消息监听与回溯

  • 前面只是列出了用得最多的几个常用的方法。但是，RabbitMQ的客户端还提供了很多重载方法和扩展方法，这些需要自行掌握。

  • 例如，在消费消息时，channel还提供了一个重载的方法：

    String basicConsume(String queue, DeliverCallback deliverCallback,  //消息传递时的回调CancelCallback cancelCallback, //消费者被取消时的回调ConsumerShutdownSignalCallback shutdownSignalCallback //通道/连接关闭时的回调) throws IOException;

  • 另外，自己再做一个小案例：

    ConnectionFactory factory = new ConnectionFactory();factory.setHost(HOST_NAME);factory.setPort(HOST_PORT);factory.setUsername(USER_NAME);factory.setPassword(PASSWORD);factory.setVirtualHost(VIRTUAL_HOST);//建立连接Connection connection = factory.newConnection();Channel channel = connection.createChannel();Map<String,Object> params = new HashMap<>();params.put("alternate-exchange",ALTER_EXCHANGE_NAME);channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT,true,false,params);channel.exchangeDeclare(ALTER_EXCHANGE_NAME,BuiltinExchangeType.DIRECT,true,false,null);channel.queueDeclare(QUEUE_NAME, true, false, false, null);channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"key1");channel.basicConsume(QUEUE_NAME, new DeliverCallback() {//正常消息传递的处理逻辑@Overridepublic void handle(String consumerTag, Delivery message) throws IOException {long deliveryTag = message.getEnvelope().getDeliveryTag();String correlationId = message.getProperties().getCorrelationId();System.out.println("收到消息的consumerTag: " + consumerTag + "\n 收到的消息message: " + new String(message.getBody())+";deliveryTag: "+deliveryTag+";correlationId: "+correlationId);channel.basicAck(deliveryTag,false);}}//消费者被取消的处理逻辑，例如：去控制台把队列删掉就会触发cancel, new CancelCallback() {@Overridepublic void handle(String consumerTag) {System.out.println("取消消息的consumerTag: " + consumerTag + "; ");}}//通道或者连接被断开的时候的处理逻辑，例如：突然关闭连接、关闭通道, new ConsumerShutdownSignalCallback() {@Overridepublic void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {System.out.println("消费者宕机的consumerTag: " + consumerTag + "\n Exception: " + sig);}});

  • 上述过程中的consumerTag代表的是与客户端的一个会话，deliveryTag代表的是这个Channel处理的一条消息。这都是RabbitMQ服务器分配的一些内部参数。日后，如果你希望对Consumer处理的每一条消息增加溯源功能时，把consumerTag+deliveryTag作为消息编号，保存下来，这就是一个不错的设计。

2、RabbitMQ常用的消息场景

  • 这一部分是学习以及使用RabbitMQ的重中之重。RabbitMQ的客户端API使用是比较简单的，但是如果面临复杂业务时，要用好RabbitMQ还是需要一些功力的。对这一部分的学习，理解业务场景是最为重要的。
  • RabbitMQ官方提供了总共七种典型的使用场景：

2.1 hello world 体验

  • 最直接的方式，P端发送一个消息到一个指定的队列，中间不需要任何exchange规则。C端按队列方式进行消费。
  • 关键代码：
    • 生产者：
      channel.queueDeclare(QUEUE_NAME,false,false,false,null);
channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
    • 消费者：
      channel.queueDeclare(QUEUE_NAME, false, false, false, null);
channel.basicConsume(QUEUE_NAME, true, Consumer callback);

2.2 Work Queues 工作序列

  • 这是RabbitMQ最基础也是最常用的一种工作机制。
  • 工作任务模式，领导部署一个任务，由下面的一个员工来处理。只关心任务被正确处理，不关心给谁处理。
  • Producer 消息发送给queue，多个 Consumer 同时往队列上消费消息。
  • 关键代码：
    • producer: 将消息直接发送到Queue上
      //任务一般是不能因为消息中间件的服务而被耽误的，所以durable设置成了true，这样，即使rabbitMQ服务断了，这个消息也不会消失
channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null); 
channel.basicPublish("", TASK_QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes("UTF-8"));
    • consumer: 每次拉取一条消息
      channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
channel.basicQos(1);
channel.basicConsume(TASK_QUEUE_NAME, false, consumer);
  • 这个模式应该是最常用的模式，有几个问题需要注意：
    • <1>：Consumer对每个消息必须应答
      • 消费端每消费完一个消息就需要给服务端一个ACK应答，这个应答可以自动应答也可以手动应答。如果消费者对某个消息消费完后一直没有给服务端应答，那么服务端会不断地将这条消息重复进行投递，这就会不断地消耗系统资源。这也就是 Poison Message (毒消息)。这是使用RabbitMQ时易犯的一个错误。
    • <2>：RabbitMQ并不能完全保证消息安全
      • 非常关键的message消息不能因为服务器出现问题而就被忽略了。如果想要保证消息不被丢失，在RabbitMQ中需要同时将队列和消息的durable属性都设置成true。
      • 但是，官方也明确说了，就算把队列和消息都持久化，RabbitMQ也并不能保证消息完全不丢失。因为，RabbitMQ对于持久化的消息，会写入到文件当中。但此时，只是写入到了PageCache缓存中，而不是磁盘。缓存中的数据断电就丢失。消息从缓存写入磁盘，需要进行一次刷盘。RabbitMQ并不会对每个消息都执行刷盘操作，而且执行该操作也是有一定的时间间隔的，因此，如果服务器异常断电，RabbitMQ在这个层面是可能面临消息丢失的问题的。
    • <3>：消息如何在多个Consumer之间分发
      • 这里，RabbitMQ默认是采用的 fair dispatch，也叫round-robin模式，就是把消息轮询，在所有consumer中轮流发送。这种方式，没有考虑消息处理的复杂度以及consumer的处理能力。而他们改进后的方案，是consumer可以向服务器声明一个prefetchCount，我把他叫做预处理能力值。
      • channel.basicQos(prefetchCount);表示当前这个consumer可以同时处理几个message。这样服务器在进行消息发送前，会检查这个consumer当前正在处理中的message(message已经发送，但是未收到consumer的basicAck)有几个，如果超过了这个consumer节点的能力值，就不再往这个consumer发布。 这种模式，官方也指出还是有问题的，消息有可能全部阻塞，所有consumer节点都超过了能力值，那消息就阻塞在服务器上，这时需要自己及时发现这个问题，采取措施，比如增加consumer节点或者其他策略;

2.3 Publish/Subscribe 发布/订阅 机制

  • type为 ‘fanout’ 的交换机：
  • 这个机制是对上面的一种补充。也就是把Producer与Consumer进行进一步的解耦。producer只负责发送消息，至于消息进入哪个queue，由exchange来分配。如上图，就是把producer发送的消息，交由exchange同时发送到两个queue里，然后由不同的Consumer去进行消费。
  • 关键代码：
    • binding：将消费的目标队列绑定到Exchange交换机上。
      channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, EXCHANGE_NAME, "");
    • producer: 只负责往exchange里发消息，后面的事情不管。
      channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
  • 关键所在：就是type为”fanout” 的exchange，这种类型的exchange只负责往所有已绑定的队列上发送消息。

2.4 Routing 基于内容的路由

  • type为 ‘direct’ 的交换机：
  • 在上一章 exchange 往所有队列发送消息的基础上，增加一个路由配置，指定exchange如何将不同类别的消息分发到不同的queue上。
  • 关键代码：
    • Producer 同样是往Exchange发送消息，但是需要指定一个routingKey
      channel.exchangeDeclare(EXCHANGE_NAME, "direct");
channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));
    • Bindings 在建立Exchange和Queue的绑定关系时，需要指定routingKey。
      channel.exchangeDeclare(EXCHANGE_NAME, "direct");
channel.queueBind(queueName, EXCHANGE_NAME, routingKey1);
channel.queueBind(queueName, EXCHANGE_NAME, routingKey2);
channel.basicConsume(queueName, true, consumer);
  • 消息就会根据routingkey转发到对应的Queue上，然后给消费者处理。

2.5 Topics 基于话题的路由

  • type为 ‘topic’ 的exchange：
  • 这个模式也就在Routing模式的基础上，对routingKey进行了模糊匹配。单词之间用 ‘,’ 隔开，‘*’ 代表一个具体的单词，‘#’ 代表0个或多个单词。
  • 关键代码：
    • Producer，依然是往Exchange发送消息，并且需要带上routingKey。
      channel.exchangeDeclare(EXCHANGE_NAME, "topic");
channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));
    • Bindings: 绑定routingKey。
      channel.exchangeDeclare(EXCHANGE_NAME, "topic");
channel.queueBind(queueName, EXCHANGE_NAME, routingKey1);
channel.queueBind(queueName, EXCHANGE_NAME, routingKey2);
channel.basicConsume(queueName, true, consumer);
  • 消息根据routingKey进行转发时，会进行模糊匹配。

2.6 Publisher Confirms 发送者消息确认

  • RabbitMQ的消息可靠性是非常高的，但是他以往的机制都是保证消息发送到了MQ之后，可以推送到消费者消费，不会丢失消息。但是发送者发送消息是否成功是没有保证的。我们可以回顾下，发送者发送消息的基础API：channel.basicPublish方法是没有返回值的，也就是说，一次发送消息是否成功，应用是不知道的，这在业务上就容易造成消息丢失。而这个模块就是通过给发送者提供一些确认机制，来保证这个消息发送的过程是成功的。

  • 发送者确认模式默认是不开启的，所以如果需要开启发送者确认模式，需要手动在channel中进行声明。

    channel.confirmSelect();

  • 在官网的示例中，重点解释了三种策略：

    • <1> 发布单条消息
      • 即：每当发布一条消息就确认一次。核心代码如下：
         channel.confirmSelect();for (int i = 0; i < MESSAGE_COUNT; i++) {String body = String.valueOf(i);channel.basicPublish("", queue, null, body.getBytes());channel.waitForConfirmsOrDie(5_000);}
      • channel.waitForConfirmsOrDie(5_000); 这个方法就会在channel端等待RabbitMQ给出一个响应，用来表明这个消息已经正确发送到了RabbitMQ服务端。但是要注意，这个方法会同步阻塞channel，在等待确认期间，channel将不能再继续发送消息，也就是说会明显降低集群的发送速度即吞吐量。
      • 官方说明了，其实channel底层是异步工作的，会将channel阻塞住，然后异步等待服务端发送一个确认消息，才解除阻塞。但是我们在使用时，可以把他当作一个同步工具来看待。
      • 然后如果到了超时时间，还没有收到服务端的确认机制，那就会抛出异常。然后通常处理这个异常的方式是记录错误日志或者尝试重发消息，但是尝试重发时一定要注意不要使程序陷入死循环。
    • <2> 发送批量消息
      • 之前单条确认的机制会对系统的吞吐量造成很大的影响，所以稍微中和一点的方式就是发送一批消息后，再一起确认。核心代码如下：
        int batchSize = 100;
int outstandingMessageCount = 0;for (int i = 0; i < MESSAGE_COUNT; i++) {String body = String.valueOf(i);ch.basicPublish("", queue, null, body.getBytes());outstandingMessageCount++;if (outstandingMessageCount == batchSize) {ch.waitForConfirmsOrDie(5_000);outstandingMessageCount = 0;}}if (outstandingMessageCount > 0) {ch.waitForConfirmsOrDie(5_000);}
      • 这种方式可以稍微缓解下发送者确认模式对吞吐量的影响。但是也有个固有的问题就是，当确认出现异常时，发送者只能知道是这一批消息出问题了， 而无法确认具体是哪一条消息出了问题。所以接下来就需要增加一个机制能够具体对每一条发送出错的消息进行处理。
    • <3> 异步确认消息
      • 实现的方式也比较简单，Producer在channel中注册监听器来对消息进行确认。
        ch.confirmSelect();ConcurrentNavigableMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();ConfirmCallback cleanOutstandingConfirms = (sequenceNumber, multiple) -> {if (multiple) {ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(sequenceNumber, true);confirmed.clear();} else {outstandingConfirms.remove(sequenceNumber);}
};ch.addConfirmListener(cleanOutstandingConfirms, (sequenceNumber, multiple) -> {String body = outstandingConfirms.get(sequenceNumber);System.err.format("Message with body %s has been nack-ed. Sequence number: %d, multiple: %b%n",body, sequenceNumber, multiple);cleanOutstandingConfirms.handle(sequenceNumber, multiple);
});
      • 这里注册了两个监听器，是因为：执行成功一个，执行失败一个；
      • 然后关于这个 ConfirmCallback，这是个监听器接口，里面只有一个方法： void handle(long sequenceNumber, boolean multiple) throws IOException; 这方法中的两个参数:
        • sequenceNumer：这个是一个唯一的序列号，代表一个唯一的消息。在RabbitMQ中，他的消息体只是一个二进制数组，默认消息是没有序列号的。那么在回调的时候，Producer怎么知道是哪一条消息成功或者失败呢？RabbitMQ提供了一个方法int sequenceNumber = channel.getNextPublishSeqNo(); 来生成一个全局递增的序列号，这个序列号将会分配给新发送的那一条消息。然后应用程序需要自己来将这个序列号与消息对应起来。
        • multiple：这个是一个Boolean型的参数。如果是false，就表示这一次只确认了当前一条消息。如果是true，就表示RabbitMQ这一次确认了一批消息，在sequenceNumber之前的所有消息都已经确认完成了。
    • <4> 三种确认机制的区别
      • 这三种确认机制都能够提升Producer发送消息的安全性。通常情况下，第三种异步确认机制的性能是最好的。
      • 实际上，在当前版本中，Publisher不光可以确认消息是否到了Exchange，还可以确认消息是否从Exchange成功路由到了Queue。在Channel中可以添加一个ReturnListener。这个ReturnListener就会监控到这一部分发送成功了，但是无法被Consumer消费到的消息。
      • 那接下来，更进一步，这部分消息要如何处理呢？还记得在Web控制台声明Exchange交换机的时候，还可以添加一个属性吗？是的。当前版本在Exchange交换机中可以添加一个属性alternate-exchange。这个属性可以指向另一个Exchange。其作用，就是将这些无法正常路由的消息转到另外的Exchange进行兜底处理。

2.7 Headers 头部路由机制

  • 这种策略在实际中用得比较少，但是在某些比较特殊的业务场景，还是挺好用的。

  • 官网示例中的集中路由策略， direct,fanout,topic等这些Exchange，都是以routingkey为关键字来进行消息路由的，但是这些Exchange有一个普遍的局限就是都是只支持一个字符串的形式，而不支持其他形式。Headers类型的Exchange就是一种忽略routingKey的路由方式。他通过Headers来进行消息路由。这个headers是一个键值对，发送者可以在发送的时候定义一些键值对，接受者也可以在绑定时定义自己的键值对。当键值对匹配时，对应的消费者就能接收到消息。匹配的方式有两种，一种是all，表示需要所有的键值对都满足才行。另一种是any，表示只要满足其中一个键值就可以。

  • 在Consumer端，声明Queue与Exchange绑定关系时，可以增加声明headers，表明自己对哪些信息感兴趣。

    	Map<String, Object> headers = new HashMap<String, Object>();headers.put("x-match","any"); //x-match:特定的参数。all表示必须全部匹配才算成功。any表示只要匹配一个就算成功。headers.put("loglevel", "info");headers.put("buslevel", "product");headers.put("syslevel", "admin");
​Connection connection = RabbitMQUtil.getConnection();Channel channel = connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.HEADERS);String queueName = channel.queueDeclare("ReceiverHeader",true,false,false,null).getQueue();  channel.queueBind(queueName, EXCHANGE_NAME,routingKey,headers);

  • 在Producer端，发送消息时，带上消息的headers相关属性。

    public class EmitLogHeader {
​private static final String EXCHANGE_NAME = "logs";/*** exchange有四种类型， fanout topic headers direct* headers用得比较少，他是根据头信息来判断转发路由规则。头信息可以理解为一个Map* @param args* @throws Exception*/public static void main(String[] args) throws Exception{// header模式不需要routingKey来转发，他是根据header里的信息来转发的。比如消费者可以只订阅logLevel=info的消息。// 然而，消息发送的API还是需要一个routingKey。 // 如果使用header模式来转发消息，routingKey可以用来存放其他的业务消息，客户端接收时依然能接收到这个routingKey消息。String routingKey = "ourTestRoutingKey";// The map for the headers.Map<String, Object> headers = new HashMap<>();headers.put("loglevel", "error");headers.put("buslevel", "product");headers.put("syslevel", "admin");String message = "LOG INFO asdfasdf";Connection connection = RabbitMQUtil.getConnection();Channel channel = connection.createChannel();//发送者只管往exchange里发消息，而不用关心具体发到哪些queue里。channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.HEADERS);AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder(); builder.deliveryMode(MessageProperties.PERSISTENT_TEXT_PLAIN.getDeliveryMode());builder.priority(MessageProperties.PERSISTENT_TEXT_PLAIN.getPriority());builder.headers(headers);
​channel.basicPublish(EXCHANGE_NAME, routingKey, builder.build(), message.getBytes("UTF-8"));
​channel.close();connection.close();}
}

  • Headers交换机的性能相对比较低，因此官方并不建议大规模使用这种交换机，也没有把他列入基础的示例当中。

3、SpringBoot集成RabbitMQ

3.1 引入依赖

  • SpringBoot官方集成了RabbitMQ，只需要快速引入依赖包即可使用。RabbitMQ与SpringBoot集成的核心maven依赖就下面一个：

    <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

3.2 配置关键参数

  • 基础的运行环境参数以及生产者的一些默认属性配置都集中到了application.yml配置文件中。所有配置项都以spring.rabbitmq开头。

    spring:rabbitmq:host: 192.168.137.120port: 5672username: adminpassword: 123456virtual-host: /connection-timeout: 600000

3.3 声明Exchange、Queue和Binding

  • 所有的exchange, queue, binding的配置，都需要以对象的方式声明。默认情况下，这些业务对象一经声明，应用就会自动到RabbitMQ上常见对应的业务对象。但是也是可以配置成绑定已有业务对象的。

3.4 使用RabbitmqTemplate对象发送消息

  • 生产者的所有属性都已经在application.properties配置文件中进行配置。项目启动时，就会在Spring容器中初始化一个RabbitmqTemplate对象，然后所有的发送消息操作都通过这个对象来进行。

3.5 使用@RabbitListener注解声明消费者

  • 消费者都是通过@RabbitListener注解来声明。在@RabbitMQListener注解中包含了非常多对Queue进行定制的属性，大部分的属性都是有默认值的。例如ackMode默认是null，就表示自动应答。在日常开发过程中，通常都会简化业务模型，让消费者只要绑定队列消费即可。
  • 使用SpringBoot框架集成RabbitMQ后，开发过程可以得到很大的简化，所以使用过程并不难，对照一下示例就能很快上手。但是，需要理解一下的是，SpringBoot集成后的RabbitMQ中的很多概念，虽然都能跟原生API对应上，但是这些模型中间都是做了转换的，比如Message，就不是原生RabbitMQ中的消息了。使用SpringBoot框架，尤其需要加深对RabbitMQ原生API的理解，这样才能以不变应万变，深入理解各种看起来简单，但是其实坑很多的各种对象声明方式。

http://www.ppmy.cn/news/1540720.html

相关文章

设备ESD防静电监控仪助力自动化产线设备稳定运行

设备ESD防静电监控仪助力自动化产线设备稳定运行

在现代自动化生产线中&#xff0c;设备的稳定运行至关重要。而静电放电&#xff08;ESD&#xff09;常常是导致这些问题的隐患之一。许多企业在生产管理中面临着设备接地状况不明、漏电检测困难、人工点检耗时巨大的难题。这些问题不仅影响生产效率&#xff0c;更可能造成潜在的…
阅读更多...
华为OD机试2024年真题（基站维修工程师）

华为OD机试2024年真题（基站维修工程师）

基站维修工程师&#xff08;200分&#xff09; 小王是一名基站维护工程师&#xff0c;负责某区域的基站维护。 某地方有n个基站(1<n<10)&#xff0c;已知各基站之间的距离s(0<s<500)&#xff0c;并且基站x到基站y的距离&#xff0c;与基站y到基站x的距离并不一定会…
阅读更多...
Git推送被拒

Git推送被拒

今天开发完成一个新的需求&#xff0c;将自己的分支合并到test分支后&#xff0c;推送到远程仓库&#xff0c;结果显示推送被拒&#xff1a; 原因是因为有人更新了test分支的代码&#xff0c;我在合并之前没有拉取最新的test分支代码&#xff0c;所以他提示我“推送前需要合并…
阅读更多...
java计算机毕设课设—飞机大战游戏(附源码、文章、相关截图、部署视频)

java计算机毕设课设—飞机大战游戏(附源码、文章、相关截图、部署视频)

这是什么系统&#xff1f; 资源获取方式再最下方 java计算机毕设课设—飞机大战游戏(附源码、文章、相关截图、部署视频) 基于Java的飞机大战游戏是一款经典的射击类游戏&#xff0c;主要包含我方飞机、敌方飞机、子弹、特殊NPC、开始背景、结束背景以及背景音乐等元素。我方…
阅读更多...
LabVIEW提高开发效率技巧----VI继承与重载

LabVIEW提高开发效率技巧----VI继承与重载

在LabVIEW开发中&#xff0c;继承和重载是面向对象编程&#xff08;OOP&#xff09;中的重要概念。通过合理运用继承与重载&#xff0c;不仅能提高代码的复用性和灵活性&#xff0c;还能减少开发时间和维护成本。下面从多个角度介绍如何在LabVIEW中使用继承和重载&#xff0c;并…
阅读更多...
analysis-ik分词器

analysis-ik分词器

analysis-ik分词器 1、安装离线在线 2、使用配置拓展词典 3、测试ik_smartik_max_word 1、安装 离线 使用离线安装下载地址https://release.infinilabs.com/analysis-ik/stable/找到对应es版本的ik分词器、下载zip后放到/elasticsearch/plugins/ik文件夹下。重启es即可生效 …
阅读更多...
使用Python-pptx轻松批量添加水印

使用Python-pptx轻松批量添加水印

哈喽，大家好，我是木头左！ 本文将详细介绍如何使用Python-pptx库批量添加文字或图片水印到每张幻灯片上。 安装Python-pptx库 确保你已经安装了Python-pptx库。如果没有，可以使用以下命令进行安装： pip install python-pptx创建一个简单的PPT文件 在开始之前，需要创建…
阅读更多...
leetcode day1

leetcode day1

最小差值 给你一个整数数组 nums&#xff0c;和一个整数 k 。 在一个操作中&#xff0c;您可以选择 0 < i < nums.length 的任何索引 i 。将 nums[i] 改为 nums[i] x &#xff0c;其中 x 是一个范围为 [-k, k] 的任意整数。对于每个索引 i &#xff0c;最多 只能 应用…
阅读更多...
最新文章

Copyright @ 2022~2023