Spring AMQP使用整理

RabbitAdmin功能

RabbitAdmin类用来管理RabbitMQ；

创建方法：

ConnectionFactory connectionFactory = new CachingConnectionFactory();

RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);

　　

  • declareExchange：创建交换机
  • deleteExchange：删除交换机
  • declareQueue：创建队列
  • deleteQueue：删除队列
  • purgeQueue：清空队列
  • declareBinding：新建绑定关系
  • removeBinding：删除绑定关系
  • getQueueProperties：查询队列属性

CachingConnectionFactory是Spring AMQP下一个连接工厂，适合SpringBoot的深度整合的连接工厂；

从构造方法出发

org.springframework.amqp.rabbit.connection.CachingConnectionFactory#CachingConnectionFactory()

最终会调用这个有参构造方法

org.springframework.amqp.rabbit.connection.CachingConnectionFactory#CachingConnectionFactory(java.lang.String, int)

CachingConnectionFactory最终会调用RabbitMQ的原生API，对RabbitMQ Client包下的ConnectionFactory的包装；

对于下面的这个写法，这个ConnectionFactory是Spring AMQP包下的，它继承了AbstractConnectionFactory，而AbstractConnectionFactory实现了Spring AMQP包下的ConnectionFactory；如下图；

ConnectionFactory connectionFactory = new CachingConnectionFactory();

org.springframework.amqp.core.Exchange实现类，每种实现对应一种交换机类型，如下图；

org.springframework.amqp.core.Queue#Queue(java.lang.String) 对应的参数，如下图；

org.springframework.amqp.core.Binding#Binding 对应的参数，如下图；

org.springframework.amqp.rabbit.core.RabbitAdmin#declareExchange RabbitAdmin使用该方法将交换机进行绑定，使用了RabbitTemplate对RabbitMQ Client包进行了封装；

最终在org.springframework.amqp.rabbit.core.RabbitAdmin#declareExchanges会有调用RabbitMQ Client的原生API，如下图；

测试代码

@Slf4j
@Configuration
public class RabbitConfig {

	@PostConstruct
	public void initRabbit() {
		CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
		connectionFactory.setHost("192.168.211.135");
		connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/dev");
		connectionFactory.setUsername("admin");
		connectionFactory.setPassword("password");

		RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);

		Exchange exchange = new TopicExchange("exchange.order");
		// 创建交换机
		rabbitAdmin.declareExchange(exchange);

		Queue queue = new Queue("queue.order");
		// 创建队列
		rabbitAdmin.declareQueue(queue);

		Binding binding = new Binding(
				"queue.order",
				Binding.DestinationType.QUEUE,
				"exchange.order",
				"key.order",
				null
		);
		// 新建绑定关系
		rabbitAdmin.declareBinding(binding);
	}
}

　　

服务重启后，在RabbitMQ的管控台会出现新建的队列，交换机；

RabbitAdmin声明式配置

  • 将Exchange，Queue，Binding声明为Bean；
  • 再将RabbitAdmin声明为Bean；
  • Exchange，Queue，Binding即可自动创建；

代码如下：

@Slf4j
@Configuration
public class RabbitConfig {

	@Bean
	public ConnectionFactory connectionFactory() {
		CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
		connectionFactory.setHost("192.168.211.135");
		connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/dev");
		connectionFactory.setUsername("admin");
		connectionFactory.setPassword("password");
//		connectionFactory.createConnection();
		return connectionFactory;
	}

	@Bean
	public Exchange exchange() {
		Exchange exchange = new TopicExchange("exchange.order");
		return exchange;
	}

	@Bean
	public Queue queue() {
		Queue queue = new Queue("queue.order");
		return queue;
	}

	@Bean
	public Binding binding() {
		Binding binding = new Binding(
				"queue.order",
				Binding.DestinationType.QUEUE,
				"exchange.order",
				"key.order",
				null
		);
		return binding;
	}

	@Bean
	public RabbitAdmin rabbitAdmin(@Autowired ConnectionFactory connectionFactory) {
		RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
		// 自动创建打开
		rabbitAdmin.setAutoStartup(true);
		return rabbitAdmin;
	}

}

　　不过@Bean这种注入方式有点懒加载的意思，就是当你注入的Bean没有被使用时，它是不会被加载的；上面配置的Bean如果没有在其他地方被使用，这些Bean是不会被加载；

org.springframework.amqp.rabbit.core.RabbitAdmin类中实现了ApplicationContextAware，InitializingBean接口；之前讲过ApplicationContextAware是可以获取Spring容器的上下文，从而可以注入别的组件注入到该对象；InitializingBean是Bean创建后进行初始化的操作，此时是Bean对象已经创建；

在RabbitAdmin实现的InitializingBean接口的afterPropertiesSet方法，有如下一个操作；

org.springframework.amqp.rabbit.connection.CachingConnectionFactory#addConnectionListener

在RabbitMQ进行连接的时候会回调添加的listener；

在RabbitAdmin实现的afterPropertiesSet方法，最终会调用一个initialize方法；

org.springframework.amqp.rabbit.core.RabbitAdmin#initialize

 通过applicationContext获取到所有的Exchange，Queue，Binding类型的Bean；

最终会创建相应的 Exchange，Queue，Binding；

当ConnectionFactory执行createConnection，相应的 Exchange，Queue，Binding通过Spring容器被创建；

connectionFactory.createConnection();

RabbitTemplate

  • RabbitTemplate与RestTemplate类似，使用了模板方法设计模式；
  • RabbitTemplate提供了丰富的功能，方便消息收发；
  • RabbitTemplate可以显式传入配置也可以隐式声明配置；

显式传入配置，使用无参构造方法是不能拿到连接工厂的；

显式传入配置时，应选用带有参数的构造方法；

显式配置如下：

@Bean
public ConnectionFactory connectionFactory() {
	CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
	connectionFactory.setHost("192.168.211.135");
	connectionFactory.setPort(5672);
	connectionFactory.setVirtualHost("/dev");
	connectionFactory.setUsername("admin");
	connectionFactory.setPassword("password");
	connectionFactory.createConnection();
	return connectionFactory;
}

@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
	RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
	return rabbitTemplate;
}

　　

声明Exchange，Queue，Binding；

@Bean
public Exchange exchange() {
	Exchange exchange = new TopicExchange("exchange.order");
	return exchange;
}

@Bean
public Queue queue() {
	Queue queue = new Queue("queue.order");
	return queue;
}

@Bean
public Binding binding() {
	Binding binding = new Binding(
			"queue.order",
			Binding.DestinationType.QUEUE,
			"exchange.order",
			"key.order",
			null
	);
	return binding;
}

　　

RabbitTemplate中发送消息的方法有send和convertAndSend；

关于send方法的使用如下：

send方法使用需要将要发送的消息转换为字节流；

Message message = new Message("test".getBytes(), null);
rabbitTemplate.send("exchange.order", "key.order", message);

上面代码运行后会报错：

根据异常追踪到抛NPE的位置：

将上面的Message创建修改如下，消息能正常发出；

Message message = new Message("test".getBytes(), new MessageProperties());

关于convertAndSend方法使用如下：

rabbitTemplate.convertAndSend("exchange.order", "key.order", "test2");

org.springframework.amqp.rabbit.core.RabbitTemplate#send(java.lang.String, java.lang.String, org.springframework.amqp.core.Message, org.springframework.amqp.rabbit.connection.CorrelationData)

 execute方法里面传入了一个lambda对象，里面的channel指的是RabbitMQ Client的Channel对象；

org.springframework.amqp.rabbit.core.RabbitTemplate#doSend最终会调用sendToRabbit方法；

org.springframework.amqp.rabbit.core.RabbitTemplate#sendToRabbit

而在convertAndSend方法中不需要对消息内容进行字节流，因为在convertAndSend方法最终会调send方法，在send方法传入的消息内容的参数做了转换；

配置发送端确认和消息返回

confirmCallback为生产者投递消息后，如果Broker收到消息后，会给生产者一个ACK；生产者通过ACK，可以确认这条消息是否正常发送到Broker，即发送端确认机制；

returnCallback为交换机到队列不成功，返回给消息生产者，触发returnCallback，即消息返回机制；

@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
	RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
	// 换机处理消息到路由失败，则会返回给生产者
	rabbitTemplate.setMandatory(true);

	// 交换机到队列不成功，返回给消息生产者，触发returnCallback
	rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
		@Override
		public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
			log.info("message:{} replyCode:{}, replyText:{}, exchange:{}, routingKey:{}",
					message, replyCode, replyText, exchange, routingKey);
		}
	});

	// 生产者投递消息后，如果Broker收到消息后，会给生产者一个ACK。生产者通过ACK，可以确认这条消息是否正常发送到Broker
	rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
		@Override
		public void confirm(CorrelationData correlationData, boolean ack, String cause) {
			log.info("correlationData:{}, ack:{}, cause:{}", correlationData, ack, cause);
		}
	});
	return rabbitTemplate;
}

配置完后，服务重启，发送消息，confirmCallback并没有回调；

主要原因是这里用的ConnectionFactory是自定义配置注入的，不是Spring Boot配置注入的，如果需要使用Spring Boot配置注入，需要application.properties配置文件配置RabbitMQ的连接信息；

RabbitTemplate类有一个publisherConfirms的成员属性；

org.springframework.amqp.rabbit.core.RabbitTemplate#publisherConfirms，通过connectionFactory.isPublisherConfirms()控制publisherConfirms；

在ConnectionFactory配置添加setPublisherConfirms，不过该方法从Spring Boot2.2已弃用，需要使用setPublisherConfirmType方法；

setPublisherConfirms对应application.properteise中确认消息发送成功，通过实现ConfirmCallBack接口，消息发送到交换器Exchange后触发回调的配置是：

spring.rabbitmq.publisher-confirms=true

SIMPLE值发布消息成功到交换器后会触发回调方法，但是无法得知是被确认的是哪条消息；

NONE值是禁用发布确认模式，是默认值；

CORRELATED值也是发布消息成功到交换器后会触发回调方法，发送的消息使用CorrelationData进行相关联；

配置成SIMPLE

connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.SIMPLE);
connectionFactory.setPublisherReturns(true);

效果如下：

配置成CORRELATED

connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
connectionFactory.setPublisherReturns(true);

发送方修改如下：

CorrelationData correlationData = new CorrelationData();
correlationData.setId("test2_Id");
rabbitTemplate.convertAndSend("exchange.order", "key.order", "test2", correlationData);

执行如下：

SimpleMessageListenerContainer简单消息监听容器

  • 设置同时监听多个队列，自动启动，自动配置RabbitMQ
  • 设置消费者数量
  • 设置消息确认模式，是否重回队列，异常捕获
  • 设置是否独占，其他线程消费属性等
  • 设置具体的监听器，消息转换器等
  • 支持动态设置，运行中修改监听器配置

使用如下：

@Bean
public SimpleMessageListenerContainer simpleMessageListenerContainer(ConnectionFactory connectionFactory) {
	SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);

	// 设置监听的队列
	container.setQueueNames("queue.order");
	// 设置并发的消费者
	container.setConcurrentConsumers(3);
	// 设置最大并发的消费者
	container.setMaxConcurrentConsumers(3);
	// 设置确认方式
	container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
	// 设置消息监听回调
	container.setMessageListener(new ChannelAwareMessageListener() {
		@Override
		public void onMessage(Message message, Channel channel) throws Exception {
			log.info("message:{}", message);
			channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
		}
	});

	// 现在每次只允许一条消息消费
	container.setPrefetchCount(1);

	return container;
}

　　

带有Channel的MessageListener扩展接口，用于手动确认；

执行如下：

message:(Body:'test2' MessageProperties [headers={spring_listener_return_correlation=b6ba6114-27c4-4c9b-ac4d-e3053ecf8762, spring_returned_message_correlation=test2_Id}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=exchange.order, receivedRoutingKey=key.order, deliveryTag=2, consumerTag=amq.ctag-HIhcGJg7Q6OdAV-q26zbnQ, consumerQueue=queue.order])

之前在代码中设置SimpleMessageListenerContainer最大并发的消费者为3个，每次只允许消费一条消息，可以从管控台看出，配置已生效；

SimpleMessageListenerContainer类的继承树如下

org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer实现了MessageListenerContainer；

MessageListenerContainer接口的继承树如下：

MessageListenerContainer最终继承了LifeCycle的接口，LifeCycle接口是Spring框架的接口，用于定义注入Spring中的Bean启动/停止生命周期控制方法的通用接口，当Bean启动交给Spring容器管理的时候，该Bean中的start方法会被回调；

org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer#start

AbstractMessageListenerContainer#start会调到实现类的doStart方法；

org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer#doStart

调用initializeConsumers方法执行初始化消费者；initializeConsumers方法里面会调用createBlockingQueueConsumer方法创建BlockingQueueConsumer对象；

关于org.springframework.amqp.rabbit.listener.BlockingQueueConsumer对象，如下：

BlockingQueueConsumer这个类Spring AMQP包下RabbitMQ专门的消费者，封装了消息broker的连接，并有自己的生命周期；

之后回到org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer#doStart方法；

将与RabbitMQ的Channel通信的BlockingQueueConsumer对象的集合封装到AsyncMessageProcessingConsumer对象中，而AsyncMessageProcessingConsumer对象又实现了Runnable接口，getTaskExecutor方法获取到的是Spring封装的SimpleAsyncTaskExecutor类型的线程池，该线程池是来一个任务创建一个线程，比较适合短期多任务，通过getTaskExecutor方法获取的线程执行AsyncMessageProcessingConsumer对象的run方法；正常情况下run方法会执行initialize方法；

org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.AsyncMessageProcessingConsumer#initialize

最终调用BlockingQueueConsumer#start方法；start方法里面会调用setQosAndreateConsumers方法，该方法用于创建消费者和设置消费者最大能消费的消息数量；

org.springframework.amqp.rabbit.listener.BlockingQueueConsumer#setQosAndreateConsumers

最终会调用consumeFromQueue方法；

org.springframework.amqp.rabbit.listener.BlockingQueueConsumer#consumeFromQueue

该方法启动一个消费者，但是channel调用basicConsume方法传入用于处理接收RabbitMQ消息时的回调Consumer是内部的InternalConsumer对象； 

当接收到消息时， InternalConsumer会回调handleDelivery方法，将消息放入到队列中；

之后会循环执行AsyncMessageProcessingConsumer#mainLoop；

receiveAndExecute方法会调用SimpleMessageListenerContainer#receiveAndExecute，SimpleMessageListenerContainer#receiveAndExecute会调用SimpleMessageListenerContainer#doReceiveAndExecute方法，doReceiveAndExecute方法最终会调用BlockingQueueConsumer#nextMessage(long)方法，消息会从队列里取出，消息接收通过basicConsumer接收，消费的消息是从mainLoop方法中取，当接收到消息时会将消息存到本地队列，等待消息消费；

取出的消息是如何触发onMessage方法？

SimpleMessageListenerContainer#doReceiveAndExecute如果没有打开consumerBatchEnabled，则执行下面的executeListener方法，executeListener调用doExecuteListener方法，doExecuteListener会调用invokeListener方法；

而invokerListener是函数式接口，不是方法体；

invokerListenr方法的实现在proxy对象；

org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer#actualInvokeListener

该方法是先判断listener的类型，之后执行doInvokerListener方法，在doInvokerListener方法中会回调MessageListener的onMessage方法；

SimpleMessageListenerContainer调用链大致如下：

MessageListenerAdapter 消息监听适配器

使用：

  • 实现handleMessage方法
  • 自定义队列名->方法名的映射关系

使用如下：

手动注入SimpleMessageListenerContainer

@Bean
public SimpleMessageListenerContainer simpleMessageListenerContainer(ConnectionFactory connectionFactory) {
	SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);

	// 设置监听的队列
	container.setQueueNames("queue.order");
	// 设置并发的消费者
	container.setConcurrentConsumers(3);
	// 设置最大并发的消费者
	container.setMaxConcurrentConsumers(3);
	// 设置确认方式
	container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
	// 设置消息监听回调
//		container.setMessageListener(new ChannelAwareMessageListener() {
//			@Override
//			public void onMessage(Message message, Channel channel) throws Exception {
//				log.info("message:{}", message);
//				channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//			}
//		});
	MessageListenerAdapter listenerAdapter = new MessageListenerAdapter();

	// 设置消息监听回调调用orderService
	listenerAdapter.setDelegate(orderService);
	container.setMessageListener(listenerAdapter);

	// 现在每次只允许一条消息消费
	container.setPrefetchCount(1);

	return container;
}

　　

注入OrderService

public interface OrderService {
	void handleMessage(String messageBody) throws IOException;
}

　　

@Slf4j
@Service
public class OrderServiceImpl implements OrderService {
	@Override
	public void handleMessage(String messageBody) throws IOException {
		log.info("orderService handleMessage:{}", messageBody);
	}
}

当有监听到消息到达时，OrderServiceImpl#handleMessage方法会被调用；

执行如下：

为何handleMessage方法会被调用？

MessageListenerAdapter的继承树如下：

MessageListenerAdapter实现了MessageListener接口，当有消息到达时，MessageListenerAdapter#onMessage方法会被触发；

org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter#onMessage

首先先拿到delegateListener对象，判断delegateListener是否为ChannelAwareMessageListener或MessageListener类型，如果delegateListener是那些类型，delegateListener就调用onMessage方法；

delegate对象在调用doSetDelegate方法时赋值；

MessageListenerAdapter#onMessage获取到消息处理的方法名，之后调用该方法；

org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter#getListenerMethodName

 默认情况下，queueOrTagToMethodName是没有设置的，因此获取的是默认的值；

不同的队列如何设置调用不同的消息处理？

将上面的代码修改如下：

MessageListenerAdapter listenerAdapter = new MessageListenerAdapter(orderService);
HashMap<String, String> methodMap = new HashMap<>(8);
methodMap.put("queue.order1", "handleMessage1");
methodMap.put("queue.order2", "handleMessage2");
listenerAdapter.setQueueOrTagToMethodName(methodMap);
container.setMessageListener(listenerAdapter);

MessageConverter 消息转换器

RabbitMQ Client原生的消息收发是通过byte[]作为消息体的，MessageConverter用来在收发消息时自动转换消息；

Jackson2JsonMessageConverter

Jackson转换Json消息格式，最常用的MessageConverter，用来转换Json格式消息；

配合ClassMapper可以直接转换为POJO对象；

自定义MessageConverter

实现MessageConverter接口；

重写toMessage，fromMessage方法；

使用如下：

实体类OrderDTO

@Data
public class OrderDTO implements Serializable {
	private String orderId;
	private Date createDate;
	private BigDecimal totalAmount;
	private BigDecimal payAmount;
	private String payType;
}

　　

发送消息：

ObjectMapper objectMapper = new ObjectMapper();
String jsonStr = objectMapper.writeValueAsString(orderDTO);
Message message = new Message(jsonStr.getBytes(), new MessageProperties());
CorrelationData correlationData = new CorrelationData();
correlationData.setId(uuid);
rabbitTemplate.convertAndSend("exchange.order", "key.order", message, correlationData);

SimpleMessageListenerContainer配置修改：

设置转换消息的类型

MessageListenerAdapter listenerAdapter = new MessageListenerAdapter(orderService);
HashMap<String, String> methodMap = new HashMap<>(8);
methodMap.put("queue.order", "handleMessage");
listenerAdapter.setQueueOrTagToMethodName(methodMap);

Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
converter.setClassMapper(new ClassMapper() {
	@Override
	public void fromClass(Class<?> clazz, MessageProperties properties) {

	}

	/**
	 * 设置转换消息的类型
	 * @param properties
	 * @return
	 */
	@Override
	public Class<?> toClass(MessageProperties properties) {
		return OrderDTO.class;
	}
});
listenerAdapter.setMessageConverter(converter);

执行如下：

分析：

org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter#onMessage

这里获取转换后的消息；

org.springframework.amqp.rabbit.listener.adapter.AbstractAdaptableMessageListener#extractMessage

获取往MessageListenerAdapter设置的消息转换器，之后调用fromMessage方法转换消息；

上面测试代码设置的转换器是Jackson2JsonMessageConverter，最终调用AbstractJackson2MessageConverter#fromMessage(org.springframework.amqp.core.Message, java.lang.Object)；

转换器没有设置alwaysConvertToInferredType和javaTypeMapper，并且classMapper不为空，则执行下面的逻辑；

toClass调用的是上面测试代码的里面的；

convertBytesToObject方法最终是调用Jackson进行Json转换；

@RabbitListener注解

　　@RabbitListener是一个组合注解，可以嵌套以下注解：

　　@Exchange:自动声明Exchange

　　@Queue:自动声明队列

　　@QueueBinding:自动声明绑定关系 

　　要使用@RabbitListener注解，需要注入RabbitListenerContainerFactory；

@Bean
public RabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
	SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
	factory.setConnectionFactory(connectionFactory);
	factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
	return factory;
}

　　

　　消息监听处理类，如下；

　　使用 @Payload和@Headers注解可以消息中的body（字符串类型）与 headers 信息；

　　@RabbitHandler用于接收消息处理，下面的写法是同一个队列下所有类型的消息都会执行该方法；

@Slf4j
@Component
@RabbitListener(containerFactory = "rabbitListenerContainerFactory", queues = {
		"queue.order" })
public class OrderMessageHandler1 {

	@RabbitHandler(isDefault = true)
	public void handleOrderQueue(Message message, @Payload String body, Channel channel) throws IOException {
		long msgTag = message.getMessageProperties().getDeliveryTag();
		log.info("body:{}, message:{}, msgTag:{}", body, message, msgTag);

		// 成功确认，使用此回执方法后，消息会被 rabbitmq broker 删除
		channel.basicAck(msgTag, false);
	}
}

　　@RabbitListener注解中containerFactory为注入的RabbitListenerContainerFactory类型的bean名，queues为要监听的队列的名；

　　执行如下：　

body:{"orderId":"dcfac502-2dc2-4f15-b7e0-352148c1c854","createDate":1624895624124,"totalAmount":null,"payAmount":null,"payType":null}, message:(Body:'[B@4644fcc8(byte[128])' MessageProperties [headers={spring_listener_return_correlation=2089040d-82e1-4f89-877c-7eb532a7c4fa, spring_returned_message_correlation=dcfac502-2dc2-4f15-b7e0-352148c1c854}, contentType=application/octet-stream, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=exchange.order, receivedRoutingKey=key.order, deliveryTag=1, consumerTag=amq.ctag-msLGsIwV9daq1O0avuXXyQ, consumerQueue=queue.order]), msgTag:1

　　

　　RabbitListenerContainerFactory的继承树如下： 

　　

　　AbstractRabbitListenerContainerFactory#setMessageConverter 设置消息转换器；

  • 消息处理方法参数是由 MessageConverter 转化，消息的body的类型默认是byte[]；如果要使用自定义 MessageConverter 则需要在 RabbitListenerContainerFactory 实例中去设置，参考上面的消息处理器（默认的实现是 SimpleRabbitListenerContainerFactory）；

  • 消息的 content_type 属性表示消息 body 数据以什么数据格式存储，接收消息除了使用 Message 对象接收消息（包含消息属性等信息）之外，还可直接使用对应类型接收消息 body 内容，但若方法参数类型不正确会抛异常：

    • application/octet-stream：二进制字节数组存储，使用 byte[]；

    • application/x-java-serialized-object：java 对象序列化格式存储，使用 Object、相应类型（反序列化时类型应该同包同名，否者会抛出找不到类异常）；

    • text/plain：文本数据类型存储，使用 String；

    • application/json：JSON 格式，使用 Object、相应类型；

　　@RabbitListener也可以修饰消息处理的方法，如下：

@RabbitListener(containerFactory = "rabbitListenerContainerFactory", queues = {
		"queue.order" })
public void handleOrderQueue(Message message, @Payload String body, Channel channel)
		throws IOException {
	long msgTag = message.getMessageProperties().getDeliveryTag();
	log.info("body:{}, message:{}, msgTag:{}", body, message, msgTag);

	// 成功确认，使用此回执方法后，消息会被 rabbitmq broker 删除
	channel.basicAck(msgTag, false);
}

　　

　　执行如下：

body:{"orderId":"1f4e343e-a26f-4410-9939-102e728e9df1","createDate":1624897186777,"totalAmount":null,"payAmount":null,"payType":null}, message:(Body:'[B@4d1af5df(byte[128])' MessageProperties [headers={spring_listener_return_correlation=a87fbdb8-12e1-4828-bdee-b92c76b910cf, spring_returned_message_correlation=1f4e343e-a26f-4410-9939-102e728e9df1}, contentType=application/octet-stream, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=exchange.order, receivedRoutingKey=key.order, deliveryTag=1, consumerTag=amq.ctag-2liRtkURbpuuRgV4EGyyZA, consumerQueue=queue.order]), msgTag:1

　　

　　通过@RabbitListener注解的属性bindings声明Binding，如果RabbitMQ中不存在该绑定所需要的Queue、Exchange、RouteKey 则自动创建，若存在则抛出异常，使用如下：

@RabbitListener(containerFactory = "rabbitListenerContainerFactory", admin = "rabbitAdmin", bindings = {
		@QueueBinding(value = @Queue(name = "queue.order"), exchange = @Exchange(name = "exchange.order", type = ExchangeTypes.TOPIC), key = "key.order") })
public void handleOrderQueue(Message message, @Payload String body, Channel channel)
		throws IOException {
	long msgTag = message.getMessageProperties().getDeliveryTag();
	log.info("body:{}, message:{}, msgTag:{}", body, message, msgTag);

	// 成功确认，使用此回执方法后，消息会被 rabbitmq broker 删除
	channel.basicAck(msgTag, false);
}

　　

　　应用启动后，查看管控台；

　　执行如下：

body:{"orderId":"97d08e05-588f-4cb0-b551-f5fb32634e96","createDate":1624898825450,"totalAmount":null,"payAmount":null,"payType":null}, message:(Body:'[B@12fea6bb(byte[128])' MessageProperties [headers={spring_listener_return_correlation=0924912e-a6a9-4e08-8196-1355a04e2325, spring_returned_message_correlation=97d08e05-588f-4cb0-b551-f5fb32634e96}, contentType=application/octet-stream, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=exchange.order, receivedRoutingKey=key.order, deliveryTag=3, consumerTag=amq.ctag-1e3NzsIvhnYRFl295JKOvQ, consumerQueue=queue.order]), msgTag:1

　　

Spring Boot整合RabbitMQ

使用application.properties配置，需要使用Spring Boot配置注入，在application.properties配置文件配置RabbitMQ的连接信息；

连接配置如下：

spring.rabbitmq.host=192.168.211.135
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=password
spring.rabbitmq.virtual-host=/dev
  • 生产者到交换机

开启confirmCallback，生产者投递消息后，如果Broker收到消息后，会给生产者一个ACK；生产者通过ACK，可以确认这条消息是否正常发送到Broker，这种方式是消息可靠性投递的核心；

spring.rabbitmq.publisher-confirm-type: correlated

Spring Boot2.2之前使用如下配置：

spring.rabbitmq.publisher-confirms=true
  • 交换机到队列

开启returnCallback配置

spring.rabbitmq.publisher-returns=true

配置交换机投递到队列失败的策略

交换机到队列不成功，则丢弃消息（默认）；交换机到队列不成功，返回给消息生产者，触发returnCallback；

#交换机处理消息到路由失败，则会返回给生产者
spring.rabbitmq.template.mandatory=true

对应的RabbitTemplate配置

template.setMandatory(true);
  • RabbitMQ的ACK
    • 消费者从RabbitMQ收到消息并处理完成后，反馈给RabbitMQ，RabbitMQ收到反馈后才将此消息从队列中删除；
    • 消费者在处理消息出现了网络不稳定、服务器异常等现象，那么就不会有ACK反馈，RabbitMQ会认为这个消息没有正常消费，会将消息重新放入队列中；
    • 只有当消费者正确发送ACK应答，RabbitMQ确认收到后，消息才会从RabbitMQ服务器的数据中删除；
    • 消息的ACK确认机制默认是自动确认的，消息如果未被进行ACK的消息确认应答，这条消息被锁定unacked；

    • 确认方式

      • 自动确认（默认）
      • 手动确认 manual

　　　　　　org.springframework.amqp.core.AcknowledgeMode 消息确认枚举

　　　　　　

application.properties配置

#开启手动确认消息，如果消息重新入队，进行重试
spring.rabbitmq.listener.simple.acknowledge-mode=manual

　　

　　使用了Spring Boot配置会自动生成RabbitAdmin，需要将上面的queue.order队列消息消费的测试代码中@RabbitListener的admin属性去掉；

　　containerFactory的创建在org.springframework.boot.autoconfigure.amqp.RabbitAnnotationDrivenConfiguration#simpleRabbitListenerContainerFactory

　　

　　RabbitListenerAnnotationBeanPostProcessor实现了BeanPostProcessor接口，BeanPostProcessor为bean的后置处理器，在bean初始化之前调用进行拦截，在bean初始化前后进行一些处理工作 ，说明此时的bean已经进行了实例化；

　　

 　　@RabbitListener上属性的Bean则在postProcessAfterInitialization方法里创建处理；

  • 死信队列

　　死信转移过程：

 　　

 　　死信队列配置类如下：

@Configuration
public class DLXQueueConfig {

	@Bean
	public Queue productDLXQueue() {
		Queue queue = new Queue("queue.product.dlx");
		queue.addArgument("x-message-ttl", 10000);
		queue.addArgument("x-dead-letter-exchange", "exchange.product.dlx");
		queue.addArgument("x-dead-letter-routing-key", "key.product.dlx");
		return queue;
	}

	@Bean
	public Exchange productExchange() {
		TopicExchange exchange = new TopicExchange("exchange.product");
		return exchange;
	}

	@Bean
	public Binding productDlxQueueBinding() {
		Binding binding = new Binding("queue.product.dlx",
				Binding.DestinationType.QUEUE,
				"exchange.product",
				"key.product",
				null);
		return binding;
	}

	@Bean
	public Queue productQueue() {
		Queue queue = new Queue("queue.product");
		return queue;
	}

	@Bean
	public Exchange productDLXExchange() {
		TopicExchange exchange = new TopicExchange("exchange.product.dlx");
		return exchange;
	}

	@Bean
	public Binding productQueueBinding() {
		Binding binding = new Binding("queue.product",
				Binding.DestinationType.QUEUE,
				"exchange.product.dlx",
				"key.product.dlx",
				null);
		return binding;
	}


}　　

　　程序启动后创建对应的Queue，Exchange，Binding; 

 

 

 

　　消息处理类

@Slf4j
@Component
public class DLXMessageHandler {

	@RabbitListener(queues = {"queue.product"})
	public void handleTTLMessage(Message message, @Payload String body, Channel channel) throws IOException {
		long msgTag = message.getMessageProperties().getDeliveryTag();
		log.info("body:{}, message:{}, msgTag:{}", body, message, msgTag);

		// 成功确认，使用此回执方法后，消息会被 rabbitmq broker 删除
		channel.basicAck(msgTag, false);

	}
}

　　

　　当往exchange.product交换机投递消息，如果该消息10秒内都不消费，10秒后该队列会变成死信队列，将消息投递到另一个交换机，交换机再将消息通过路由键路由到对应的队列；

　　执行如下：

body:{"orderId":"c7f57429-41b1-483e-bf4d-edf24783e043","createDate":1624974305090,"totalAmount":null,"payAmount":null,"payType":null}, message:(Body:'[B@110ea79f(byte[128])' MessageProperties [headers={spring_listener_return_correlation=ab1b027e-aad0-4a1d-8551-6c27eac0772d, spring_returned_message_correlation=c7f57429-41b1-483e-bf4d-edf24783e043, x-first-death-exchange=exchange.product, x-death=[{reason=expired, count=1, exchange=exchange.product, time=Tue Jun 29 21:45:13 CST 2021, routing-keys=[key.product], queue=queue.product.dlx}], x-first-death-reason=expired, x-first-death-queue=queue.product.dlx}, contentType=application/octet-stream, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=exchange.product.dlx, receivedRoutingKey=key.product.dlx, deliveryTag=2, consumerTag=amq.ctag-wGj9Z_aaSndUaUTKHIFC-g, consumerQueue=queue.product]), msgTag:2

　　

　　当消息发送的是一个实体，不是字符串，如下：

OrderDTO orderDTO = new OrderDTO();
String uuid = UUID.randomUUID().toString();
orderDTO.setOrderId(uuid);
orderDTO.setCreateDate(new Date());

CorrelationData correlationData = new CorrelationData();
correlationData.setId(uuid);
rabbitTemplate.convertAndSend("exchange.product", "key.product", orderDTO, correlationData);

　　

　　

　　消息处理修改如下：

@Slf4j
@Component
@RabbitListener(queues = {"queue.product"})
public class DLXMessageHandler {

	@RabbitHandler
	public void handleTTLMessage(Message message, OrderDTO orderDTO, Channel channel) throws IOException {
		long msgTag = message.getMessageProperties().getDeliveryTag();
		log.info("body:{}, message:{}, msgTag:{}", orderDTO, message, msgTag);

		// 成功确认，使用此回执方法后，消息会被 rabbitmq broker 删除
		channel.basicAck(msgTag, false);

	}
}

　　执行如下：

body:OrderDTO(orderId=d36438b9-9cf7-48a8-a835-064b4a90201f, createDate=Tue Jun 29 22:27:19 CST 2021, totalAmount=null, payAmount=null, payType=null), message:(Body:'[B@7cae8af(byte[271])' MessageProperties [headers={spring_listener_return_correlation=77d4d8fc-7d6e-4ab9-a7bb-3d3c6ac30f02, spring_returned_message_correlation=d36438b9-9cf7-48a8-a835-064b4a90201f, x-first-death-exchange=exchange.product, x-death=[{reason=expired, count=1, exchange=exchange.product, time=Tue Jun 29 22:27:27 CST 2021, routing-keys=[key.product], queue=queue.product.dlx}], x-first-death-reason=expired, x-first-death-queue=queue.product.dlx}, contentType=application/x-java-serialized-object, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=exchange.product.dlx, receivedRoutingKey=key.product.dlx, deliveryTag=3, consumerTag=amq.ctag-yukRV_EsHHtH6-pVxx6yZQ, consumerQueue=queue.product]), msgTag:3

　　

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/coder-zyc/p/14916736.html