先看下图
在这里插入图片描述
以上图例展示了mq事务消息解决分布式事务的producer环节，consumer正常消费即可。
show your code
根据以上流程我们可以用rocketmq很简单的实现如下代码。为了减少部分业务代码入侵做了一点点封装；
以下项目基于springboot2.1.3，此处引入jdbc，大家需要注意，可以和mybatis、mybatis-plus共用事务管理器（想了解jdbc与mybatis如何共用事务管理器，自行百度）。假如你是jpa 或者habinate 你就不能这样封装。
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-aop</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jdbc</artifactId>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.1.1</version>
</dependency>
需在你的业务库插入如下表记录事务日志。
CREATE TABLE `transaction_log` (
`id` int(20) NOT NULL AUTO_INCREMENT,
`trx_id` varchar(128) NOT NULL,
PRIMARY KEY (`id`),
UNIQUE KEY `idx_transaction_id` (`trx_id`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4;
由于我们使用rocketmq-spring-boot-starter这个自动配置，所以我们可以直接使用RocketMQTemplate来发送消息，非常方便
demo中配置
rocketmq.name-server=127.0.0.1:9876
rocketmq.producer.group=tx_group
rocketmq.producer.send-message-timeout=3000
package com.xxx.fw.rocketmq.trx.core;
import com.xxx.fw.rocketmq.trx.config.TrxContextHolder;
import javax.annotation.Resource;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;
/**
* @Description 抽象类让业务继承此类，并实现doBusiness方法，实现不同的业务定制
* @Author 姚仲杰
* * @Date 2020/12/21 20:51
*/
public abstract class AbstractTransactionListener implements RocketMQLocalTransactionListener {
private final static Logger LOGGER= LoggerFactory.getLogger(AbstractTransactionListener.class);
@Resource
TransactionLogService transactionLogService;
public abstract void doBusiness(Object o);
/**这个方法中执行本地事务
* @param message 已发送到mq的事务消息
* @param o 要保存到库的对象
* @return
*/
@Override
public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
RocketMQLocalTransactionState state;
TrxContextHolder.setTrxId(message.getHeaders().getId().toString());
try{
LOGGER.info("执行业务逻辑，trx_id:[{}]",message.getHeaders().getId());
doBusiness(o);
state=RocketMQLocalTransactionState.COMMIT;
LOGGER.info("执行业务逻辑---[COMMIT]");
}catch(Exception ex){
LOGGER.info("消息事务回滚[ROLLBACK] {}",ex);
state=RocketMQLocalTransactionState.ROLLBACK;
}
return state;
}
/**此方法提供统一回查
* @param message 回查的消息数据
* @return
*/
@Override
public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
RocketMQLocalTransactionState state=RocketMQLocalTransactionState.UNKNOWN;
if (transactionLogService.query(message.getHeaders().getId().toString())>0){
LOGGER.info("commit msg trx_id:{}",message.getHeaders().getId());
state=RocketMQLocalTransactionState.COMMIT;
}else{
LOGGER.info("LocalTransaction review [UNKNOWN] will try again");
state=RocketMQLocalTransactionState.UNKNOWN;
}
return state;
}
}
自定义注解
package com.xxx.fw.rocketmq.trx.config;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
* @Description TODO
* @Author 姚仲杰
* @Date 2020/12/28 11:28
*/
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Inherited
public @interface MqTrx {
}
切面处理
package com.xxx.fw.rocketmq.trx.aspect;
import com.xxx.fw.rocketmq.trx.config.MqTrx;
import com.xxx.fw.rocketmq.trx.config.TrxContextHolder;
import com.xxx.fw.rocketmq.trx.core.TransactionLogService;
import javax.annotation.Resource;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.annotation.After;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.EnableTransactionManagement;
/**
* @Description TODO
* @Author 姚仲杰
* @Date 2020/12/28 11:29
*/
//此处需介绍下注解顺序，当前注解需包在事务注解中执行，注解先进后出所以让事务注解order为0 本注解顺序为1 则先进事务注解在进本注解
@EnableTransactionManagement(order = 0)
@Order(1)
@Component
@Aspect
public class TrxAspect {
public static final Logger LOGGER= LoggerFactory.getLogger(TrxAspect.class);
@Resource
TransactionLogService transactionLogService;
@Pointcut("@annotation(com.xxx.fw.rocketmq.trx.config.MqTrx)")
public void pointcut(){};
@After("pointcut() && @annotation(mqTrx)")
public void after(JoinPoint joinPoint, MqTrx mqTrx){
try {
String id = TrxContextHolder.getTrxId();
transactionLogService.insert(id);
LOGGER.info("事务消息日志入库成功，trx_id:[{}]", id);
}finally {
TrxContextHolder.remove();
}
}
}
mq事务上下文管理
package com.xxx.fw.rocketmq.trx.config;
import java.util.HashMap;
import java.util.Map;
/**
* @Description TODO
* @Author 姚仲杰
* @Date 2020/12/28 11:42
*/
public class TrxContext {
private ThreadLocal<Map<String,String>> threadLocal=new ThreadLocal<Map<String,String>>(){
@Override
protected Map<String, String> initialValue() {
return new HashMap<String, String>();
}
};
public String put(String key, String value) {
return threadLocal.get().put(key, value);
}
public String get(String key) {
return threadLocal.get().get(key);
}
public String remove(String key) {
return threadLocal.get().remove(key);
}
public Map<String, String> entries() {
return threadLocal.get();
}
}
package com.xxx.fw.rocketmq.trx.config;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @Description 此类用于传递事务id到切面中，即发送完成后执行本地事务前，将trx_id放入本地线程，然后通过切面把事务id写入到事务日志表
* @Author 姚仲杰
* @Date 2020/12/28 11:46
*/
public class TrxContextHolder {
private static final Logger LOGGER = LoggerFactory.getLogger(TrxContextHolder.class);
public static final TrxContext TRX_CONTEXT_HOLDER=new TrxContext();
public static final String TRX_ID="TRX_ID";
public static String getTrxId(){
return TRX_CONTEXT_HOLDER.get(TRX_ID);
}
public static void setTrxId(String trxId){
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("set trx_id:[{}]", trxId);
}
TRX_CONTEXT_HOLDER.put(TRX_ID, trxId);
}
public static String remove() {
String trxId = TRX_CONTEXT_HOLDER.remove(TRX_ID);
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("remove trx_id:[{}] ", trxId);
}
return trxId;
}
}
package com.xxx.fw.rocketmq.trx.core;
import javax.annotation.Resource;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
/**
* @Description 这里使用jdbcTemplate内置日志入库和查询操作让业务开发无需关注这些代码只需要添加一个注解即可
* @Author 姚仲杰
* @Date 2020/12/28 19:32
*/
@Component
public class TransactionLogService {
private final static String INSERT_TRX_LOG="insert into transaction_log (trx_id)value('%s')";
private final static String CHECK_TRX_LOG="select count(*) from transaction_log where trx_id='%s'";
@Resource
JdbcTemplate jdbcTemplate;
public int insert(String trxId){
if (StringUtils.isEmpty(trxId)){
throw new IllegalArgumentException("trxId must not be null");
}
int update = jdbcTemplate.update(String.format(INSERT_TRX_LOG, trxId));
return update;
}
public int query(String trxId){
if (StringUtils.isEmpty(trxId)){
throw new IllegalArgumentException("trxId must not be null");
}
int count = jdbcTemplate
.queryForObject(String.format(CHECK_TRX_LOG, trxId), int.class);
return count;
}
}
提供封装的发送服务以及工具类
package com.xxx.fw.rocketmq.trx.core;
import com.xxx.fw.rocketmq.trx.util.TagsUtil;
import javax.annotation.Resource;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
/**
* @Description 为了拼接tag实现不同业务
* @Author 姚仲杰
* @Date 2020/12/28 20:21
*/
@Component
public class TrxService {
@Resource
RocketMQTemplate rocketMQTemplate;
/**
* @param msg 发送给消息队列得消息
* @param tag 消息队列得tag，组装后形成trx_topic:tag
* @param o 要入库得数据
* @return
*/
public TransactionSendResult send(String msg,String tag,Object o){
Message message = MessageBuilder.withPayload(msg).build();
TransactionSendResult result = rocketMQTemplate
.sendMessageInTransaction(TagsUtil.bindTag(tag), message, o);
return result;
}
}
工具类
package com.xxx.fw.rocketmq.trx.util;
import org.springframework.util.StringUtils;
/**
* @Description 拼接tag
* @Author 姚仲杰
* @Date 2020/12/28 19:59
*/
public class TagsUtil {
public static final String TRX_TOPIC="trx_topic:";
public static String bindTag(String tag){
if (StringUtils.isEmpty(tag)){
throw new IllegalArgumentException("trx_tag must not be empty");
}
return TRX_TOPIC+tag;
}
}
网友评论