RabbitMQ消息发送与消费（三）

分手后的思念是犯贱 2022-09-01 13:58 232阅读 0赞

## 发送消息 ##

调用Channel的basicPublish方法来发送消息，basicPublish有多个构造方法，其他构造方法都是由下面构造方法缺省某些参数构成。

void basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, BasicProperties props, byte[] body)
                throws IOException;

方法参数详细说明如下：

exchange：交换器名称。指定消息需要发送到那个交换器，如果为空字符串则会发送到RabbitMQ默认交换器。

routingKey：路由键。决定消息发送到当前交换器的那个队列。

mandatory：路由失败处理方式。true：交换器根据自身类型和路由键无法找到一个合适的队列时RabbitMQ会调用Basic.Return命令将消息返回给生产者。false：直接丢弃。

immediate：true：如果消息路由到的队列上没有一个消费者，那么消息不会存入队列。如果消息路由到的所有队列都没有消费者时，该消息会返回给生产者。

props：消息的基本属性。包含14中：

contentType：消息内容的类型。例如：text/plain、application/json等。

contentEncoding：消息内容编码格式。

headers：消息的headers，Map类型。

deliveryMode：是否是持久化消息。1：非持久化；2：持久化。

priority：消息优先级。

correlationId：关联ID。

replyTo：用户指定回复的队列名称。

expiration：过期时间（ms）。

messageId：消息ID。

timestamp：时间戳。

type：类型。

userId：用户ID。

appId：应用程序ID。

clusterId：集群ID。

body：消息。

使用mandatory需要给Channel添加返回监听，示例：

channel.queueBind(queueName, exchangeName, "topic.#");
            channel.basicPublish(exchangeName, "111.test", true, false, null, "你好，Topic！".getBytes());
            channel.addReturnListener((replyCode, replyText, exchange, routingKey, properties, body) -> {
                String message = new String(body);
                System.out.println("返回结果：" + message);
            });

## 消费消息 ##

RabbitMQ的消费模式分为两种：推模式（push）和拉模式（pull）。推模式采用Basic.Consume进行消费，拉模式采用Basic.Get进行消费。

### 推模式 ###

接收消息一般通过实现Consumer接口或者集成DefaultConsumer类来实现。

channel.basicConsume(queueName, false, "myConsumerTag", new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String routingKey = envelope.getRoutingKey();
                    String exchange = envelope.getExchange();
                    System.out.println("exchange：" + exchange);
                    System.out.println("routingKey：" + routingKey);
                    System.out.println(new String(body));
                    long deliveryTag = envelope.getDeliveryTag();
                    channel.basicAck(deliveryTag, true);
                }
            });

basicConsume方法的重载方法有很多，这里列举参数最全的来说明。

String basicConsume(String queue, boolean autoAck, String consumerTag, boolean noLocal, boolean exclusive, Map<String, Object> arguments, Consumer callback) throws IOException;

参数说明：

queue：队列名称。

autoAck：是否自动确认。如果设置为false，需要调用channel.basicAck来确认消息。

consumerTag：消费者标签。

noLocal：true：不能将同一个Connection中生者发送的消息传送到这个Connection中的消费者。

exclusive：是否排他。

argments：消费者的其他参数。

callback：消费者回调函数。用来出来RabbitMQ推送过来的消息。

### 拉模式 ###

通过channel.basicGet可以获取单条信息，返回值为GetResponse。当前方法没有别的重载方法。

GetResponse getResponse = channel.basicGet(queueName, false);
            System.out.println(new String(getResponse.getBody()));
            channel.basicAck(getResponse.getEnvelope().getDeliveryTag(), false);

区别：

推模式将信道（Channel）设置为投递模式，知道取消队列订阅为止。在投递模式期间，RabbitMQ会不断的将消息推送给消费者，但是还是会受到basicQos的影响（basicQos为消费者没有确认消息前，RabbitMQ推送给消费者的消息条数，例如：basicQos(1)表示服务器推送给客户端一条消息，在消费者还没确认之前不会继续给消费者推送。） 。如果只想从队列获取单条消息而不是持续订阅可以使用basicGet进行消费。

### 消息确认与拒绝 ###

为了保证消息从队列可靠的到达消费者，RabbitMQ提供消息确认机制。消费者订阅队列的时候将autoAck设置为false时，RabbitMQ会等待消费者显示地回复确认信号之后才会将消息从内存或者磁盘中移除。当autoAck设置为true时，RabbitMQ会自动把发出设置为确认，然后从内存或磁盘中删除，而不管消费者是否真正消费到这些消息。

消息确认方法：

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

deliveryTag：消息编号。

multiple：true：批量确认，确认编号小于当前编号的所有消息。false：单条确认。

消息拒绝方法：

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

basicNack可以批量拒绝，basicReject只能单挑拒绝。

deliveryTag：消息编号。

multiple：true：批量拒绝，拒绝当前编号之前的所有消息。false：单条拒绝。

requeue：设置消息是否重新进入队列。true：重新进入队列，false：消息从消息队列删除。