 

# 第1章    分布式事务 	  

# 1 关于分布式事务

### 3.3.1 导入数据库

```sql
CREATE DATABASE `stockdb`;
DROP TABLE IF EXISTS `t_stock`;
CREATE TABLE `t_stock` (
  `id` varchar(100) NOT NULL,
  `product_id` varchar(100) DEFAULT NULL,
  `product_name` varchar(200) DEFAULT NULL,
  `stock_id` varchar(100) DEFAULT NULL,
  `quantity` int(11) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
```

```sql
CREATE DATABASE `logdb`;
DROP TABLE IF EXISTS `t_log_info`;
CREATE TABLE `t_log_info` (
  `id` varchar(100) NOT NULL,
  `method` varchar(100) DEFAULT NULL,
  `action` varchar(500) DEFAULT NULL,
  `username` varchar(100) DEFAULT NULL,
  `create_time` date DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
```

### 3.3.2 导入工程

### 3.3.3 编写测试类和测试方法

```java
package com.itheima.test;


import com.itheima.domain.Stock;
import com.itheima.service.StockService;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import java.util.UUID;

/**
 * @author 黑马程序员
 * @Company http://www.itheima.com
 */
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:applicationContext.xml")
public class AtomikosTest {

    @Autowired
    private StockService stockService;

    @Test
    public void testSave(){
        //1.创建对象
        Stock stock = new Stock();
        //2.填充数据
        String id = UUID.randomUUID().toString().replace("-","").toUpperCase();
        stock.setId(id);
        stock.setProductId("1030102301111");
        stock.setProductName("华为p8");
        stock.setQuantity(100);
        //3.执行保存
        stockService.save(stock);
    }
}
```

### 3.3.4 编写atomikos核心代码

```xml
 <!--
        配置事务管理器atomikos事务管理器
     -->
    <bean id="atomikosTransactionManager" class="com.atomikos.icatch.jta.UserTransactionManager" init-method="init" destroy-method="close">
        <property name="forceShutdown" value="false"/>
    </bean>
    <!--
        本地事务管理器
    -->
    <bean id="atomikosUserTransaction" class="com.atomikos.icatch.jta.UserTransactionImp">
        <property name="transactionTimeout" value="300000"/>
    </bean>

    <!--JTA事务管理器-->
    <bean id="springTransactionManager" class="org.springframework.transaction.jta.JtaTransactionManager">
        <property name="transactionManager">
            <ref bean="atomikosTransactionManager"/>
        </property>
        <property name="userTransaction">
            <ref bean="atomikosUserTransaction"/>
        </property>
        <property name="allowCustomIsolationLevels" value="true"/>
    </bean>


    <!--数据源基础配置-->
    <bean id="abstractXADataSource" class="com.atomikos.jdbc.AtomikosDataSourceBean" init-method="init" destroy-method="close" abstract="true">
        <property name="xaDataSourceClassName" value="com.mysql.jdbc.jdbc2.optional.MysqlXADataSource"/>
        <property name="poolSize" value="10"/>
        <property name="minPoolSize" value="10"/>
        <property name="maxPoolSize" value="30"/>
        <property name="borrowConnectionTimeout" value="60"/>
        <property name="reapTimeout" value="20"/>
        <property name="maxIdleTime" value="60"/>
        <property name="maintenanceInterval" value="60"/>
        <property name="testQuery">
            <value>SELECT 1</value>
        </property>
    </bean>


    <!-- 数据库基本信息配置 -->
    <bean id="dataSourceStock" parent="abstractXADataSource">
        <property name="uniqueResourceName">
            <value>dataSourceStock</value>
        </property>
        <!--数据库驱动-->
        <property name="xaDataSourceClassName" value="com.mysql.jdbc.jdbc2.optional.MysqlXADataSource"/>
        <property name="xaProperties">
            <props>
                <prop key="URL">${jdbc.stock.url}</prop>
                <prop key="user">${jdbc.stock.username}</prop>
                <prop key="password">${jdbc.stock.password}</prop>
            </props>
        </property>
    </bean>

    <!--日志数据源-->
    <bean id="dataSourceLog" parent="abstractXADataSource">
        <property name="uniqueResourceName">
            <value>dataSourceLog</value>
        </property>
        <property name="xaDataSourceClassName" value="com.mysql.jdbc.jdbc2.optional.MysqlXADataSource"/>
        <property name="xaProperties">
            <props>
                <prop key="URL">${jdbc.log.url}</prop>
                <prop key="user">${jdbc.log.username}</prop>
                <prop key="password">${jdbc.log.password}</prop>
            </props>
        </property>
    </bean>


    <!--SqlSessionFactoryBean的配置-->
    <bean id="sqlSessionFactoryBeanStock" class="org.mybatis.spring.SqlSessionFactoryBean">
        <property name="typeAliasesPackage" value="com.itheima.domain" />
        <property name="dataSource" ref="dataSourceStock"/>
    </bean>

    <!--操作日志数据源的SqlSessionFactoryBean-->
    <bean id="sqlSessionFactoryBeanLog" class="org.mybatis.spring.SqlSessionFactoryBean">
        <property name="typeAliasesPackage" value="com.itheima.domain" />
        <property name="dataSource" ref="dataSourceLog"/>
    </bean>


    <!--包扫描,库存Dao包扫描-->
    <bean id="mapperScannerConfigurerStock" class="org.mybatis.spring.mapper.MapperScannerConfigurer">
        <property name="basePackage" value="com.itheima.dao.stock" />
        <property name="sqlSessionFactoryBeanName" value="sqlSessionFactoryBeanStock" />
    </bean>

    <!--包扫描,日志Dao包扫描-->
    <bean id="mapperScannerConfigurerLog" class="org.mybatis.spring.mapper.MapperScannerConfigurer">
        <property name="basePackage" value="com.itheima.dao.log" />
        <property name="sqlSessionFactoryBeanName" value="sqlSessionFactoryBeanLog" />
    </bean>
```





### 4.3.3 项目简介

#### 1）数据库表结构

```sql

DROP TABLE IF EXISTS `pay`;
CREATE TABLE `pay` (
  `id` varchar(100) NOT NULL,
  `ispay` int(11) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

INSERT INTO `pay` VALUES ('1234', '0');
```

### 4.4.2 导入坐标

#### 1）父工程坐标

```xml
<properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
    </properties>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.3.RELEASE</version>
        <relativePath/>
    </parent>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
    </dependencies>

    <!--锁定子模块使用的spring_cloud版本-->
    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>Greenwich.SR1</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <repositories>
        <repository>
            <id>spring-snapshots</id>
            <name>Spring Snapshots</name>
            <url>https://repo.spring.io/snapshot</url>
            <snapshots>
                <enabled>true</enabled>
            </snapshots>
        </repository>
        <repository>
            <id>spring-milestones</id>
            <name>Spring Milestones</name>
            <url>https://repo.spring.io/milestone</url>
            <snapshots>
                <enabled>false</enabled>
            </snapshots>
        </repository>
    </repositories>

    <pluginRepositories>
        <pluginRepository>
            <id>spring-snapshots</id>
            <name>Spring Snapshots</name>
            <url>https://repo.spring.io/snapshot</url>
            <snapshots>
                <enabled>true</enabled>
            </snapshots>
        </pluginRepository>
        <pluginRepository>
            <id>spring-milestones</id>
            <name>Spring Milestones</name>
            <url>https://repo.spring.io/milestone</url>
            <snapshots>
                <enabled>false</enabled>
            </snapshots>
        </pluginRepository>
    </pluginRepositories>
```

#### 2）支付服务坐标

```xml
<dependencies>
        <!--mybatis 依赖-->
        <dependency>
            <groupId>org.mybatis</groupId>
            <artifactId>mybatis</artifactId>
            <version>3.5.4</version>
        </dependency>
        <!--mybatis jar包依赖-->
        <dependency>
            <groupId>org.mybatis.spring.boot</groupId>
            <artifactId>mybatis-spring-boot-starter</artifactId>
            <version>1.3.0</version>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.47</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>druid</artifactId>
            <version>1.1.21</version>
        </dependency>
        <!--整合rokectmq-->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.0.3</version>
        </dependency>
    </dependencies>
```

#### 3）订单服务坐标

```xml
<dependencies>
        <!-- https://mvnrepository.com/artifact/org.apache.rocketmq/rocketmq-client -->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.7.0</version>
        </dependency>
        <!--整合rokectmq-->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.0.3</version>
        </dependency>
    </dependencies>
```

### 4.4.4 编写消息生产者-支付服务的Controller

```java
@RestController
public class PayController {
    @Resource
    private TransactionListener transactionListener;
    @RequestMapping(value = "/pay/updateOrder", method = RequestMethod.POST)
    public String payOrder(@RequestParam("payid") String id, @RequestParam("ispay") int ispay) {
        try {
            //创建事务消息消费者
            TransactionMQProducer transactionMQProducer = new TransactionMQProducer("txmessage_trans-client-group");
            //指定链接的服务器地址(nameserver)
            transactionMQProducer.setNamesrvAddr("127.0.0.1:9876");
            //创建消息回查的类,我们自己的监听器
            transactionMQProducer.setTransactionListener(transactionListener);


            //创建发送的消息
            Message message = new Message(
                    "txmessage_topic", "txmessage_tags", "txmessage_keys", "txmessage的消息".getBytes(RemotingHelper.DEFAULT_CHARSET)
            );
            //启动发送者
            transactionMQProducer.start();
            //发送消息
            //传递参数is和ispay
            Map payAgrs=new HashMap();
            payAgrs.put("id", id);
            payAgrs.put("ispay",ispay);
            transactionMQProducer.sendMessageInTransaction(message,payAgrs );
            //关闭消息的发送者
            transactionMQProducer.shutdown();
        } catch (Exception e) {
            e.printStackTrace();
            return "发送消息给mq失败!";
        }
        //如果没有问题,
        return "发送消息给mq成功";
    }

}
```

### 4.4.5 编写消息监听者-支付服务的Listener

```java
@Component
public class PayTransactionListener implements TransactionListener {

    //记录对应事务消息的执行状态   1:正在执行，2：执行成功，3：失败了
    //对于mq来说,正在事务发起方正在执行查询结果,只要未收到明确的commit或者rollback,都是未知结果unkown
    //对于mq来说,commit执行成功,才发送消息
    //对于mq来说,事务执行失败了将不再发送消息,并且将消息队列中的half消息干掉,以免再次扫描到再次回查
    //通过事务的id来辨别不同的事务

    private ConcurrentHashMap<String,Integer> transMap = new ConcurrentHashMap<String,Integer>();

//    @Resource
    @Autowired
    private PayService payService;
    /**
     * 消息发送方执行自身业务操作的方法
     * @param msg 发送方发送的东西
     * @param arg 额外的参数
     * @return
     */
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) throws RuntimeException {
            //业务代码写这里

        String transactionId = msg.getTransactionId();
        //设置执行状态为正在执行,state=1
        transMap.put(transactionId, 1);
        //取id和ispay参数
        Map payArgs= (Map) arg;
        String id= (String) payArgs.get("id");
        Integer ispay= (Integer) payArgs.get("ispay");
        Pay pay = new Pay();
        pay.setId(id);
        pay.setIspay(ispay);
        try {
            //控制本地事务
            System.out.println("支付表更新开始");
            payService.update(pay);
            System.out.println("支付表更新成功");
            //测试用例1
//            int i=1/0;
           // 测试用例2 测试网络超时状态
//            Thread.sleep(70000);
            System.out.println("更新订单状态");
            System.out.println("订单已更新");
            //执行成功时,返回提交事务消息成功的标识
            transMap.put(transactionId, 2);
//            if(1==1){
//                return  LocalTransactionState.UNKNOW;
//            }


        }catch (Exception e){
            e.printStackTrace();
            //发生异常时，返回回滚事务消息
            //执行成功时,返回提交事务消息成功的标识
            transMap.put(transactionId, 3);
            System.out.println("事务执行失败,事务执行状态为:"+ LocalTransactionState.ROLLBACK_MESSAGE);
            return  LocalTransactionState.ROLLBACK_MESSAGE;
        }
        System.out.println("事务执行成功,事务执行状态为:"+ LocalTransactionState.COMMIT_MESSAGE);
        return LocalTransactionState.COMMIT_MESSAGE;

    }




    /***
     * 事务超时，回查方法
     * @param msg:携带要回查的事务ID
     * @return
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        //根据transaction的id回查该事务的状态,并返回给消息队列
        //未知状态:查询事务状态,但始终无结果,或者由于网络原因发送不成功,对mq来说都是未知状态,LocalTransactionState.UNKNOW
        //正确提交返回LocalTransactionState.COMMIT_MESSAGE
        //事务执行失败返回LocalTransactionState.ROLLBACK_MESSAGE
        String transactionId = msg.getTransactionId();
        Integer state = transMap.get(transactionId);
        System.out.println("回查的事务id为:"+transactionId+",当前的状态为"+state);

        if (state==2){
            //执行成功,返回commit
            System.out.println("回查结果为事务正确提交,返回状态为:"+ LocalTransactionState.COMMIT_MESSAGE);
            return  LocalTransactionState.COMMIT_MESSAGE;

        }else if(state==3){
            //执行失败,返回rollback
            System.out.println("回查结果为事务回滚,返回状态为:"+ LocalTransactionState.ROLLBACK_MESSAGE);
            return  LocalTransactionState.ROLLBACK_MESSAGE;
        }

            //正在执行
            System.out.println("回查正在执行,返回状态为:"+ LocalTransactionState.UNKNOW);
            return  LocalTransactionState.UNKNOW;

    }
}
```

### 4.4.6 编写消息消费者-订单服务的启动类

```java
@SpringBootApplication
public class OrderApplication {

	public static void main(String[] args) throws MQClientException {
		SpringApplication.run(OrderApplication.class, args);
		//创建消息的消费者
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("txmessage_trans-client-group");
		//设置要链接的服务器地址(nameserver)
		consumer.setNamesrvAddr("127.0.0.1:9876");
		//设置单次消费的消息的数量
		consumer.setConsumeMessageBatchMaxSize(5);
		//设置消息消费的顺序
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
		//设置消费者监听哪些消息
		consumer.subscribe("txmessage_topic", "txmessage_tags");
		//进行消息的接收，并返回接收消息的结果
		consumer.registerMessageListener(new MessageListenerConcurrently() {
			@Override
			public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

				try {
					for(MessageExt mes :list){
						String topic = mes.getTopic();
						String tags = mes.getTags();
						String keys = mes.getKeys();
						String s = new String(mes.getBody(), "utf-8");
						String transactionId = mes.getTransactionId();
						System.out.println("接收到的transactionid:"+transactionId+", topic:"+topic+",tags:"+tags+",消息:"+s);

					}
				}catch( Exception e){
					e.printStackTrace();
				}

				return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
			}
		});
		//启动消费者
		System.out.println("启动完成");
		consumer.start();
	}
}
```

