MQ发布确认springboot版本
【摘要】 确认机制方案编辑代码架构图 编辑配置文件 spring.rabbitmq.publisher-confirm-type=correlated⚫ NONE禁用发布确认模式，是默认值⚫ CORRELATED发布消息成功到交换器后会触发回调方法⚫ SIMPLE 经测试有两种效果，其一效果和CORRELATED值一样会触发回调方法， 其二在发布消息成功后使用rabbitTemplate调...
配置文件
spring.rabbitmq.publisher-confirm-type=correlated
⚫ NONE
禁用发布确认模式，是默认值
⚫ CORRELATED
发布消息成功到交换器后会触发回调方法
⚫ SIMPLE
经测试有两种效果，其一效果和CORRELATED值一样会触发回调方法， 其二在发布消息成功后使用rabbitTemplate调用waitForConfirms或waitForConfirmsOrDie方法 等待broker节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是 waitForConfirmsOrDie方法如果返回false则会关闭channel，则接下来无法发送消息到broker
spring.application.name=demo6
spring.rabbitmq.host=192.168.231.135
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123
server.port=8989
spring.rabbitmq.publisher-confirm-type=CORRELATED
添加配置类
/**
* 发布确认的配置类
*/
@Configuration
@Slf4j
public class config {
public static final String CONFIRM_EXCHANGE_NAME="confirm.exchange";
public static final String CONFIRM_QUEUE_NAME="confirm.queue";
//声明业务交换机
@Bean("confirmExchange")
public DirectExchange confirmExchange()
{
return new DirectExchange(CONFIRM_EXCHANGE_NAME);
}
//声明确认队列
@Bean("confirmQueue")
public Queue confirmQueue()
{
return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
}
//声明确认队列和绑定关系
@Bean
public Binding queueBinding(@Qualifier("confirmQueue") Queue queue,@Qualifier("confirmExchange") DirectExchange exchange)
{
return BindingBuilder.bind(queue).to(exchange).with("key1");
}
}
回调接口
@Component
@Slf4j
public class MyCallBack implements RabbitTemplate.ConfirmCallback {
@Override
public void confirm(CorrelationData correlationData, boolean ack ,String cause) {
String id=correlationData!=null?correlationData.getId():"";
if(ack)
{
log.info("交换机已经收到id为{}的消息",id);
}
else
{
log.info("交换机还未收到id未：{}的消息，原因是{}",cause);
}
}
}消息生产者
@RestController
@RequestMapping("/confirm")
@Slf4j
public class produce {
public static final String confirm_exchange_name="confirm.exchange";
@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private MyCallBack myCallBack;
@PostConstruct
public void init()
{
rabbitTemplate.setConfirmCallback(myCallBack);
}
@GetMapping("sendMessage1/{message}")
public void sendMessage1(@PathVariable String message)
{
//指定消息id为1
CorrelationData correlationData1=new CorrelationData("1");
String routingKey="key1";
rabbitTemplate.convertAndSend(confirm_exchange_name,routingKey,message+routingKey,correlationData1);
log.info("发送的消息的内容{}",message);
CorrelationData correlationData2=new CorrelationData("2");
routingKey="key2";
rabbitTemplate.convertAndSend(confirm_exchange_name,routingKey,message+routingKey,correlationData2);
log.info("发送的消息的内容{}",message);
}
}消息消费者
@Component
@Slf4j
public class ConfirmConsumer {
public static final String CONFIRM_QUEUE_NAME="confirm.queue";
@RabbitListener(queues = CONFIRM_QUEUE_NAME)
public void receiveMsg(Message message)
{
String s = new String(message.getBody());
log.info("接收到队列confirm.queue消息：{}",s);
}
}发送消息
可以看到，发送了两条消息，第一条消息的 RoutingKey 为 "key1"，第二条消息的 RoutingKey 为 "key2"，两条消息都成功被交换机接收，也收到了交换机的确认回调，但消费者只收到了一条消息，因为第二条消息的 RoutingKey 与队列的 BindingKey 不一致，也没有其它队列能接收这个消息，所有第二条消息被直接丢弃了。
【版权声明】本文为华为云社区用户原创内容，转载时必须标注文章的来源（华为云社区）、文章链接、文章作者等基本信息， 否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱：
cloudbbs@huaweicloud.com
- 点赞
- 收藏
- 关注作者
评论（0）