RabbitMQ学习-客户端API学习(springboot集成)
写在前面这里介绍一下RabbitMQ客户端的操作方式包括以下两个方面Rest APISDK API一、
·
写在前面
这里介绍一下RabbitMQ客户端的操作方式
一、项目构建
1.1、 引入依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
1.2、yml 配置
spring:
rabbitmq:
addresses: ip:5672
username: root
password: root
virtual-host: /vhost
# 开启confirms回调 P -> Exchange
publisher-confirms: true
# 开启returnedMessage回调 Exchange -> Queue
publisher-returns: true
# 设置手动确认(ack) Queue
listener:
simple:
acknowledge-mode: manual
template:
mandatory: true
1.3、producer
- 先定义回掉函数
//回调函数: confirm确认
final RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
System.err.println("correlationData: " + correlationData);
String messageId = correlationData.getId();
if (ack) {
//如果confirm返回成功
} else {
//失败则进行具体的后续操作:重试 或者补偿等手段
System.err.println("异常处理...");
}
}
};
- producer 1
public void sendOrder(Order order) throws Exception {
// 通过实现 ConfirmCallback 接口,消息发送到 Broker 后触发回调,确认消息是否到达 Broker 服务器,也就是只确认是否正确到达 Exchange 中
rabbitTemplate.setConfirmCallback(confirmCallback);
// 消息唯一ID
CorrelationData correlationData = new CorrelationData(order.getMessageId());
rabbitTemplate.convertAndSend("ls-exchange", "ls.ABC", order, correlationData);
}
- producer 2
public void sendOrder3(Order order) throws Exception {
rabbitTemplate.setConfirmCallback(confirmCallback);
CorrelationData correlationData = new CorrelationData(order.getMessageId());
rabbitTemplate.convertAndSend("ex1",
"ls.ABC",
order,
message -> {
message.getMessageProperties().setPriority(1);
return message;
}, correlationData
);
}
1.4、consumer
//配置监听的哪一个队列,同时在没有queue和exchange的情况下会去创建并建立绑定关系
@RabbitListener(bindings = @QueueBinding(
value = @Queue(value = "ls-queue", durable = "true"),
exchange = @Exchange(name = "ls-exchange", durable = "true", type = "topic"),
key = "ls.*"))
@RabbitHandler//如果有消息过来,在消费的时候调用这个方法
public void onOrderMessage(@Payload Order order, @Headers Map<String, Object> headers, Channel channel) throws IOException {
logger.info("---收到消息,开始消费--,订单ID:{},Header:{},channel:{}", order.getId(), headers.toString(), channel.toString());
Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
channel.basicAck(deliveryTag, false);
}
/**
* 不定义交换机,收发消息
*
* @param order
* @param headers
* @param channel
* @throws IOException
*/
@RabbitListener(queues = "defaultExchangeQueue", concurrency = "5-40")
public void onOrderMessage3(@Payload String order, @Headers Map<String, Object> headers, Channel channel) throws IOException {
//消费者操作
logger.info("--不定义交换机,开始消费--,订单ID:{},Header:{},channel:{}", order, headers.toString(), channel.toString());
logger.info("重试次数:{}", headers.get("retry"));
Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
//ACK,确认一条消息已经被消费
channel.basicAck(deliveryTag, false);
}
更多推荐
所有评论(0)