0
0

RabbitMQ可靠投递之confirmCallback确认模式------RabbitMQ

保证消息不丢失,使用事务消息,会导致性能下降250倍,为此需要使用确认机制
confirmCallback确认模式,从消息生产者到服务器交换机
returnCallback若交换机未能投递到队列,需要重新回退到交换机内
开启发送端确认后
如果没有投递成功,消息会被记录和保存的

保证消息不丢失,使用事务消息,会导致性能下降250倍,为此需要使用确认机制
confirmCallback确认模式,从消息生产者到服务器交换机
returnCallback若交换机未能投递到队列,需要重新回退到交换机内
开启发送端确认后
如果没有投递成功,消息会被记录和保存的

spring.application.name=Mall-order
spring.cloud.nacos.config.server-addr=127.0.0.1:8848
spring.cloud.nacos.discovery.server-addr=127.0.0.1:8848
spring.rabbitmq.host=192.168.56.10
spring.rabbitmq.port=5672
spring.rabbitmq.virtual-host=/
#只要抵达队列,以异步形式回调
spring.rabbitmq.template.mandatory=true
#新版本的发送确认
spring.rabbitmq.publisher-confirm-type=CORRELATED
spring.cloud.nacos.config.namespace=becc5563-6c9b-47ff-a6b1-580982722371

spring.application.name=Mall-order
spring.cloud.nacos.config.server-addr=127.0.0.1:8848
spring.cloud.nacos.discovery.server-addr=127.0.0.1:8848
spring.rabbitmq.host=192.168.56.10
spring.rabbitmq.port=5672
spring.rabbitmq.virtual-host=/
#只要抵达队列,以异步形式回调
spring.rabbitmq.template.mandatory=true
#新版本的发送确认
spring.rabbitmq.publisher-confirm-type=CORRELATED
spring.cloud.nacos.config.namespace=becc5563-6c9b-47ff-a6b1-580982722371

package com.alatus.mall.order.controller;

import com.alatus.mall.order.entity.OrderEntity;
import com.alatus.mall.order.entity.OrderReturnReasonEntity;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Date;
import java.util.UUID;

@RestController
public class RabbitController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @GetMapping("/sendMessage")
    public String sendMessage(){
        for (int i = 0; i < 10; i++) {
            if(i % 2 == 0){
                OrderReturnReasonEntity reasonEntity = new OrderReturnReasonEntity();
                reasonEntity.setId(1L);
                reasonEntity.setCreateTime(new Date());
                reasonEntity.setName("哈哈"+i);
                rabbitTemplate.convertAndSend("helloWorld","hello",reasonEntity);
            }
            else{
                OrderEntity entity = new OrderEntity();
                entity.setOrderSn(UUID.randomUUID().toString());
                rabbitTemplate.convertAndSend("helloWorld","hello",entity);
            }
        }
        return "OK";
    }
}

package com.alatus.mall.order.controller;

import com.alatus.mall.order.entity.OrderEntity;
import com.alatus.mall.order.entity.OrderReturnReasonEntity;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Date;
import java.util.UUID;

@RestController
public class RabbitController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @GetMapping("/sendMessage")
    public String sendMessage(){
        for (int i = 0; i < 10; i++) {
            if(i % 2 == 0){
                OrderReturnReasonEntity reasonEntity = new OrderReturnReasonEntity();
                reasonEntity.setId(1L);
                reasonEntity.setCreateTime(new Date());
                reasonEntity.setName("哈哈"+i);
                rabbitTemplate.convertAndSend("helloWorld","hello",reasonEntity);
            }
            else{
                OrderEntity entity = new OrderEntity();
                entity.setOrderSn(UUID.randomUUID().toString());
                rabbitTemplate.convertAndSend("helloWorld","hello",entity);
            }
        }
        return "OK";
    }
}

package com.alatus.mall.order.config;

import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

@Configuration
public class RabbitConfig {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
//    定制RabbitCallback,构造器创建完成,执行此方法
    @PostConstruct
    public void initRabbitTemplate(){
//        设置确认回调
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println(""+correlationData+ack+cause);
            }
        });
    }
}

package com.alatus.mall.order.config;

import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

@Configuration
public class RabbitConfig {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
//    定制RabbitCallback,构造器创建完成,执行此方法
    @PostConstruct
    public void initRabbitTemplate(){
//        设置确认回调
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println(""+correlationData+ack+cause);
            }
        });
    }
}

package com.alatus.mall.order.service.impl;

import com.alatus.mall.order.entity.OrderEntity;
import com.alatus.mall.order.entity.OrderReturnReasonEntity;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;
import java.util.Map;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.alatus.common.utils.PageUtils;
import com.alatus.common.utils.Query;

import com.alatus.mall.order.dao.OrderItemDao;
import com.alatus.mall.order.entity.OrderItemEntity;
import com.alatus.mall.order.service.OrderItemService;


@Service("orderItemService")
@RabbitListener(queues = "helloJava")
public class OrderItemServiceImpl extends ServiceImpl<OrderItemDao, OrderItemEntity> implements OrderItemService {
    @Override
    public PageUtils queryPage(Map<String, Object> params) {
        IPage<OrderItemEntity> page = this.page(
                new Query<OrderItemEntity>().getPage(params),
                new QueryWrapper<OrderItemEntity>()
        );

        return new PageUtils(page);
    }

    @RabbitHandler
    public void getMessage(Message message, OrderReturnReasonEntity entity, Channel channel){
        System.out.println(entity);
    }
    @RabbitHandler
    public void getMessage1(OrderEntity entity){
        System.out.println(entity);
    }
}

package com.alatus.mall.order.service.impl;

import com.alatus.mall.order.entity.OrderEntity;
import com.alatus.mall.order.entity.OrderReturnReasonEntity;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;
import java.util.Map;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.alatus.common.utils.PageUtils;
import com.alatus.common.utils.Query;

import com.alatus.mall.order.dao.OrderItemDao;
import com.alatus.mall.order.entity.OrderItemEntity;
import com.alatus.mall.order.service.OrderItemService;


@Service("orderItemService")
@RabbitListener(queues = "helloJava")
public class OrderItemServiceImpl extends ServiceImpl<OrderItemDao, OrderItemEntity> implements OrderItemService {
    @Override
    public PageUtils queryPage(Map<String, Object> params) {
        IPage<OrderItemEntity> page = this.page(
                new Query<OrderItemEntity>().getPage(params),
                new QueryWrapper<OrderItemEntity>()
        );

        return new PageUtils(page);
    }

    @RabbitHandler
    public void getMessage(Message message, OrderReturnReasonEntity entity, Channel channel){
        System.out.println(entity);
    }
    @RabbitHandler
    public void getMessage1(OrderEntity entity){
        System.out.println(entity);
    }
}
标签： 后端 spring boot 微服务
本文转载自: https://blog.csdn.net/2201_75960169/article/details/142636151
版权归原作者 旧约Alatus 所有， 如有侵权，请联系我们删除。
登录后发布评论

“RabbitMQ可靠投递之confirmCallback确认模式------RabbitMQ”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

linux的软链接详解

如何在Windows系统上，下载并配置macOS 10.14虚拟机

一款完全开源并免费的监测与分析系统，支持监测，预警，分析，报告，支持本地化部署（附源码）

Git学习-1

毕设开源 python大数据旅游数据分析可视化系统(源码分享)

Kafka 快速入门（一）

基于大数据的电信诈骗行为可视化系统含预测研究【lightGBM，XGBoost，随机森林】

文章导航