RabbitMQ系列【16】AmqpTemplate接口详解

迷南。 2023-10-09 15:10 33阅读 0赞

> **有道无术，术尚可求，有术无道，止于术。**

#### 文章目录 ####

*   *  前言
     *  AmqpTemplate
     *  API
     *   *  send
         *  convertAndSend
         *  receive
         *  receiveAndConvert
         *  receiveAndReply
         *  sendAndReceive
         *  convertSendAndReceive

### 前言 ###

`RabbitTemplate` 是`spring-amqp`提供的一个`RabbitMQ` 消息操作模板类，在之前我们使用它完成了简单的消息发送。

`RabbitTemplate`主要提供了**发送消息、接收消息**以及其他附加功能，内部封装了`RabbitMQ`原生`API`，大大简化了使用`RabbitMQ`操作。

`RabbitTemplate` 主要实现了`AmqpTemplate`和`RabbitOperations`接口：  
![在这里插入图片描述][f3b689646ef14fde9176f7ccc9771843.png]

### AmqpTemplate ###

`AmqpTemplate`接口主要声明了三类方法：

public interface AmqpTemplate {
        
    	// 发送消息
        void send(Message var1) throws AmqpException;
    
        // 接收消息
        Message receive() throws AmqpException;
    	
        // 发送消息并接收回复
        Message sendAndReceive(Message var1) throws AmqpException;
    }

### API ###

首先，我们看下`AmqpTemplate`中声明的各种方法。

#### send ####

`send`方法一共有三个，需要创建`Message`消息对象，将消息封装到该对象内发送，如果没有指定交换机、路由键，将使用默认值，也就是空字符串。

// 发送消息到默认交换机、默认路由KEY
    	void send(Message message) throws AmqpException;
    
    	// 发送消息到默认交换机、使用指定路由KEY
    	void send(String routingKey, Message message) throws AmqpException;
    
    	// 发送消息到指定交换机、使用指定路由KEY
    	void send(String exchange, String routingKey, Message message) throws AmqpException;

示例：

Message message = new Message("消息".getBytes());
            rabbitTemplate.send(message);
            rabbitTemplate.send("route.key", message);
            rabbitTemplate.send("exchange_name", "route.key", message);

#### convertAndSend ####

`convertAndSend` 方法可以转换对象并发送，并可以添加一个消息处理器`MessagePostProcessor`。

// 将Java对象转换为Amqp｛@link Message｝并将其发送到默认交换机、使用默认路由KEY
    	void convertAndSend(Object message) throws AmqpException;
    
    	// 将Java对象转换为Amqp｛@link Message｝并将其发送到默认交换机、使用自定义路由KEY
    	void convertAndSend(String routingKey, Object message) throws AmqpException;
    
    	// 将Java对象转换为Amqp｛@link Message｝并将其发送到自定义交换机、使用自定义路由KEY
    	void convertAndSend(String exchange, String routingKey, Object message) throws AmqpException;
    
    	 // 将Java对象转换为Amqp｛@link Message｝并将其发送到自定义交换机、使用自定义路由KEY
    	 // 在发送消息之前添加一个消息处理器MessagePostProcessor 
    	void convertAndSend(Object message, MessagePostProcessor messagePostProcessor) throws AmqpException;
    
    	// 将Java对象转换为Amqp｛@link Message｝并将其发送到默认交换机、使用自定义路由KEY
    	// 在发送消息之前添加一个消息处理器MessagePostProcessor 
    	void convertAndSend(String routingKey, Object message, MessagePostProcessor messagePostProcessor)
    			throws AmqpException;
    
    	// 将Java对象转换为Amqp｛@link Message｝并将其发送到自定义交换机、使用自定义路由KEY
    	// 在发送消息之前添加一个消息处理器MessagePostProcessor
    	void convertAndSend(String exchange, String routingKey, Object message, MessagePostProcessor messagePostProcessor)
    			throws AmqpException;

`MessagePostProcessor`是一个函数型接口，提供了一个`postProcessMessage`方法处理消息，由于直接发送的是对象，如果需要设置一些消息的属性，就需要使用该接口进行设置，例如：

MessagePostProcessor messagePostProcessor = message1 -> {
        
                MessageProperties messageProperties = message1.getMessageProperties();
                messageProperties.setExpiration("1000");
                return message1;
            };
            rabbitTemplate.convertAndSend("","","消息",messagePostProcessor);

#### receive ####

一般获取消息有两种处理模式：

*  **push**：由`RabbitMQ`主动将消息推送给订阅队列的消费者，调用`channel.basicConsume`方法。
 *  **pull**：主动从指定队列中拉取消息，需要消费者调用`channel.basicGet`方法。

`receive` 方法，就是从主动队列中获取消息。

// 如果默认队列中有消息，则接收消息。立即返回，可能有NULL值
    	@Nullable
    	Message receive() throws AmqpException;
    
    	// 从指定队列中获取消息。立即返回，可能有NULL值
    	@Nullable
    	Message receive(String queueName) throws AmqpException;
    
    	// 如果默认队列中有消息，则接收消息。可能有NULL值，并指定一个超时时间
    	@Nullable
    	Message receive(long timeoutMillis) throws AmqpException;
    
    	// 从指定队列中获取消息。可能有NULL值，并指定一个超时时间
    	@Nullable
    	Message receive(String queueName, long timeoutMillis) throws AmqpException;

#### receiveAndConvert ####

`receiveAndConvert`可以拉取消息并进行对象转换。

// 如果默认队列中有消息，则接收消息并将其转换为Java对象。立即返回，可能为null 值。
    	@Nullable
    	Object receiveAndConvert() throws AmqpException;
    
    	// 从指定队列中接收消息并将其转换为Java对象。立即返回，可能为null 值。
    	@Nullable
    	Object receiveAndConvert(String queueName) throws AmqpException;
    
    	// 如果默认队列中有消息，则接收消息并将其转换为Java对象。立即返回，可能有NULL值，并指定一个超时时间
    	@Nullable
    	Object receiveAndConvert(long timeoutMillis) throws AmqpException;
    
    	// 从指定队列中接收消息并将其转换为Java对象，并指定一个超时时间，可能为null 值。
    	@Nullable
    	Object receiveAndConvert(String queueName, long timeoutMillis) throws AmqpException;
    
    	//  如果默认队列中有消息，则接收消息并将其转换为Java对象。立即返回，可能为null 值。并可以添加一个消息转换器SmartMessageConverter。
    	@Nullable
    	<T> T receiveAndConvert(ParameterizedTypeReference<T> type) throws AmqpException;
    
    	// 从指定队列中接收消息并将其转换为Java对象。立即返回，可能为null 值。并可以添加一个消息转换器SmartMessageConverter。
    	@Nullable
    	<T> T receiveAndConvert(String queueName, ParameterizedTypeReference<T> type) throws AmqpException;
    
    	// 如果默认队列中有消息，则接收消息并将其转换为Java对象。可能有NULL值，并指定一个超时时间及消息转换器SmartMessageConverter。
    	@Nullable
    	<T> T receiveAndConvert(long timeoutMillis, ParameterizedTypeReference<T> type) throws AmqpException;
    
    	// 从指定队列中接收消息并将其转换为Java对象。可能为null 值。并指定一个超时时间及消息转换器SmartMessageConverter。
    	@Nullable
    	<T> T receiveAndConvert(String queueName, long timeoutMillis, ParameterizedTypeReference<T> type)
    			throws AmqpException;

示例：

// 接收消息，队列存在是，会报错；队列中没有消息，返回NULL
            Message bootQueue1= rabbitTemplate.receive("bizQueue");
            Message bootQueue2 = rabbitTemplate.receive("bizQueue",1000);
            Message bootQueue3= rabbitTemplate.receive("backupQueue");

使用消息转换器，可以直接发送、接收对象：

// User 需要实现Serializable
            User user = new User();
            user.setName("张三");
            rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
            rabbitTemplate.convertAndSend(MqBizConfig.BIZ_EXCHANGE,MqBizConfig.BIZ_ROUTE_KEY,user);
            User receiveUser = rabbitTemplate.receiveAndConvert("bizQueue", new ParameterizedTypeReference<User>() {
        });

#### receiveAndReply ####

`receiveAndReply`支持在获取消息时传入一个回调函数`ReceiveAndReplyCallback`，处理接收到消息和回复消息的业务逻辑。

`receiveAndReply`应用于`RPC`模式`Server`端，`Server`收到消息，并回复消息给客户端：  
![在这里插入图片描述][8da3a23100344325b465fda68fd9ef91.png]  
该模式用的比较少，实现起来也比较麻烦，这里就不演示了。

// 收到消息并回复，R：接收到的消息 S: 返回的消息
    	<R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback) throws AmqpException;
    
    	<R, S> boolean receiveAndReply(String queueName, ReceiveAndReplyCallback<R, S> callback) throws AmqpException;
    
    	<R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback, String replyExchange, String replyRoutingKey)
    			throws AmqpException;
    
    	<R, S> boolean receiveAndReply(String queueName, ReceiveAndReplyCallback<R, S> callback, String replyExchange,
    			String replyRoutingKey) throws AmqpException;
    
    	<R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback,
    			ReplyToAddressCallback<S> replyToAddressCallback) throws AmqpException;
    
    	<R, S> boolean receiveAndReply(String queueName, ReceiveAndReplyCallback<R, S> callback,
    			ReplyToAddressCallback<S> replyToAddressCallback) throws AmqpException;

#### sendAndReceive ####

`sendAndReceive`也属于`RPC`模式，发送消息并接收回复消息，属于`Client`端。

@Nullable
    	Message sendAndReceive(Message message) throws AmqpException;
    
    	@Nullable
    	Message sendAndReceive(String routingKey, Message message) throws AmqpException;
    
    	@Nullable
    	Message sendAndReceive(String exchange, String routingKey, Message message) throws AmqpException;

#### convertSendAndReceive ####

`convertSendAndReceive`以及`convertSendAndReceiveAsType`是对`sendAndReceive`的扩展，可以直接发送对象消息，并可以设置类型转换器：

@Nullable
    	Object convertSendAndReceive(Object message) throws AmqpException;
    
    	@Nullable
    	Object convertSendAndReceive(String routingKey, Object message) throws AmqpException;
    
    	@Nullable
    	Object convertSendAndReceive(String exchange, String routingKey, Object message) throws AmqpException;
    
    	@Nullable
    	Object convertSendAndReceive(Object message, MessagePostProcessor messagePostProcessor) throws AmqpException;
    
    	@Nullable
    	Object convertSendAndReceive(String routingKey, Object message, MessagePostProcessor messagePostProcessor)
    			throws AmqpException;
    
    	@Nullable
    	Object convertSendAndReceive(String exchange, String routingKey, Object message,
    			MessagePostProcessor messagePostProcessor) throws AmqpException;
    
    	@Nullable
    	<T> T convertSendAndReceiveAsType(Object message, ParameterizedTypeReference<T> responseType)
    			throws AmqpException;
    
    	@Nullable
    	<T> T convertSendAndReceiveAsType(String routingKey, Object message,
    			ParameterizedTypeReference<T> responseType) throws AmqpException;
    
    	@Nullable
    	<T> T convertSendAndReceiveAsType(String exchange, String routingKey, Object message,
    			ParameterizedTypeReference<T> responseType) throws AmqpException;
    
    	@Nullable
    	<T> T convertSendAndReceiveAsType(Object message, MessagePostProcessor messagePostProcessor,
    			ParameterizedTypeReference<T> responseType) throws AmqpException;
    
    	@Nullable
    	<T> T convertSendAndReceiveAsType(String routingKey, Object message,
    			MessagePostProcessor messagePostProcessor, ParameterizedTypeReference<T> responseType)
    			throws AmqpException;
    
    	@Nullable
    	<T> T convertSendAndReceiveAsType(String exchange, String routingKey, Object message,
    			MessagePostProcessor messagePostProcessor, ParameterizedTypeReference<T> responseType)
    			throws AmqpException;

[f3b689646ef14fde9176f7ccc9771843.png]: https://img-blog.csdnimg.cn/f3b689646ef14fde9176f7ccc9771843.png
[8da3a23100344325b465fda68fd9ef91.png]: https://img-blog.csdnimg.cn/8da3a23100344325b465fda68fd9ef91.png