RabbitMQ之常用API

忘是亡心i 2023-06-16 12:56 7阅读 0赞

#### 一.消费者确认机制 ####

##### 一.消费者确认 #####

消费者确认或者说消费者应答指的是RabbitMQ需要确认消息到底有没有被收到。

###### RabbitMQ中的两种确认方式： ######

*  自动确认方式：RabbitMQ成功将消息发出（即将消息成功写入TCP Socket）中立即认为本次投递已经被正确处理，不管消费者端是否成功处理本次投递
 *  手动处理方式：消费者收到消息后，手动调用basic.ack/basic.nack/basic.reject后，RabbitMQ收到这些消息后，才认为本次投递成功

###### 1. 自动应答 ######

boolean autoAck = true;
    channel.basicConsume(QUEUE_NAME, autoAck, consumer);

###### 2. 手动应答 ######

boolean autoAck = false;
    channel.basicConsume(QUEUE_NAME, autoAck, consumer);

#### 二.消息消费 ####

(1) 单一消费

> channel.basicAck(envelope.getDeliveryTag(), false)

(2) 批量消费

> channel.basicAck(envelope.getDeliveryTag(), true)

*  deliveryTag:该消息的index
 *  multiple：是否批量.true:将一次性ack所有小于deliveryTag的消息。

(3) basicNack

> channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);

*  deliveryTag:该消息的index
 *  multiple：是否批量.true:将一次性拒绝所有小于deliveryTag的消息。
 *  requeue：被拒绝的是否重新入队列

(4) 拒绝单条消息 basicReject

> void basicReject(long deliveryTag, boolean requeue) throws IOException;

*  deliveryTag:发布的每一条消息都会获得一个唯一的deliveryTag，它在channel范围内是唯一的
 *  requeue：表示如何处理这条消息，为true表示重新放入RabbitMQ的发送队列中，为false表示通知RabbitMQ销毁该消息

channel.basicNack 与 channel.basicReject 的区别在于basicNack可以拒绝多条消息，而basicReject一次只能拒绝一条消息

(5) 拒绝多条消息 basicReject

> void basicNack(long deliveryTag, boolean multiple, boolean requeue) throws IOException;

*  deliveryTag:发布的每一条消息都会获得一个唯一的deliveryTag，它在channel范围内是唯一的
 *  multiple：批量确认标志，为true表示包含当前消息在内的所有比该消息的deliveryTag值小的消息都被拒绝， 除了已经被应答的消息。为false则表示只拒绝本条消息
 *  requeue：表示如何处理这条消息，为true表示重新放入RabbitMQ的发送队列中，为false表示通知RabbitMQ销毁该消息

需要注意：如果队列中只有一个消费者的时候，需要确认不会因为拒绝消息并重新放入消息队列中而导致在同一个消费者身上发生死循环。

(6) 重新投递 basicRecover

> Basic.RecoverOk basicRecover(boolean requeue);

重新投递并没有所谓的像basicReject中的basicReject的deliveryTag参数，所以重新投递好像是将消费者还没有处理的所有的消息都重新放入到队列中，而不是将某一条消息放入到队列中，与basicReject不同的是，重新投递可以指定投递的消息是否允许当前消费者消费。

(7) 是否重复投递

> envelope.isRedeliver()

*  true :重复投递
 *  false: 首次投递

#### 三.队列Queue ####

(1) 定义i队列 queueDeclare

> channel.queueDeclare(QUEUE\_NAME, false, false, false, null);

*  durable：true、false true：在服务器重启时，能够存活
 *  exclusive ：是否为当前连接的专用队列，在连接断开后，会自动删除该队列，生产环境中应该很少用到吧。
 *  autodelete：当没有任何消费者使用时，自动删除该队列。this means that the queue will be deleted when there are no more processes consuming messages from it.

#### 四.通道Channel相关 ####

(1) 绑定队列

> Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,  
> Map<String, Object> arguments) throws IOException;

channel.queueDeclare("Queue", true, false, false, null);

(2) 删除队列

> Queue.DeleteOk queueDelete(String queue) throws IOException;

(3) 删除队列

> Queue.DeleteOk queueDelete(String queue, boolean ifUnused, boolean ifEmpty) throws IOException;

*  ifUnused 如果为true,则只有在通道没有在使用才被删除。
 *  ifEmpty 如果为true,则只有在通道为空才被删除。

(5) 清空队列

> Queue.PurgeOk queuePurge(String queue) throws IOException;

(6) 发布消息

> void basicPublish(String exchange, String routingKey, BasicProperties props, byte\[\] body) throws IOException;

*  exchange 交换机，可为null
 *  routingKey 路由密钥,如队列名称
 *  props 属性 ，可为null
 *  body 消息主体

如下：

channel.queueDeclare(queueName, true, false, false, null);

(7) 清空队列

> Queue.PurgeOk queuePurge(String queue) throws IOException;

(14) basicCancel

同时对于Consumer来说重写handleDelivery方法也是十分方便的。更复杂的Consumer会重写（override）更多的方法，比如handleShutdownSignal当channels和connections close的时候会调用，handleConsumeOk在其他callback方法之前调用，返回consumer tags.

Consumer同样可以override handleCancelOk和handleCancel方法，这样在显示的或者隐式的取消的时候调用。

你可以通过Channel.basicCancel方法显示的cancel一个指定的Consumer

channel.basicCancel(consumerTag);

(译者注：这句代码首先触发handleConsumerOk，之后触发handleDelivery方法，最后触发handleCancelOk方法。)

单个Consumer在Connection上都分配单个的线程来调用这些callback的方法，也就是说Consumer这里安全的调用阻塞式的方法，比如queueDeclare, txCommit, basicCancel或者basicPublish。

##### 二.发送消息确认 #####

##### (1) 单次确认 #####

waitForConfirms

Channel channel = connection.createChannel();
            channel.queueDeclare(queueName, true, false, false, null);
            //将当前信道设置成了confirm模式
            channel.confirmSelect();
    
            channel.basicPublish("", queueName, null, message.getBytes());
    
            try {
                //等待broker服务端返回ack或者nack消息
                boolean success = channel.waitForConfirms();
                Log.d(TAG, "队列-" + queueName + "-发送消息:" + message+"  状态："+success);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

可以看到上面生产者通过Confirm.Select将当前Channel信道设置成confirm模式，broker代理服务器收到之后回传Confirm.Select-Ok同一将当前Channel设置成confirm模式，此外看到返回5条Basic.Ack消息；

##### (2) 批量确认 #####

addConfirmListener

//开启确认模式
    channel.confirmSelect();
    channel.addConfirmListener(new ConfirmListener() {
                //消息失败处理
                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    //deliveryTag；唯一消息标签
                    //multiple：是否批量
                    System.err.println("-------no ack!-----------");
                }
                //消息成功处理
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    System.err.println("-------ack!-----------");
                }
            });

[rabbitmq中文参考文档][rabbitmq]

[rabbitmq]: http://rabbitmq.mr-ping.com/ClientDocumentation/java-api-guide.html