Spring Cloud Alibaba技术栈(下)

下载地址：http://seata.io/zh-cn/blog/download.html

<dependencies>
        <!--服务注册-->
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-alibaba-nacos-discovery</artifactId>
        </dependency>
        <!--Seata-->
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
        </dependency>
        <!--web 项目的基础依赖-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
    </dependencies>

public interface StorageService {

    /**
     * 完成对商品库存扣减操作
     * @param productNo
     * @param count
     */
    void deduct(String productNo, int count) ;
}


@Service
public class StorageServiceImpl implements StorageService {

    @Autowired
    private StorageTblDao storageTblDao;

    private static Logger logger = LoggerFactory.getLogger(StorageServiceImpl.class);

    @Transactional
    @Override
    public void deduct(String productNo, int count) {
        logger.info("开始扣减商品{}的库存, 数量为{}", productNo, count);
        //1.查询库存
        //StorageTbl storageTbl = storageTblDao.selectByPrimaryKey(Integer.parseInt(productNo));
        StorageTblExample storageTblExample = new StorageTblExample();
        storageTblExample.createCriteria().andCommodityCodeEqualTo(productNo);
        List<StorageTbl> storageTbls = storageTblDao.selectByExample(storageTblExample);
        StorageTbl storageTbl = storageTbls.get(0);

        if (storageTbl==null){
            throw new IllegalArgumentException("商品不存在");
        }
        //2.扣减操作(扣减后的金额)
        int idleCount =storageTbl.getCount()-count;
        if (idleCount<0){
            throw new RuntimeException("存库不足!");
        }
        //3.设置商品库存
        storageTbl.setCount(idleCount);
        //4.保存到数据库中
        storageTblDao.updateByPrimaryKeySelective(storageTbl);

        logger.info("扣减库存商品{}成功, 剩余的库存为{}",productNo, idleCount);
    }
}

@SpringBootApplication
@EnableDiscoveryClient
@MapperScan("ah.szxy.mapper")
@RestController
public class StorageServiceApplication {
    public static void main(String[] args) {
        SpringApplication.run(StorageServiceApplication.class,args);
    }

    @Autowired
    private StorageService storageService;

    //使用外部接口暴露
    @GetMapping("/deduct/{productNo}/{count}")
    public ResponseEntity<Void> deduct(@PathVariable("productNo") String productNo, @PathVariable("count") Integer count){
        storageService.deduct(productNo, count);
        return ResponseEntity.ok().build();
    }
}

server:
  port: 8093

spring:
  application:
    name: storage-service
  cloud:
    nacos:
      discovery:
        server-addr: 47.97.169.52:8848
    alibaba:
      seata:
        tx-service-group: ${spring.application.name}
  datasource:
    driver-class-name: com.mysql.cj.jdbc.Driver
    name: storageDataSource
    type: com.alibaba.druid.pool.DruidDataSource
    url: jdbc:mysql://119.45.189.14:3306/seata?useSSL=false&serverTimezone=UTC
    username: root
    password: root123
    druid:
      max-active: 20  # 最大连接数
      min-idle: 2     # 最小活跃数
      initial-size: 2 # 初始连接数
seata:
  service:
    vgroup-mapping:
      storage-service: default
    grouplist:
      default: 127.0.0.1:8091  # seata默认端口
    disable-global-transaction: false # 开启全局事务
  enabled: true  # 开启 seata

mybatis-plus:
  mapper-locations: classpath:/mapper/*.xml

http://localhost:8096/purchase/SXT_USER_2/HUAWEI_0001/1  //经测试,发生;额回滚
http://localhost:8096/purchase/SXT_USER_1/HUAWEI_0001/1  //未发生回滚

//这里我们选择 4.4.0 版本的原因在于，我们 spring cloud alibaba 版本为：2.2.0.RELEASE，它里面控制的 rocketMQ 的版是 4.4.0。
下载地址:  http://rocketmq.apache.org/release_notes/release-notes-4.4.0/

变量名: ROCKETMQ_HOME
变量值: D:\SoftWare\rocket-mq\rocketmq-all-4.4.0-bin-release //(RocketMq软件所在目录) 

logging.level.com.alibaba.cloud.stream.binder.rocketmq=DEBUG

#rocketmq 服务器 namerserver的地址
spring.cloud.stream.rocketmq.binder.name-server=127.0.0.1:9876

#stream->bindings->output(input)
#output1
#发送消息的目的地址
spring.cloud.stream.bindings.output1.destination=test-topic
#消息的默认类型
spring.cloud.stream.bindings.output1.content-type=application/json
#生产者组
spring.cloud.stream.rocketmq.bindings.output1.producer.group=binder-group
#消息的同步发送
spring.cloud.stream.rocketmq.bindings.output1.producer.sync=true

#output2 主要演示事务消息的发送
spring.cloud.stream.bindings.output2.destination=TransactionTopic
spring.cloud.stream.bindings.output2.content-type=application/json
#发送的是事务消息
spring.cloud.stream.rocketmq.bindings.output2.producer.transactional=true
spring.cloud.stream.rocketmq.bindings.output2.producer.group=myTxProducerGroup

#output3 用它演示消息的手动拉取
spring.cloud.stream.bindings.output3.destination=pull-topic
spring.cloud.stream.bindings.output3.content-type=text/plain
spring.cloud.stream.rocketmq.bindings.output3.producer.group=pull-binder-group

spring.application.name=rocketmq-produce-example

server.port=28081

management.endpoints.web.exposure.include=*
management.endpoint.health.show-details=always

package ah.szxy.listener;

import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;

@RocketMQTransactionListener(
        txProducerGroup = "myTxProducerGroup" ,
        corePoolSize = 2 ,
        maximumPoolSize = 5
)
public class RockerMQLocalTransactionListenerImpl implements RocketMQLocalTransactionListener {

    /**
     * 当我们发送半（half）消息成功后，mq 服务要求我们执行本地的事务，并且返回本地事务的执行结果
     * RocketMQLocalTransactionState：
     *  COMMIT 提交->其他的消费者将收到该消息
     *  ROLLBACK 回滚->mq ->半消息删除
     *  UNKNOWN -> mq 会再次检查本地的事务->checkLocalTransaction
     *
     * @param message
     * @param o
     * @return
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        String type = message.getHeaders().get("type").toString();
        switch (type){
            case "1" :
                System.out.println("本地事务执行状态未知");
                return RocketMQLocalTransactionState.UNKNOWN ;
            case "2":
                System.out.println("本地事务执行状态成功");
                return RocketMQLocalTransactionState.COMMIT ;
            case "3":
                System.out.println("本地事务执行状态失败");
                return RocketMQLocalTransactionState.ROLLBACK ;
        }
        return null ;
    }

    /**
     * 当mq 收到我们的本地的事务为UNKNOWN ,它会再次来检查我们的本地事务状态，要求返回一个本地事务的状态
     * @param message
     * @return
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        System.out.println("来检查了，本次我提交本地事务");
        return RocketMQLocalTransactionState.COMMIT;
    }
}

#rocketmq nameserver的地址
spring.cloud.stream.rocketmq.binder.name-server=127.0.0.1:9876

#stream->bindings->input

#input1
spring.cloud.stream.bindings.input1.destination=test-topic
spring.cloud.stream.bindings.input1.content-type=text/plain
spring.cloud.stream.bindings.input1.group=test-group1
spring.cloud.stream.rocketmq.bindings.input1.consumer.orderly=true


#input2
spring.cloud.stream.bindings.input2.destination=test-topic
spring.cloud.stream.bindings.input2.content-type=text/plain
spring.cloud.stream.bindings.input2.group=test-group2
spring.cloud.stream.rocketmq.bindings.input2.consumer.orderly=false
spring.cloud.stream.rocketmq.bindings.input2.consumer.tags=tagStr
spring.cloud.stream.bindings.input2.consumer.concurrency=20
spring.cloud.stream.bindings.input2.consumer.maxAttempts=1

#input3
spring.cloud.stream.bindings.input3.destination=test-topic
spring.cloud.stream.bindings.input3.content-type=application/json
spring.cloud.stream.bindings.input3.group=test-group3
spring.cloud.stream.rocketmq.bindings.input3.consumer.tags=tagObj
spring.cloud.stream.bindings.input3.consumer.concurrency=20

#input4
spring.cloud.stream.bindings.input4.destination=TransactionTopic
spring.cloud.stream.bindings.input4.content-type=text/plain
spring.cloud.stream.bindings.input4.group=transaction-group
spring.cloud.stream.bindings.input4.consumer.concurrency=5


#input5 手动消息的拉取
spring.cloud.stream.bindings.input5.destination=pull-topic
spring.cloud.stream.bindings.input5.content-type=text/plain
spring.cloud.stream.bindings.input5.group=pull-topic-group

spring.application.name=rocketmq-consume-example

server.port=28082

management.endpoints.web.exposure.include=*
management.endpoint.health.show-details=always