微服务框架下的数据一致性第二篇

Posted daishoucheng

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了微服务框架下的数据一致性第二篇相关的知识，希望对你有一定的参考价值。

　　上篇文章主要讲述的是如何实现的思路问题，本文内容主要是结合我们代码的具体实现来讲解一下（springboot环境下实现）。

　　首先我们需要定义一些常量和工具类：如ExchangeEnum（交换器常量）和QueueEnum（队列常量）以及一些工具类，这都是准备工作（具体可以看github上的代码）。下面正式上代码。

　　创建Rabbitmq的基本配置主要是为了创建ConnectionFactory连接，单例的RabbitTemplate连接以及监听者SimpleMessageListenerContainer，代码如下

/**
 * @功能:【RabbitMQConfig RabbitMQ配置】
 * @作者:代守诚
 * @日期:2018/10/19
 * @时间:9:38
 */
@Configuration
@ConfigurationProperties("common.rabbitmq.config")
public class RabbitmqConfig {

    /**
     * 服务器主机地址
     */
    @Value("${spring.rabbitmq.config.host}")
    private String host;
    /**
     * 端口
     */
    @Value("${spring.rabbitmq.config.port}")
    private Integer port;
    /**
     * 用户名
     */
    @Value("${spring.rabbitmq.config.username}")
    private String username;
    /**
     * 密码
     */
    @Value("${spring.rabbitmq.config.password}")
    private String password;
    /**
     * 虚拟机地址
     */
    @Value("${spring.rabbitmq.config.virtualHost}")
    private String virtualHost;

    /**
     * 连接器
     */
    @Bean
    public ConnectionFactory createConnectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        /**
         * 同样一个RabbitTemplate只支持一个ConfirmCallback。
         * 对于发布确认，template要求CachingConnectionFactory的publisherConfirms属性设置为true。
         * 如果客户端通过setConfirmCallback(ConfirmCallback callback)注册了RabbitTemplate.ConfirmCallback，那么确认消息将被发送到客户端。
         * 这个回调函数必须实现以下方法：
         * void confirm(CorrelationData correlationData, booleanack);
         */
        connectionFactory.setPublisherConfirms(true);//开启confirm模式
        /**
         * 对于每一个RabbitTemplate只支持一个ReturnCallback。
         * 对于返回消息，模板的mandatory属性必须被设定为true，
         * 它同样要求CachingConnectionFactory的publisherReturns属性被设定为true。
         * 如果客户端通过调用setReturnCallback(ReturnCallback callback)注册了RabbitTemplate.ReturnCallback，那么返回将被发送到客户端。
         * 这个回调函数必须实现下列方法：
         *void returnedMessage(Message message, intreplyCode, String replyText,String exchange, String routingKey);
         */
        connectionFactory.setPublisherReturns(true);
        System.out.println("创建消息ConnectionFactory" + username);
        return connectionFactory;
    }

    /***
     *  消息操作模板 这里写成多例，是为了后续 针对特定业务定义特定的回调
     * @return
     */
    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE) //每次注入的时候回自动创建一个新的bean实例
    public RabbitTemplate amqpTemplate() {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(createConnectionFactory());
        rabbitTemplate.setExchange(ExchangeEnum.BD_FENQI_EXCHANGE.getValue());
        System.out.println("*************" + JsonTool.getJsonString(rabbitTemplate));
        return rabbitTemplate;
    }

    /**
     * queue listener  观察 监听模式
     * 当有消息到达时会通知监听在对应的队列上的监听对象
     *
     * @return
     */
    public SimpleMessageListenerContainer getSimpleMessageListenerContainer(ChannelAwareMessageListener mqConsumer) {
        System.err.println("#############进入监听者#############");
        SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(createConnectionFactory());
        listenerContainer.setExposeListenerChannel(true);//暴露监听渠道
        listenerContainer.setMaxConcurrentConsumers(10); //最大消费者数，大于ConcurrentConsumers
        listenerContainer.setConcurrentConsumers(3); //消费者数量
        listenerContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认
        listenerContainer.setMessageListener(mqConsumer);//监听的消费者
        return listenerContainer;
    }

    public String getHost() {
        return host;
    }

    public void setHost(String host) {
        this.host = host;
    }

    public Integer getPort() {
        return port;
    }

    public void setPort(Integer port) {
        this.port = port;
    }

    public String getUsername() {
        return username;
    }

    public void setUsername(String username) {
        this.username = username;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    public String getVirtualHost() {
        return virtualHost;
    }

    public void setVirtualHost(String virtualHost) {
        this.virtualHost = virtualHost;
    }
}

　　具体的实现意义都是有注释的，同时我们还需要创建持久化的队列，持久化的交换器以及绑定关系

/**
 * @功能:【QueueConfig 队列配置】
 * @作者:代守诚
 * @日期:2018/10/19
 * @时间:10:36
 */
@Configuration
public class QueueConfig {
    /**
     * 注释
     * durable="true" 持久化 rabbitmq重启的时候不需要创建新的队列
     * auto-delete 表示消息队列没有在使用时将被自动删除 默认是false
     * exclusive  表示该消息队列是否只在当前connection生效,默认是false
     */

    @Bean
    public Queue applyTestQueue() {
        return new Queue(QueueEnum.BD_FENQI_APPLY_QUEUE.getQueueValue(), true, false, false);
    }

}
/**
 * @功能:【ExchangeConfig 消息交换机配置】
 * @作者:代守诚
 * @日期:2018/10/19
 * @时间:9:41
 */
@Configuration
public class ExchangeConfig {

    /**
     * 注释
     * 1.定义direct exchange，绑定queueTest
     * 2.durable="true" rabbitmq重启的时候不需要创建新的交换机
     * 3.direct交换器相对来说比较简单，匹配规则为：如果路由键匹配，消息就被投送到相关的队列
     * fanout交换器中没有路由键的概念，他会把消息发送到所有绑定在此交换器上面的队列中。
     * topic交换器你采用模糊匹配路由键的原则进行转发消息到队列中
     * key: queue在该direct-exchange中的key值，当消息发送给direct-exchange中指定key为设置值时，
     * 消息将会转发给queue参数指定的消息队列
     */
    @Bean
    public DirectExchange bdfqChange() {
        return new DirectExchange(ExchangeEnum.BD_FENQI_EXCHANGE.getValue(), true, false);
    }
}
/**
 * @功能:【BindConfig 绑定队列消息】
 * @作者:代守诚
 * @日期:2018/10/19
 * @时间:15:49
 */
@Configuration
public class BindConfig extends RabbitmqConfig {

    @Bean
    public Binding bdfqBinding(@Qualifier("applyTestQueue") Queue queue, @Qualifier("bdfqChange") DirectExchange directExchange) {
        return BindingBuilder.bind(queue).to(directExchange).with(QueueEnum.BD_FENQI_APPLY_QUEUE.getRoutingKey());
    }

    /**
     * 绑定监听
     */
    @Bean
    public SimpleMessageListenerContainer applyContainer(@Qualifier("applyTestQueue") Queue applyTestQueue, LoanApplyMQConsumer loanApplyMQConsumer) {
        System.err.println("***************" + JsonTool.getFormatJsonString(loanApplyMQConsumer));
        SimpleMessageListenerContainer applyContainer = getSimpleMessageListenerContainer(loanApplyMQConsumer);
        applyContainer.setQueues(applyTestQueue);//设置监听队列的名称
        System.err.println("***************" + JsonTool.getFormatJsonString(applyContainer));
        return applyContainer;
    }
}

接下来我们需要创建生产者和消费者的基类，供具体的生产者和消费者继承使用：

/**
 * @功能:【MsgSendConfirmCallBack 继承RabbitTemplate.ConfirmCallback, 消息确认监听器】
 * @作者:代守诚
 * @日期:2018/10/19
 * @时间:17:44
 */
public abstract class MsgSendConfirmReturnsCallBack implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

    private RabbitTemplate rabbitTemplate;

    public MsgSendConfirmReturnsCallBack(RabbitTemplate re) {
        System.out.println("**************开始加载********************");
        this.rabbitTemplate = re;
        this.rabbitTemplate.setConfirmCallback(this); //设置
        this.rabbitTemplate.setMessageConverter(new FastJsonMQMessageConvert());
        System.out.println("**************加载成功********************" + JsonTool.getJsonString(rabbitTemplate));
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack) {
            System.out.println("***********消息发送成功id**************" + correlationData.getId());
            changeStatus(correlationData.getId());//处理自己的业务逻辑
        } else {
            System.out.println("***********消息发送失败id**************" + correlationData.getId());
            //处理自己的业务逻辑,比如说开启重试机制(本系统根据自身的业务需求,在定时任务中统一处理)
        }
    }

    @Override
    public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText, String exchange, String routingKey) {
        System.err.println("*******收到回调*******");
        System.err.println("********return--message:" + new String(message.getBody()) + ",replyCode:" + replyCode + ",replyText:" + replyText + ",exchange:" + exchange + ",routingKey:" + routingKey + "********");
    }

    /**
     * 外部请求发送消息
     *
     * @param routingKey
     * @return
     */
    public boolean sendMsg(String routingKey, Object msg) {
        Message message = new Message(null, msg, 1, System.currentTimeMillis(), routingKey);
        return sendMsg(message);
    }

    /**
     * 回调处理
     *
     * @param correlationDataId
     */
    protected abstract void changeStatus(String correlationDataId);

    private boolean sendMsg(Message inputMessage) {
        String id = UUID.randomUUID().toString().replaceAll("-", "");
        boolean isSuccess;
        try {
            // //重置消息id
            inputMessage.setId(id);
            //存入redis中
//            baseRedisDao.zadd(ERedisCacheKey.KEY_RABBIT_MSG_CACHE.getCode(),Double.parseDouble(String.valueOf(System.currentTimeMillis())),
//                    id,0); //保存消息编号
//            baseRedisDao.saveOrUpdate(ERedisCacheKey.KEY_RABBIT_ROUTING_KEY.getCode()+id,inputMessage.getRoutingKey()); //保存消息类型，以routingKey作为标志
//            baseRedisDao.saveOrUpdate(ERedisCacheKey.KEY_RABBIT_MSG_CONTENT_CACHE.getCode()+id,inputMessage); //保存消息主体内容
            rabbitTemplate.convertAndSend(inputMessage.getRoutingKey(), inputMessage.getMsg(), new CorrelationData(id));
            isSuccess = true;
        } catch (Exception e) {
            System.err.println("发送消息异常id:" + id);
            System.err.println(e);
            isSuccess = false;
        }
        return isSuccess;
    }

    public static class Message {
        /**
         * uuid
         */
        private String id;
        /**
         * 消息内容
         */
        private Object msg;
        /**
         * 已发送次数
         */
        private Integer count;
        /**
         * 有效期
         */
        private Long time;
        /**
         * routingKey
         */
        private String routingKey;

        public String getRoutingKey() {
            return routingKey;
        }

        public void setRoutingKey(String routingKey) {
            this.routingKey = routingKey;
        }

        public Message() {
        }

        public Message(String id, Object msg, Integer count, Long time, String routingKey) {
            this.id = id;
            this.msg = msg;
            this.count = count;
            this.time = time;
            this.routingKey = routingKey;
        }

        public String getId() {
            return id;
        }

        public void setId(String id) {
            this.id = id;
        }

        public Object getMsg() {
            return msg;
        }

        public void setMsg(Object msg) {
            this.msg = msg;
        }

        public Integer getCount() {
            return count;
        }

        public void setCount(Integer count) {
            this.count = count;
        }

        public Long getTime() {
            return time;
        }

        public void setTime(Long time) {
            this.time = time;
        }
    }
}

其中confirm用于处理消费发送成功后MQ给我们回调确认，returnMessage同样的道理，上一篇文章也是有介绍过的。

/**
 * @功能:【BaseConsumer 消费者基类】
 * @作者:代守诚
 * @日期:2018/10/22
 * @时间:15:46
 */
public abstract class BaseConsumer implements ChannelAwareMessageListener {

    private FastJsonMQMessageConvert convert = new FastJsonMQMessageConvert();

    protected <T> T fromMessage(Message message, Class<T> cls) throws MessageConversionException {
        return JsonTool.getObj(String.valueOf(convert.fromMessage(message)), cls);
    }

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        try {
            processMessage(message);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            System.err.println("********处理异常:" + JsonTool.getString(message));
            System.err.println("*********" + e);
            processExceptionMessage(message, channel);
        }
    }

    protected abstract void processExceptionMessage(Message message, Channel channel) throws Exception;

    protected abstract void processMessage(Message message);

}

onMessage用于接收MQ服务器推送过来的消息，成功的话正确给响应，出现异常不发送响应。

下面进入正式的生产者和消费者：

生产者：

/**
 * @功能:【LoanApplyMQProducer 申请生产者】
 * @作者:代守诚
 * @日期:2018/10/19
 * @时间:17:01
 */
@Component
public class LoanApplyMQProducer extends MsgSendConfirmReturnsCallBack {

    public LoanApplyMQProducer(@Qualifier("amqpTemplate") RabbitTemplate rabbitTemplate) {
        super(rabbitTemplate);
    }

    @Override
    protected void changeStatus(String correlationDataId) {
        //根据需求完成状态修改
    }

    public boolean send(MemberPojo memberPojo) {
        memberPojo.setId(92L);
        memberPojo.setIdCardCode("3*************8");
        memberPojo.setBirthday(new Timestamp(System.currentTimeMillis()));
        memberPojo.setNickName("王者荣耀");
        memberPojo.setPhoneNumber("176****5555");
        memberPojo.setSex(1);
        memberPojo.setUserName("CN Dota2");
        return sendMsg(QueueEnum.BD_FENQI_APPLY_QUEUE.getRoutingKey(), memberPojo);
    }

消费者：

/**
 * @功能:【LoanApplyMQConsumer 申请消费者】
 * @作者:代守诚
 * @日期:2018/10/19
 * @时间:17:04
 */
@Component
public class LoanApplyMQConsumer extends BaseConsumer {


    @Override
    protected void processExceptionMessage(Message message, Channel channel) throws Exception {
        channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
    }

    @Override
    protected void processMessage(Message message) {
        //处理自身的业务逻辑
        System.err.println("************消费者开始处理业务逻辑***************");
        MemberPojo memberPojo = fromMessage(message, MemberPojo.class);
        System.err.println("**********" + JsonTool.getJsonString(fromMessage(message, MemberPojo.class)));
    }
}

测试方法：

/**
 * @功能:【RabbitmqTestController 测试rabbitmq方法】
 * @作者:代守诚
 * @日期:2018/10/23
 * @时间:11:38
 */
@RestController
@RequestMapping("/rabbit")
public class RabbitmqTestController {
    @Autowired
    private LoanApplyMQProducer loanApplyMQProducer;

    @RequestMapping(value = "/test", method = {RequestMethod.POST, RequestMethod.GET})
    public void rabbitmqTest() {
        System.err.println("**********发送方法准备就绪***********");
        boolean flag = loanApplyMQProducer.send(new MemberPojo());
        System.err.println("*********发送成功***********" + flag);
    }
}

　　这就是具体的实现代码。如果你细心就会发现我们的代码中还有一个没有持久化，那就是message（消息）。如果一旦发生服务器宕机的情况，那么发送的队列里面的message都会消失，但是如果我们使用rabbitmq自带的消息持久化工具，我们同样会遇到如果消息的数据量过大的情况，会造成持久化的时候发生大量的IO操作，十分的消耗性能。这样我们就需要考虑一个更加优秀的方案来执行。作者的公司使用的是redis来完成这个操作，我们会在发送消息的时候将message（消息）存放在指定的redis队列中，在生产者收到ack回执后我们将其删除，对于没有收到回执的消息队列会将其对应的消息id打印出来方便日后查询。另外我们会启动一个定时任务，定时去redis中轮询那些在一定时间内（比如说十分钟）还没有删除的数据，查询到这些数据我们会将其重新发送到rabbitmq中，但是重试的次数一般都是有限制的（我们限制4次重试失败默认删除），一旦超过这个限制我们就会默认删除数据，进行手动处理。

　　ps：由于这只是一个demo版本，我并没有加入logger和redis的操作。

 

代码连接：https://github.com/shouchengdai/rabbitmq_test

　　

以上是关于微服务框架下的数据一致性第二篇的主要内容，如果未能解决你的问题，请参考以下文章

Mybatis框架第二篇

微服务治理实践 | 金丝雀发布

微服务架构下的session一致性

微服务架构下的数据一致性

Flask第二篇——服务器相关

微服务架构下的数据一致性保证