1. 创建队列
public Queue(String name, boolean durable, boolean exclusive, boolean autoDelete, @Nullable Map<String, Object> arguments) {
super(arguments);
Assert.notNull(name, "'name' cannot be null");
this.name = name;
this.actualName = StringUtils.hasText(name) ? name : Base64UrlNamingStrategy.DEFAULT.generateName() + "_awaiting_declaration";
this.durable = durable; // 是否持久化
this.exclusive = exclusive; // 队列中消息是否只提供一个消费者消费
this.autoDelete = autoDelete; // 当全部消费者断开连接后，队列中消息是否自动删除
}
2. 创建exchange
public TopicExchange(String name, boolean durable, boolean autoDelete, Map<String, Object> arguments) {
super(name, durable, autoDelete, arguments);
}
3. 绑定exchange与队列
public Binding(String destination, Binding.DestinationType destinationType, String exchange, String routingKey, @Nullable Map<String, Object> arguments) {
super(arguments);
this.destination = destination;
this.destinationType = destinationType;
this.exchange = exchange;
this.routingKey = routingKey;
}
Binding.DestinationType可以是：QUEUE, 或EXCHANGE
destination为目的地，它可以是一个队列或者是exchange
exchange为数据发送的源exchange
routeKey为匹配投递的路由key．
4.消息发送
生产者可以使用rabbitTemplate通过以下方式发送消息：
- 4.1. 将object序列化为json或其他格式(默认采用的是JAVA的序列化)，直接发送到指定exchange,通过routeKey转发到相应的队列．
public void convertAndSend(String exchange, String routingKey, Object object, @Nullable CorrelationData correlationData) throws AmqpException {
this.send(exchange, routingKey, this.convertMessageIfNecessary(object), correlationData);
}
- 4.2 发送前先通过postProcessor完成消息的处理，然后将处理后的消息发送．
public void convertAndSend(String exchange, String routingKey, Object message, MessagePostProcessor messagePostProcessor, @Nullable CorrelationData correlationData) throws AmqpException {
Message messageToSend = this.convertMessageIfNecessary(message);
messageToSend = messagePostProcessor.postProcessMessage(messageToSend, correlationData);
this.send(exchange, routingKey, messageToSend, correlationData);
}
5. 消息消费
- 5.1 消费者监听队列
通过@RabbitListener明一个消费者，并声明他监听的队列．通过@RabbitHandler声明消息被接收后的处理器．同一个消费者可以声明多个处理器，用来处理不同类型的消息内容实体．
@RabbitListener(queues = {"packet-out-of-ware-queue"})
public class OrderConsumer {
@RabbitHandler
public void receiveMsg(Message message, OrderEntity orderEntity){
System.out.println(message + "\n" + orderEntity);
}
// 处理来自于packet-out-of-ware队列中，消息体为OrderRefundReson的消息
@RabbitHandler
public void receiveMsg(Message message, OrderRefundReson orderRefundReson){
System.out.println(message + "\n" + orderEntity);
}
}
6. 消息可靠投递
6.1 生产者发送，broker确认
包含两个阶段确认：
- 生产者消息发送到broker, broker收到后返回确认信息,需配置
spring.rabbitmq..publisher-confirm-type: simple (或 correlated)
- 生产者消息发送到broker并由exchange将消息发送到指定队列, 如果没能找到队列，返回失败信息，需配置
spring.rabbitmq.publisher-returns: true
spring.rabbitmq.template.mandatory: true
@PostConstruct
public void initRabbitTemplate(){
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
/**
* 档消息被投递到exchange，发送确认后，触发回调
* @param correlationData：关联信息，如消息id
* @param ack
* @param cause
*/
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
System.out.println("confirming... " + "correlationDataId: " + correlationData.getId() + "ack: " + ack);
}
});
rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
/**
* 当消息未被投递到指定队列，如根据routeKey查找队列不存在则触发returnedMessage
* @param message
* @param replyCode
* @param replyText
* @param exchange
* @param routingKey
*/
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
}
});
}
6.2 消费者确认
默认为自动确认，可以选择通过与手动确认
@RabbitHandler
public void receiveMsg(Message message, OrderEntity orderEntity, Channel channel){
try {
System.out.println(message + "\n" + orderEntity);
boolean done = handleMsg(orderEntity);
long deliveryTag = message.getMessageProperties().getDeliveryTag();
if (done){
// 处理成功，以非批量的方式返回确认信息
// basicNack(deliveryTag, multiple)
channel.basicAck(deliveryTag, false);
}else {
// 处理失败，以非批量的方式返回Ｎack信息,同时可选择是否将消息重新放回队列．
//basicNack(deliveryTag, multiple, requeue)
channel.basicNack(deliveryTag, false, false);
}
}catch (Exception e) {
e.printStackTrace();
}
}
