安科网

RocketMQ事务消息

meilongwhpu

meilongwhpu

2020-06-08

关注 关注

RocketMQ事务消息（Transactional Message）是指应用本地事务和发送消息操作可以被定义到全局事务中，要么同时成功，要么同时失败。RocketMQ的事务消息提供类似 X/Open XA 的分布事务功能，通过事务消息能达到分布式事务的最终一致。

Half Message(半消息)

是指暂不能被Consumer消费的消息。Producer 已经把消息成功发送到了 Broker 端，但此消息被标记为暂不能投递状态，处于该种状态下的消息称为半消息。需要 Producer

对消息的二次确认后，Consumer才能去消费它。

消息回查

由于网络闪段，生产者应用重启等原因。导致 Producer 端一直没有对 Half Message(半消息) 进行 二次确认。这是Brock服务器会定时扫描长期处于半消息的消息，会

主动询问 Producer端 该消息的最终状态(Commit或者Rollback),该消息即为 消息回查。

事务消息共有三种状态，提交状态、回滚状态、中间状态：

  • TransactionStatus.CommitTransaction: 提交事务，它允许消费者消费此消息。
  • TransactionStatus.RollbackTransaction: 回滚事务，它代表该消息将被删除，不允许被消费。
  • TransactionStatus.Unknown: 中间状态，它代表需要检查消息队列来确定状态。

RocketMQ事务消息

事务消息发送及提交：

  1. 发送消息（half消息）
  2. 服务端响应消息写入结果
  3. 根据发送结果执行本地事务（如果写入失败，此时half消息对业务不可见，本地逻辑不执行）
  4. 根据本地事务状态执行Commit或者Rollback（Commit操作生成消息索引，消息对消费者可见）

补偿流程：

  1. 对没有Commit/Rollback的事务消息（pending状态的消息），从服务端发起一次“回查”
  2. Producer收到回查消息，检查回查消息对应的本地事务的状态
  3. 根据本地事务状态，重新Commit或者Rollback

补偿阶段用于解决消息Commit或者Rollback发生超时或者失败的情况。

一阶段的消息如何对用户不可见

事务消息相对普通消息最大的特点就是一阶段发送的消息对用户是不可见的。

如何做到写入了消息但是对用户不可见?——写入消息数据，但是不创建对应的消息的索引信息。

RocketMQ事务消息

熟悉RocketMQ的同学应该都清楚，消息在服务端的存储结构如上，每条消息都会有对应的索引信息，Consumer通过索引读取消息。

那么实现一阶段写入的消息不被用户消费（需要在Commit后才能消费），只需要写入Storage Queue，但是不构建Index Queue即可。

RocketMQ中具体实现策略是：写入的如果事务消息，对消息的Topic和Queue等属性进行替换，同时将原来的Topic和Queue信息存储到消息的属性中。

实例

假如客户需要在电商购买东西，流程如下

1、购买商品，客户付款，客户账户金额减少，调用Order API + Pay API；

2、MQ产生Order日志，状态Ordered；

3、商家减库存，调用Stock API；

4、MQ产生减库存Stock日志，状态Stocked；

5、商家收款，调用Order API + Pay API；

6、MQ产生Order日志，状态Finished；

现实情况会更复杂，这6个步骤只是简单处理。

本地事务为：1、3、5

MQ事务为：2、4、6

为了能解决该问题，同时又不和业务耦合，RocketMQ提出了“事务消息”的概念。RocketMQ事务消息（Transactional Message）是指应用本地事务和发送消息操作可以被定义到全局事务中，要么同时成功，要么同时失败。RocketMQ的事务消息提供类似 X/Open XA 的分布事务功能，通过事务消息能达到分布式事务的最终一致。

具体来说，就是把消息的发送分成了2个阶段：Prepare阶段和确认阶段。

具体来说，上面的2个步骤，被分解成3个步骤： 
(1) 发送Prepared消息 
(2) update DB 
(3) 根据update DB结果成功或失败，Confirm或者取消Prepared消息。

事务监听

package com.xin.rocketmq.demo.testrun;

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

public class TransactionListenerImpl implements TransactionListener {
    private AtomicInteger transactionIndex = new AtomicInteger(0);
    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();

    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        int value = transactionIndex.getAndIncrement();
        int status = value % 3;
        localTrans.put(msg.getTransactionId(), status);

        //执行购买商品，客户付款，客户账户金额减少，调用Order API + Pay API；
        System.out.println("执行客户付款:" + msg.getTransactionId());
        //商家减库存，调用Stock API；
        System.out.println("执行库存:" + msg.getTransactionId());
        //商家收款，调用Order API + Pay API；
        System.out.println("执行商家收款:" + msg.getTransactionId());

        return LocalTransactionState.UNKNOW;
    }
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        Integer status = localTrans.get(msg.getTransactionId());
        if (null != status) {
            switch (status) {
                case 0:
                    return LocalTransactionState.UNKNOW;
                case 1:
                    return LocalTransactionState.COMMIT_MESSAGE;
                case 2:
                    return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        }
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}

事务生产者

package com.xin.rocketmq.demo.testrun;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.concurrent.*;

public class TransactionProducer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
        TransactionListener transactionListener = new TransactionListenerImpl();
        TransactionMQProducer producer = new TransactionMQProducer("please_rename_unique_group_name");
        producer.setNamesrvAddr("192.168.10.11:9876");
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("client-transaction-msg-check-thread");
                return thread;
            }
        });
        producer.setExecutorService(executorService);
        producer.setTransactionListener(transactionListener);
        producer.start();
        String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
        for (int i = 0; i < 2; i++) {
            try {
                Message msg =
                        new Message("TopicTest1234", tags[i % tags.length], "KEY" + i,
                                ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = producer.sendMessageInTransaction(msg, null);
                System.out.printf("%s%n", sendResult);
                Thread.sleep(10);
            } catch (MQClientException | UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        }
        for (int i = 0; i < 100000; i++) {
            Thread.sleep(1000);
        }
        producer.shutdown();
    }
}

一批次处理2个订单，结果

SendResult [sendStatus=SEND_OK, msgId=A9FEC2CC2D8C18B4AAC2268193040000, offsetMsgId=null, messageQueue=MessageQueue [topic=TopicTest1234, brokerName=localhost.localdomain, queueId=3], queueOffset=103]
执行客户付款:A9FEC2CC2D8C18B4AAC2268193A20001
执行库存:A9FEC2CC2D8C18B4AAC2268193A20001
执行商家收款:A9FEC2CC2D8C18B4AAC2268193A20001
SendResult [sendStatus=SEND_OK, msgId=A9FEC2CC2D8C18B4AAC2268193A20001, offsetMsgId=null, messageQueue=MessageQueue [topic=TopicTest1234, brokerName=localhost.localdomain, queueId=0], queueOffset=104]
执行客户付款:A9FEC2CC2D8C18B4AAC2268193B10002
执行库存:A9FEC2CC2D8C18B4AAC2268193B10002
执行商家收款:A9FEC2CC2D8C18B4AAC2268193B10002......

当发送半消息成功时，我们使用 executeLocalTransaction 方法来执行本地事务。它返回前一节中提到的三个事务状态之一。checkLocalTranscation 方法用于检查本地事务状态，并回应消息队列的检查请求。它也是返回前一节中提到的三个事务状态之一。

模拟两条记录全部成功走完事务

package com.xin.rocketmq.demo.testrun;

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import static java.lang.System.*;

public class TransactionListenerImpl implements TransactionListener {
    private AtomicInteger transactionIndex = new AtomicInteger(0);
    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();

    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        int value = transactionIndex.getAndIncrement();
        int status = value % 3;
        localTrans.put(msg.getTransactionId(), status);

        //执行购买商品，客户付款，客户账户金额减少，调用Order API + Pay API；
        out.println("执行客户付款:" + msg.getTransactionId());
        //商家减库存，调用Stock API；
        out.println("执行库存:" + msg.getTransactionId());
        //商家收款，调用Order API + Pay API；
        out.println("执行商家收款:" + msg.getTransactionId());

        return LocalTransactionState.COMMIT_MESSAGE;
    }
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}
开启消费者，发现两条事务都确实走完了。

执行客户付款:A9FEC2CC347C18B4AAC2269439D50000
执行库存:A9FEC2CC347C18B4AAC2269439D50000
执行商家收款:A9FEC2CC347C18B4AAC2269439D50000
SEND_OK -- SendResult [sendStatus=SEND_OK, msgId=A9FEC2CC347C18B4AAC2269439D50000, offsetMsgId=null, messageQueue=MessageQueue [topic=topic_family, brokerName=localhost.localdomain, queueId=2], queueOffset=257]
执行客户付款:A9FEC2CC347C18B4AAC226943A790001
执行库存:A9FEC2CC347C18B4AAC226943A790001
执行商家收款:A9FEC2CC347C18B4AAC226943A790001
SEND_OK -- SendResult [sendStatus=SEND_OK, msgId=A9FEC2CC347C18B4AAC226943A790001, offsetMsgId=null, messageQueue=MessageQueue [topic=topic_family, brokerName=localhost.localdomain, queueId=3], queueOffset=258]

Receive message[msgId=A9FEC2CC347C18B4AAC2269439D50000] 86728ms later
Receive message[msgId=A9FEC2CC347C18B4AAC226943A790001] 87048ms later

模拟1条记录在本地库存事务失败回滚

package com.xin.rocketmq.demo.testrun;

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import static java.lang.System.*;

public class TransactionListenerImpl implements TransactionListener {
    private AtomicInteger transactionIndex = new AtomicInteger(0);
    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();

    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        int value = transactionIndex.getAndIncrement();
        int status = value % 3;
        localTrans.put(msg.getTransactionId(), status);

        //执行购买商品，客户付款，客户账户金额减少，调用Order API + Pay API；
        out.println("执行客户付款:" + msg.getTransactionId());
        //商家减库存，调用Stock API；
        out.println("执行库存:" + msg.getTransactionId());
        //商家收款，调用Order API + Pay API；
        if (status == 1)//第二个订单库存异常，模拟库存失败，需要回滚
            return LocalTransactionState.ROLLBACK_MESSAGE;
        out.println("执行商家收款:" + msg.getTransactionId());

        return LocalTransactionState.COMMIT_MESSAGE;
    }
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        Integer status = localTrans.get(msg.getTransactionId());
        if (null != status) {
            switch (status) {
                case 0:
                    return LocalTransactionState.UNKNOW;
                case 1:
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                case 2:
                    return LocalTransactionState.COMMIT_MESSAGE;
            }
        }
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}

两条事务，发现只有一条可以被成功消费。另一条回滚了。

Receive message[msgId=A9FEC2CC355818B4AAC2269681E00000] 

rocketmq 事务 科技新闻

meilongwhpu

meilongwhpu

0 关注 0 粉丝 0 动态

关注 关注

相关推荐

消息疯狂堆积！RocketMQ 出 Bug 了?

用过 MQ 的同学，可能会遇到过消息堆积的问题。而我最近也踩上了这个坑，但是发现结果竟然是这么一个意料之外的原因而导致的。那一晚月和风高，我正准备踏上回家的路，突然收到告警短信轰炸！于是乎赶回公司赶紧打开电脑，登上 RocketMQ 后台查看

LCFlxfldy 2020-08-17

深入理解RocketMQ延迟消息

延迟消息是实际开发中一个非常有用的功能，本文第一部分从整体上介绍秒级精度延迟消息的实现思路，在第二部分结合RocketMQ的延迟消息实现，进行细致的讲解，点出关键部分的源码。第三步介绍延迟消息与消息重试的关系。Broker判断这是一个延迟消息后，将其通过临

IT农场 2020-11-13

RabbitMQ都写了，RocketMQ怎么能落下？

整体架构最近看到了我在Github上写的rabbitmq-examples陆续被人star了，就想着写个rocketmq-examples。对rabbitmq感兴趣的小伙伴可以看我之前的文章。下面把RocketMQ的各个特性简单介绍一下，这样在用的时候心里

ljcsdn 2020-07-27

(五)消息队列

此时，为了提高用户体验和吞吐量，其实可以异步地调用系统B、C、D的接口。所以，我们可以通过引入消息中间件，再系统A执行完后将数据写入到消息队列中直接返回，而BCD系统则负责监听执行。通过以上分析可知，消息中间件最佳选型为RocketMQ。

LCFlxfldy 2020-07-05

RocketMQ(2)---核心概念、特性、使用等

对于RocketMQ而言，感觉官方提供的东西还是可以的：https://github.com/apache/rocketmq/tree/master/docs/cn

lypgcs 2020-06-27

RocketMq producer 发送一条消息所经过的流程

　　RocketMq producer 在发送一条消息时候，从 producer --nameSrv -- Broker 中间经过了什么样子的数据交互。从NameSrv 中获取topic 配置的相关信息，比如 broker 地址，队列数 之类的。

陈晨软件五千言 2020-06-17

RocketMQ系列（三）消息的生产与消费

前面的章节，我们已经把RocketMQ的环境搭建起来了，是一个两主两从的异步集群。接下来，我们就看看怎么去使用RocketMQ，在使用之前，先要在NameServer中创建Topic，我们知道RocketMQ是基于Topic的消息队列，在生产者发送消息的时

qingyuerji 2020-06-14

RocketMQ系列（四）顺序消费

折腾了好长时间才写这篇文章，顺序消费，看上去挺好理解的，就是消费的时候按照队列中的顺序一个一个消费；而并发消费，则是消费者同时从队列中取消息，同时消费，没有先后顺序。RocketMQ也有这两种方式的实现，但是在实践的过程中，就是不能顺序消费，好不容易能够实

MojitoBlogs 2020-06-14

RocketMQ系列（五）广播与延迟消息

今天要给大家介绍RocketMQ中的两个功能，一个是“广播”，这个功能是比较基础的，几乎所有的mq产品都是支持这个功能的；另外一个是“延迟消费”，这个应该算是RocketMQ的特色功能之一了吧。广播是把消息发送给订阅了这个主题的所有消费者。首先，消费者端的

lypgcs 2020-06-14

RocketMQ系列（一）基本概念

RocketMQ是阿里出品的一款开源的消息中间件，让其声名大噪的就是它的事务消息的功能。在企业中，消息中间件选择使用RocketMQ的还是挺多的，这一系列的文章都是针对RocketMQ的，咱们先从RocketMQ的一些基本概念和环境的搭建开始聊起。Rock

陈晨软件五千言 2020-06-14

RocketMq 在Netty 下是如何进行消息封装传输

　　要了解rocketMq 需要知道 数据在 rocketMq 中 是如何进行传输，在底层的结构到底是一个什么亚子，这个需要我们对Netty 对字符编解码有一些了解。　　我们从生产者发送消息，broker 接收消息 为例，来开展底层对消息结构。　　1. 如

meilongwhpu 2020-06-13

RocketMQ

然后执行‘start mqnamesrv.cmd’，启动NAMESERVER。成功后会弹出提示框，此框勿关闭。#set it false if you don‘t want use dashboard.default true. #set the mess

陈晨软件五千言 2020-06-11

RocketMQ集群监控平台搭建

RocketMQ有一个对其扩展的开源项目incubator-rocketmq-externals，这个项目中有一个子模块叫rocketmq-console，这个便是管理控制台项目了，先将incubator-rocketmq-externals拉到本地，因为

qingyuerji 2020-06-09

RocketMQ系列（三）消息的生产与消费

前面的章节，我们已经把RocketMQ的环境搭建起来了，是一个两主两从的异步集群。接下来，我们就看看怎么去使用RocketMQ，在使用之前，先要在NameServer中创建Topic，我们知道RocketMQ是基于Topic的消息队列，在生产者发送消息的时

MojitoBlogs 2020-06-09

聊一聊高并发高可用那些事 - Kafka篇

偶尔推出一个优惠活动时，请求量极速上升。看了几个招聘网站，提到较多的消息队列有：RabbitMQ、RocketMQ、Kafka 以及 Redis 的消息队列和发布订阅模式。在高可用环境中，Kafka 需要部署多台，避免 Kafka 宕机后，服务无法访问。K

meilongwhpu 2020-06-08

RocketMQ-Namesrv源码解析

Name Server是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。Broker部署相对复杂，Broker分为Master和Slave，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master，Master和Slave

lypgcs 2020-06-07

深度剖析 Kafka/RocketMQ 顺序消息的一些坑

我觉得这个问题问得很频繁，而且非常经典，在这里我就以 Kafka 为例子，说说我对 Kafka 顺序消息的一些理解吧，如有理解不对的地方麻烦留言指点一下。针对以上第二点，下面盘点一下 Kafka 集群中有哪些意外情况会打乱消息的顺序。

MojitoBlogs 2020-06-04

RocketMQ源码 — 九、 RocketMQ延时消息

上一节消息重试里面提到了重试的消息可以被延时消费，其实除此之外，用户发送的消息也可以指定延时时间，然后在指定延时时间之后投递消息，然后被consumer消费。broker收到消息在准备将消息写入存储的时候，判断是延时消息则更改Message的topic为延

meilongwhpu 2020-05-30

rocketmq初识

通常一个消息队列需要掌握的知识点有Topic（主体）、Producer（生产者）、Consumer（消费者）、Queue（队列）、Delivery Semantics. 蛋疼的是不同的消息队列关于这些名词叫法不一样，含义也不是很精确。所以阿里起了一个项目O

LCFlxfldy 2020-05-29

[SpringBoot] Spring Boot（13）RabbitMQ安装与集成

　　RabbitMQ是一个开源的消息代理软件，它的核心作用就是创建消息队列，异步接收和发送消息，MQ的全程是：Message Queue中文的意思是消息队列。　　●延迟处理：可以进行事件后置，比如订单超时业务，用户下单30分钟未支付取消订单；　　现在市面上

sweetgirl0 2020-05-17
meilongwhpu

meilongwhpu

W3CSchool教程
HTML 教程
CSS 教程
Bootstrap 教程
Javascript 教程
jQuery 教程
后端教程
C 教程
Java 教程
PHP 教程
Python 教程
Go 教程
移动开发
Android 教程
Swift 教程
Kotlin 教程
jQuery Mobile 教程
ionic 教程
关于我们
新闻动态
联系方式
招聘英才
安科实验室
帮助与反馈

安科网(Ancii)，中国第一极客网

Copyright © 2013 - 2019 Ancii.com

京ICP备18063983号 京公网安备11010802014868号