1. Oinone社区首页
  2. Oinone 7天入门到精通

4.1.21 框架之分布式消息

史, 昂 • • Oinone 7天入门到精通 • 阅读 962

消息中间件是在分布式开发中常见的一种技术手段，用于模块间的解耦、异步处理、数据最终一致等场景。

一、介绍

oinone对开源的RocketMQ进行了封装，是平台提供的一种较为简单的使用方式，并非是对RocketMQ进行的功能扩展。同时也伴随着两个非常至关重要的目的：

  1. 适配不同企业对RocketMQ的不同版本选择，不至于改上层业务代码。目前已经适配RocketMQ的开源版本和阿里云版本。 下个版本会对API进行升级支持不同类型MQ，以适配不同企业对MQ的不同要求，应对一些企业客户已经对MQ进行技术选择

  2. 对协议头进行扩展：如多租户的封装，saas模式中为了共用MQ基础资源，需要在消息头中加入必要租户信息。

二、使用准备

demo工程默认已经依赖消息，这里只是做介绍无需大家额外操作，大家可以用maven依赖树命令查看引用关系。

依赖包

增加对pamirs-connectors-event的依赖


<dependency>
    <groupId>pro.shushi.pamirs.framework</groupId>
    <artifactId>pamirs-connectors-event</artifactId>
</dependency>

图4-1-21-1 分布式消息的依赖包

相关功能引入

增加模型、触发器都依赖MQ

<!-- 增强模型 -->
<!-- 增强模型 -->
<dependency>
    <groupId>pro.shushi.pamirs.core</groupId>
    <artifactId>pamirs-channel</artifactId>
</dependency>
<!-- 触发器 api -->
<dependency>
    <groupId>pro.shushi.pamirs.core</groupId>
    <artifactId>pamirs-trigger-api</artifactId>
</dependency>
<!-- 触发器 core -->
<dependency>
    <groupId>pro.shushi.pamirs.core</groupId>
    <artifactId>pamirs-trigger-core</artifactId>
</dependency>

图4-1-21-2 增加模型、触发器都依赖MQ

yml配置文件参考

详见4.1.1【模块之yml文件结构详解】的“pamirs.event”部分。

三、使用说明

发送消息（NotifyProducer）

概述

NotifyProducer是Pamirs Event中所有生产者的基本API，它仅仅定义了消息发送的基本行为，例如生产者自身的属性，启动和停止，当前状态，以及消息发送方法。它本身并不决定消息如何发送，而是根据具体的实现确定其功能。

目前仅实现了RocketMQProducer，你可以使用下面介绍的方法轻松使用这些功能。

使用方法

Notify注解方式

使用示例

@Component
public class DemoProducer {

    @Notify(topic = "test", tags = "model")
    public DemoModel sendModel() {
        return new DemoModel();
    }

    @Notify(topic = "test", tags = "dto")
    public DemoDTO sendDTO() {
        return new DemoDTO();
    }
}

图4-1-21-3 Notify注解方式使用示例

解释说明

  1. 使用Component注解方式注册Spring Bean。

  2. Notify注解指定topic和tags。

  3. topic和tags对应NotifyEvent中的topic和tags。

RocketMQProducer方法调用

使用示例

@Component
public class SendRocketMQMessage {

    @Autowired
    private RocketMQProducer rocketMQProducer;

    /**
     * 发送普通消息
     */
    public void sendNormalMessage() {
        rocketMQProducer.send(new NotifyEvent("test", "model", new DemoModel()));
        rocketMQProducer.send(new NotifyEvent("test", "dto", new DemoDTO()));
    }

    /**
     * 发送有序消息
     */
    public void sendOrderlyMessage() {
        DemoModel data = new DemoModel();
        data.setAge(10);
        rocketMQProducer.send(new NotifyEvent("test", "model", data)
                .setQueueSelector((queueSize, event) -> {
                    DemoModel body = (DemoModel) event.getBody();
                    return body.getAge() % queueSize;
                }));
    }

    /**
     * 发送事务消息
     */
    public void sendTransactionMessage() {
        rocketMQProducer.send(new NotifyEvent("test", "model", new DemoModel())
                .setIsTransaction(true)
                .setGroup("demoTransactionListener"));
    }
}

图4-1-21-4 RocketMQProducer方法调用

解释说明

  1. 使用Component注解方式注册Spring Bean。
  2. 使用Autowired注解方式装配RocketMQProducer实例。
  3. 使用send方法发送指定消息。
  4. 在【发送普通消息】方法中，该实现效果与Notify注解方式所示完全一致。
  5. 在【发送有序消息】方法中，队列选择器是必须配置的，queueSize属性为MQ队列的总数量，在broker中配置。有序消息必须配合有序消费者才能达到有序消费的目的，否则还是无序的常规消息，消费者需要配置@NotifyListener(consumerType=ConsumerType.ORDERLY )注解。
  6. 在【发送事务消息】方法中指定的group为ProducerGroup，事务消息是通过不同的Producer发出的，事务消息监听请参考TransactionListener注解的相关使用方法。（示例中的group与下文介绍中的一致）

使用TransactionListener开启事务消息监听

使用示例

@Component
@TransactionListener
public class DemoTransactionListener implements NotifyTransactionListener {

    @Override
    public NotifyTransactionState checkLocalTransaction(NotifyEvent event) {
        return NotifyTransactionState.COMMIT;
    }

}

图4-1-21-5 使用TransactionListener开启事务消息监听

解释说明

  1. 实现NotifyTransactionListener接口。

  2. 使用Component注解方式注册Spring Bean。

  3. 添加TransactionListener注解注册事务监听，生成对应的生产者。

  4. 当前ProducerGroup将使用这个类的BeanName，即"demoTransactionListener"。如果你想自定义ProducerGroup，可以使用TransactionListener的value属性进行设置。

消费消息（NotifyConsumer）

概述

NotifyConsumer是Pamirs Evnet中所有消费者的基本API，与NotifyProducer类似，它仅仅定义了消息消费的基本行为，例如消费者自身的属性，启动和停止，当前状态，以及消息的监听和订阅方法。它本身并不决定消息如何消费以及如何被订阅，而是根据具体的实现确定其功能。

目前仅实现了RocketMQEventPushConsumer，你可以使用下面介绍的方法轻松使用这些功能。

使用方法

在类上使用NotifyListener注解

使用示例

@Component
@NotifyListener(topic = "test", tags = "model")
public class DemoConsumerClass implements NotifyEventListener {

    @Override
    public void consumer(NotifyEvent event) {
        DemoModel data = (DemoModel) event.getBody();
        // do some things.
    }
}

图4-1-21-6 在类上使用NotifyListener注解

解释说明

  1. 实现NotifyEventListener接口。

  2. 使用Component注解方式注册Spring Bean。

  3. 当前ConsumerGroup将使用这个类的BeanName，即"demoConsumerClass"。如果你想自定义ConsumerGroup，可以使用Component的value属性进行设置。

  4. 从body中将可以获取生产者发送的数据对象，并且已经做好了类型处理，可以直接使用。

  5. 使用原生的RocketMQ发送的消息，类型可能是无法识别的，你可以使用NotifyListener中提供的bodyClass来指定类型。

  6. topic和tags对应NotifyEvent中的topic和tags。

在方法上使用NotifyListener注解

使用示例

@Component
public class DemoConsumerMethod {

    @Bean
    @NotifyListener(topic = "test", tags = "model")
    public NotifyEventListener modelConsumer() {
        return event -> {
            DemoModel data = (DemoModel) event.getBody();
            // do some things.
        };
    }

    @Bean
    @NotifyListener(topic = "test", tags = "dto")
    public NotifyEventListener dtoConsumer() {
        return event -> {
            DemoDTO data = (DemoDTO) event.getBody();
            // do some things.
        };
    }
}

图4-1-21-7 在方法上使用NotifyListener注解

解释说明

  1. 使用Bean注解方式注册Spring Bean。

  2. 方法返回值为NotifyEventListener类型。

  3. 当前ConsumerGroup将使用对应方法生成的BeanName，即"modelConsumer"和"dtoConsumer"。如果你想自定义ConsumerGroup，可以使用Bean的value属性进行设置。

实战

约定：每个模块下的Topic和Tags必须定义常量池进行统一管理，主要是为了方便维护与管理，技术没有限制

常规消息（举例）

Step1 新建PetNotifyEnum

用PetNotifyEnum来管理模块的所有Topic和Tags的常量定义

package pro.shushi.pamirs.demo.api.mq;

public class PetNotifyEnum {

    public static class PetItemInventroy{

        public interface Topic{
            String PET_ITEM_INVENTROY_CHANGE ="petItemInventroyChange";
        }
        public interface Tag{
            String CREATE ="create";
            String UPDATE ="update";
            String DELETE ="delete";
        }

    }

}

图4-1-21-8 新建PetNotifyEnum

Step2 新建PetItemInventoryMqProducer

新建PetItemInventroy模型对应消息生产者，用于发送PetItemInventroy模型变动的相关消息

package pro.shushi.pamirs.demo.core.mq;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import pro.shushi.pamirs.demo.api.model.PetItemInventroy;
import pro.shushi.pamirs.demo.api.mq.PetNotifyEnum;
import pro.shushi.pamirs.framework.connectors.event.engine.NotifyEvent;
import pro.shushi.pamirs.framework.connectors.event.rocketmq.RocketMQProducer;

@Component
public class PetItemInventoryMqProducer {

    @Autowired
    private RocketMQProducer rocketMQProducer;

    /**
     * 发送普通消息
     */
    public void sendNormalMessage(PetItemInventroy data,String tag) {
        rocketMQProducer.send(new NotifyEvent(PetNotifyEnum.PetItemInventroy.Topic.PET_ITEM_INVENTROY_CHANGE, tag, data));
    }

}

图4-1-21-9 新建PetItemInventoryMqProducer

Step3 新建PetItemInventoryMqConsumer

新建PetItemInventoryMqConsumer，订阅PetItemInventroy模型变动的相关消息，并进行相关处理

package pro.shushi.pamirs.demo.core.mq;

import org.springframework.stereotype.Component;
import pro.shushi.pamirs.demo.api.model.PetItemInventroy;
import pro.shushi.pamirs.demo.api.mq.PetNotifyEnum;
import pro.shushi.pamirs.framework.connectors.event.annotation.NotifyListener;
import pro.shushi.pamirs.framework.connectors.event.api.NotifyEventListener;
import pro.shushi.pamirs.framework.connectors.event.engine.NotifyEvent;
import pro.shushi.pamirs.meta.annotation.fun.extern.Slf4j;

@Component
@NotifyListener(topic = PetNotifyEnum.PetItemInventroy.Topic.PET_ITEM_INVENTROY_CHANGE, tags = "*")
@Slf4j
public class PetItemInventoryMqConsumer implements NotifyEventListener {
    @Override
    public void consumer(NotifyEvent event) {
        PetItemInventroy petItemInventroy = (PetItemInventroy)event.getBody();
        log.info("petItemInventroy的消息, 库存数量为：" + petItemInventroy.getQuantity());
    }
}

图4-1-21-10 PetItemInventoryMqConsumer

Step4 修改PetItemInventroyAction

在修改PetItemInventroy完成之后，发送topic为【PetNotifyEnum.PetItemInventroyMq.Topic.PET_ITEM_INVENTROY_CHANGE】，tag为【PetNotifyEnum.PetItemInventroyMq.Tag.UPDATE】的消息出去

package pro.shushi.pamirs.demo.core.action;
……引入依赖类
@Model.model(PetItemInventroy.MODEL_MODEL)
@Component
public class PetItemInventroyAction {

    @Autowired
    private PetItemInventoryMqProducer petItemInventoryMqProducer;

    @Function.Advanced(type= FunctionTypeEnum.UPDATE)
    @Function.fun(FunctionConstants.update)
    @Function(openLevel = {FunctionOpenEnum.API})
    public PetItemInventroy update(PetItemInventroy data){
        ……其他代码
        petItemInventoryMqProducer.sendNormalMessage(data, PetNotifyEnum.PetItemInventroy.Tag.UPDATE);
        return data;
    }
}

图4-1-21-11 修改PetItemInventroyAction

Step5 重新看效果

  1. 编辑商品库存记录

image.png

图4-1-21-12 编辑商品库存记录

  1. 查看后端日志是否打印

image.png

图4-1-21-13 查看后端日志是否打印

顺序消息（举例）

Step1 修改PetItemInventoryMqProducer

增加一个顺序消息发送的方法，发送时根据商品库存的Id就分队列，相同队列顺序消费

package pro.shushi.pamirs.demo.core.mq;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import pro.shushi.pamirs.demo.api.model.PetItemInventroy;
import pro.shushi.pamirs.demo.api.mq.PetNotifyEnum;
import pro.shushi.pamirs.framework.connectors.event.engine.NotifyEvent;
import pro.shushi.pamirs.framework.connectors.event.rocketmq.RocketMQProducer;

@Component
public class PetItemInventoryMqProducer {

    @Autowired
    private RocketMQProducer rocketMQProducer;

    /**
     * 发送普通消息
     */
    public void sendNormalMessage(PetItemInventroy data,String tag) {
        rocketMQProducer.send(new NotifyEvent(PetNotifyEnum.PetItemInventroy.Topic.PET_ITEM_INVENTROY_CHANGE, tag, data));
    }

    /**
     * 发送有序消息
     */
    public void sendOrderlyMessage(PetItemInventroy data,String tag) {

        rocketMQProducer.send(new NotifyEvent(PetNotifyEnum.PetItemInventroy.Topic.PET_ITEM_INVENTROY_CHANGE, tag, data)
                .setQueueSelector((queueSize, event) -> {
                    PetItemInventroy body = (PetItemInventroy) event.getBody();
                    Long queue = body.getId().longValue() % Long.valueOf(queueSize);
                    return queue.intValue();
                }));

    }
}

图4-1-21-14 修改PetItemInventoryMqProducer

Step2 修改PetItemInventoryMqConsumer

增加@NotifyListener(consumerType= ConsumerType.ORDERLY)的注解

package pro.shushi.pamirs.demo.core.mq;

import org.springframework.stereotype.Component;
import pro.shushi.pamirs.demo.api.model.PetItemInventroy;
import pro.shushi.pamirs.demo.api.mq.PetNotifyEnum;
import pro.shushi.pamirs.framework.connectors.event.annotation.NotifyListener;
import pro.shushi.pamirs.framework.connectors.event.api.NotifyEventListener;
import pro.shushi.pamirs.framework.connectors.event.engine.NotifyEvent;
import pro.shushi.pamirs.framework.connectors.event.enumeration.ConsumerType;
import pro.shushi.pamirs.meta.annotation.fun.extern.Slf4j;

@Component
//@NotifyListener(topic = PetNotifyEnum.PetItemInventroy.Topic.PET_ITEM_INVENTROY_CHANGE, tags = "*")
@NotifyListener(topic = PetNotifyEnum.PetItemInventroy.Topic.PET_ITEM_INVENTROY_CHANGE, tags = "*",consumerType= ConsumerType.ORDERLY)
@Slf4j
public class PetItemInventoryMqConsumer implements NotifyEventListener {
    @Override
    public void consumer(NotifyEvent event) {
        PetItemInventroy petItemInventroy = (PetItemInventroy)event.getBody();
        log.info("petItemInventroy的消息, 库存数量为：" + petItemInventroy.getQuantity());
    }
}

图4-1-21-15 修改PetItemInventoryMqProducer

Step3 修改PetItemInventroyAction

调用petItemInventoryMqProducer的顺序消息发送接口

package pro.shushi.pamirs.demo.core.action;
……引入依赖类
@Model.model(PetItemInventroy.MODEL_MODEL)
@Component
public class PetItemInventroyAction {

    @Autowired
    private PetItemInventoryMqProducer petItemInventoryMqProducer;

    @Function.Advanced(type= FunctionTypeEnum.UPDATE)
    @Function.fun(FunctionConstants.update)
    @Function(openLevel = {FunctionOpenEnum.API})
    public PetItemInventroy update(PetItemInventroy data){
        ……其他代码
//        petItemInventoryMqProducer.sendNormalMessage(data, PetNotifyEnum.PetItemInventroy.Tag.UPDATE);
        petItemInventoryMqProducer.sendOrderlyMessage(data, PetNotifyEnum.PetItemInventroy.Tag.UPDATE);
        return data;
    }
}

图4-1-21-16 修改PetItemInventroyAction

Step4 重启看效果

同常规消息

事务消息（举例）

注：这种写法默认第一次是UNKNOWN，然后通过MQ回调二次确认，在消息时效性要求高的场景下是不符合要求的。对实效性要求高的，请见下面事务消息-优化（举例）章节。

Step1 新建NotifyTransactionListener

package pro.shushi.pamirs.demo.core.mq;

import org.springframework.stereotype.Component;
import pro.shushi.pamirs.framework.connectors.event.annotation.TransactionListener;
import pro.shushi.pamirs.framework.connectors.event.api.NotifyTransactionListener;
import pro.shushi.pamirs.framework.connectors.event.engine.NotifyEvent;
import pro.shushi.pamirs.framework.connectors.event.enumeration.NotifyTransactionState;
import pro.shushi.pamirs.meta.annotation.fun.extern.Slf4j;

@Component
@Slf4j
@TransactionListener
public class PetItemInventoryMqTransactionListener implements NotifyTransactionListener {

    @Override
    public NotifyTransactionState checkLocalTransaction(NotifyEvent event) {
        log.info("消息回滚");
        return NotifyTransactionState.ROLLBACK;
    }
}

图4-1-21-17 新建NotifyTransactionListener

Step2 修改PetItemInventoryMqProducer

增加发送事务性消息的方法，通过.setIsTransaction(true)来显示设置该消息是事务消息，通过setGroup("petItemInventoryMqTransactionListener")来匹配事务监听处理器

/**
 * 发送事务消息
 */
public void sendTransactionMessage(PetItemInventroy data,String tag) {

    rocketMQProducer.send(new NotifyEvent(PetNotifyEnum.PetItemInventroy.Topic.PET_ITEM_INVENTROY_CHANGE, tag, data)
            .setQueueSelector((queueSize, event) -> {
                PetItemInventroy body = (PetItemInventroy) event.getBody();
                Long queue = body.getId().longValue() % Long.valueOf(queueSize);
                return queue.intValue();
            })
            .setIsTransaction(true)
            .setGroup("petItemInventoryMqTransactionListener")
    );
}

图4-1-21-18 修改PetItemInventoryMqProducer

Step3 修改PetItemInventroyAction

调用petItemInventoryMqProducer的事务性消息发送接口

package pro.shushi.pamirs.demo.core.action;
……引入依赖类
@Model.model(PetItemInventroy.MODEL_MODEL)
@Component
public class PetItemInventroyAction {

    @Autowired
    private PetItemInventoryMqProducer petItemInventoryMqProducer;

    @Function.Advanced(type= FunctionTypeEnum.UPDATE)
    @Function.fun(FunctionConstants.update)
    @Function(openLevel = {FunctionOpenEnum.API})
    public PetItemInventroy update(PetItemInventroy data){
        ……其他代码
        //petItemInventoryMqProducer.sendNormalMessage(data, PetNotifyEnum.PetItemInventroy.Tag.UPDATE);
        //petItemInventoryMqProducer.sendOrderlyMessage(data, PetNotifyEnum.PetItemInventroy.Tag.UPDATE);
        petItemInventoryMqProducer.sendTransactionMessage(data, PetNotifyEnum.PetItemInventroy.Tag.UPDATE);
        return data;
    }
}

图4-1-21-19 修改PetItemInventroyAction

Step4 重启看效果

  1. 编辑商品库存记录

image.png

图4-1-21-20 编辑商品库存记录

  1. 查看后端日志是否打印

消息回滚，没有消息消费的日志，从日志打印时间上看MQ回调会有延迟

image.png

图4-1-21-21 MQ日志打印一

image.png

图4-1-21-22 MQ日志打印二

Step5 修改NotifyTransactionListener

package pro.shushi.pamirs.demo.core.mq;

import org.springframework.stereotype.Component;
import pro.shushi.pamirs.framework.connectors.event.annotation.TransactionListener;
import pro.shushi.pamirs.framework.connectors.event.api.NotifyTransactionListener;
import pro.shushi.pamirs.framework.connectors.event.engine.NotifyEvent;
import pro.shushi.pamirs.framework.connectors.event.enumeration.NotifyTransactionState;
import pro.shushi.pamirs.meta.annotation.fun.extern.Slf4j;

@Component
@Slf4j
@TransactionListener
public class PetItemInventoryMqTransactionListener implements NotifyTransactionListener {

    @Override
    public NotifyTransactionState checkLocalTransaction(NotifyEvent event) {
        log.info("消息提交");
        //正常业务情况下，这里要增加自己的逻辑判断，来确定返回状态值
        return NotifyTransactionState.COMMIT;
    }
}

图4-1-21-23 修改NotifyTransactionListener

Step6 重启看效果

消息提交，并看到消息消费的日志，从日志打印时间上看MQ回调会有延迟

image.png

图4-1-21-24 MQ日志打印三

image.png

图4-1-21-25 MQ日志打印三

事务消息-优化（举例）

Step1 修改PetItemInventoryMqProducer

修改PetItemInventoryMqProducer事务性消息发送方法sendTransactionMessage，增加NotifyExecuteLocalTransactionCallback入参。

/**
 * 发送事务消息
 */
public void sendTransactionMessage(PetItemInventroy data, String tag, NotifyExecuteLocalTransactionCallback callback) {

    rocketMQProducer.send(new NotifyEvent(PetNotifyEnum.PetItemInventroy.Topic.PET_ITEM_INVENTROY_CHANGE, tag, data)
            .setQueueSelector((queueSize, event) -> {
                PetItemInventroy body = (PetItemInventroy) event.getBody();
                Long queue = body.getId().longValue() % Long.valueOf(queueSize);
                return queue.intValue();
            })
            .setIsTransaction(true)
            .setGroup("petItemInventoryMqTransactionListener")
            .setExecuteLocalTransactionCallback(callback)
    );
}

图4-1-21-26 修改PetItemInventoryMqProducer

Step2 修改PetItemInventroyAction

修改PetItemInventroyAction的update方法

package pro.shushi.pamirs.demo.core.action;
……引入依赖类
@Model.model(PetItemInventroy.MODEL_MODEL)
@Component
public class PetItemInventroyAction {

    @Autowired
    private PetItemInventoryMqProducer petItemInventoryMqProducer;

    @Function.Advanced(type= FunctionTypeEnum.UPDATE)
    @Function.fun(FunctionConstants.update)
    @Function(openLevel = {FunctionOpenEnum.API})
    public PetItemInventroy update(PetItemInventroy data){
        //petItemInventoryMqProducer.sendNormalMessage(data, PetNotifyEnum.PetItemInventroy.Tag.UPDATE);
        //petItemInventoryMqProducer.sendOrderlyMessage(data, PetNotifyEnum.PetItemInventroy.Tag.UPDATE);
//        petItemInventoryMqProducer.sendTransactionMessage(data, PetNotifyEnum.PetItemInventroy.Tag.UPDATE);
        petItemInventoryMqProducer.sendTransactionMessage(data, PetNotifyEnum.PetItemInventroy.Tag.UPDATE,new NotifyExecuteLocalTransactionCallback(){
            @Override
            public NotifyTransactionState callback(NotifyTransactionState executeState, NotifyEvent event) {

                List<PetItemInventroy> inventroys = new ArrayList<>();
                inventroys.add(data);
                PamirsSession.directive().disableOptimisticLocker();
                try{
                    int i = data.updateBatch(inventroys);
                } finally {
                    PamirsSession.directive().enableOptimisticLocker();
                }
                //业务代码执行完毕，提交消息
                return NotifyTransactionState.COMMIT;
            }
        });
        return data;
    }
}

图4-1-21-27 修改PetItemInventroyAction

Step3 重启看效果

编辑商品库存记录，发现消息的处理几乎没有延迟

image.png

图4-1-21-28 示例效果几乎没有延迟

Step4 模拟业务成功，消息发送失败

修改PetItemInventroyAction的update方法

@Function.Advanced(type= FunctionTypeEnum.UPDATE)
@Function.fun(FunctionConstants.update)
@Function(openLevel = {FunctionOpenEnum.API})
public PetItemInventroy update(PetItemInventroy data){
    //petItemInventoryMqProducer.sendNormalMessage(data, PetNotifyEnum.PetItemInventroy.Tag.UPDATE);
    //petItemInventoryMqProducer.sendOrderlyMessage(data, PetNotifyEnum.PetItemInventroy.Tag.UPDATE);
//        petItemInventoryMqProducer.sendTransactionMessage(data, PetNotifyEnum.PetItemInventroy.Tag.UPDATE);
    petItemInventoryMqProducer.sendTransactionMessage(data, PetNotifyEnum.PetItemInventroy.Tag.UPDATE,new NotifyExecuteLocalTransactionCallback(){

        @Override
        public NotifyTransactionState callback(NotifyTransactionState executeState, NotifyEvent event) {

            List<PetItemInventroy> inventroys = new ArrayList<>();
            inventroys.add(data);
            PamirsSession.directive().disableOptimisticLocker();
            try{
                int i = data.updateBatch(inventroys);
            } finally {
                PamirsSession.directive().enableOptimisticLocker();
            }
            //模拟业务成功，消息发送失败
            throw new RuntimeException();
            //业务代码执行完毕，提交消息
            //return NotifyTransactionState.COMMIT;
        }
    });
    return data;
}

图4-1-21-29 修改PetItemInventroyAction的update方法

Step5 重启看效果

这个效果跟第一种事务消息一样，消息提交延迟严重。

image.png

图4-1-21-30 MQ日志打印

image.png

图4-1-21-31 MQ日志打印

Step6 总结

优化后的事务消息，在正常情况下时效性非常高，异常情况下也能通过NotifyTransactionListener做保障

Oinone社区 作者：史, 昂原创文章，如若转载，请注明出处：https://doc.oinone.top/oio4/9296.html

访问Oinone官网:https://www.oinone.top获取数式Oinone低代码应用平台体验

赞 (0)
史, 昂的头像史, 昂数式管理员
0 0
上一篇 2024年5月23日
下一篇 2024年5月23日

相关推荐

  • 第2章 Oinone的技术独特性

    本章的主要目的是通过分析企业商业支撑软件的项目特性和关注点，找到企业软件发展的另一个本质变化——新技术流派的产生。在对“互联网架构做为最佳实践为何失效”的思考基础上，我们分析互联网中台架构的发展历史以及企业实际现状，找出其水土不服的原因。进而引出Oinone的低代码开发平台如何结合互联网架构并完成创新，以满足企业数字化转型的需求。 具体而言，本章包括以下内容： 企业软件发展的另一个本质变化：新技术流派的产生； 最佳实践为何失效？Oinone如何打造具有企业特色的互联网架构； Oinone独特性之源：元数据与设计原则； Oinone独特性之单体与分布式的灵活切换； Oinone独特性之每一个需求都是一个模块； Oinone独特性之低无一体。

    史, 昂的头像 史, 昂
    Oinone 7天入门到精通 2024年5月23日
    1.4K00
  • 3.5.7.9 自定义多Tab Oinone 7天入门到精通

    3.5.7.9 自定义多Tab

    在业务中，可能会遇到需要对多tab的交互或UI做全新设计的需求，这个时候可以自定义多tab组件支持。 首先继承平台的MultiTabsWidget组件，将自己vue文件设置到component处 import { MultiTabsWidget, SPI, ViewWidget } from '@kunlun/dependencies'; import Component from './CustomMultiTabs.vue'; @SPI.ClassFactory( ViewWidget.Token({ // 这里的tagName跟平台的组件一样，这样才能覆盖平台的组件 tagName: ['multi-tabs', 'MultiTabs'] }) ) export class CustomMultiTabsWidget extends MultiTabsWidget{ public initialize(props) { super.initialize(props); // 设置自定义的vue组件 this.setComponent(Component); return this; } } vue文件中继承平台的props，编写自定义页面代码 export const MultiTabsProps = { /** * 组件是否可见 */ invisible: { type: Boolean }, /** * tab列表数据 */ tabs: { type: Array as PropType<MultiTab[]> }, /** * 当前激活的tab */ activeTab: { type: Object as PropType<MultiTab> }, /** * 鼠标悬浮所在的tab */ hoverTab: { type: Object as PropType<MultiTab> }, /** * 鼠标经过tab事件回调 */ onMouseenterTab: { type: Function as PropType<(tab: MultiTab) => void> }, /** * 鼠标离开tab事件回调 */ onMouseleaveTab: { type: Function as PropType<(tab: MultiTab) => void> }, /** * 点击tab */ onClickTab: { type: Function as PropType<(tab: MultiTab) => void> }, /** * 刷新当前tab */ onRefreshCurrentTab: { type: Function as PropType<(tab: MultiTab) => void> }, /** * 关闭当前tab */ onCloseCurrentTab: { type: Function as PropType<(tab: MultiTab) => void> }, /** * 关闭除当前tab外的其他所有tab */ onCloseOtherTabs: { type: Function as PropType<(tab: MultiTab) => void> }, /** * 关闭当前tab左侧的所有tab */ onCloseLeftTabs: { type: Function as…

    史, 昂的头像 史, 昂
    2024年5月23日
    1.1K00
  • 工作流 Oinone 7天入门到精通

    工作流

    1. 查看、处理流程 1.1 流程查看 流程管理页面共同点： 选项分类筛选 标签筛选 应用下拉选筛选 根据流程名称搜索 流程管理页面名词解释： 任务待办：当前登录用户未处理的流程节点 我发起的：当前登录用户人为触发的流程（模型触发） 抄送：抄送给当前登录用户的节点（审批/填写） 我已办结：由当前登录用户完成人工/自动同意、人工拒绝或人工填写的节点 无需办理：当前登录用户转交的任务/被退回、被撤销、被或签、被其他分支任务拒绝的还未办理的任务 1.2 流程处理 1.2.1 任务待办 任务待办中点击“审批/填写”会进入流程详情处理页面，主要展示 1. 操作区 2. 流程发起人及状态 3. 模型视图内容 4. 流程时间线及其他记录。 审批代办操作区可能包含“分享、同意、拒绝、退回、加签、转交、返回”，填写代办操作区可能包含“分享、转交、提交、暂存、返回”，审批/填写操作区包含哪些动作由流程设计决定。 1.2.2 我发起的 我发起的流程列表中主要分为进行中和已完成的流程。进行中的流程可以进行查看、催办、撤销的操作，已完成的流程可以进行查看操作。 查看我发起的流程，进入流程详情页面，也是根据流程状态展示对应操作功能，进行中的流程有分享、催办、撤销、返回按钮，已完成的流程有分享、返回按钮。 1.2.3 抄送 抄送列表中每条抄送只可以进行查看操作，查看进入流程的详情页面，有分享和返回的操作。 1.2.4 我已办结 我已办结列表中可以进行查看操作，查看进入代办的详情页面，可以进行分享和返回的操作。 1.2.5 无需办理 无需办理列表中可以进行查看操作，查看进入代办的详情页面，可以进行分享和返回的操作。 2. 流程运行记录查看 所有运行流程都会记录在流程运行记录中，可以根据流程的所属应用，流程名称，触发方式和状态进行搜索，流程运行记录详情中展示流程运行的具体节点，运行时间，当前运行节点，异常信息等。

    史, 昂的头像 史, 昂
    2024年6月20日
    1.2K00

  • 致Oinone读者

    欢迎来到Oinone生态，我们为您提供了一站式低代码商业支撑平台，数式科技已经用其服务了如中烟、得力、上海电气、中航金网、雾芯科技等多个知名企业，我们的技术实力在商业场景得到了很好的验证。我们希望通过开源Oinone项目，为中国软件行业带来变革，提升整体工程化水平，与广大软件工程师一起为客户创造价值！《Oinone 7天从入门到精通》一书是Oinone开源项目的配套书籍，系统化地介绍了如何基于Oinone开源项目，快速开发高质量的软件系统。此外，如果您开发的项目是用于商业化而非企业自用，我们还为您提供了一种可选的商业化变现途径：申请成为Oinone的合作伙伴，并提交相关产品，详情请访问 www.oinone.top 网站。 书籍纲领 本书的章节安排如下： 第一章至第二章：【揭开面纱，理解Oinone】和【Oinone的技术独特性】。这两个章节可以帮助您更好地理解我们设计Oinone的初衷以及特性的由来。 第三章：面向研发人员的【Oinone的基础入门】。如果您是专业的研发人员，本章可以帮助您快速上手并做出业务系统。只要按着里面的case一步步操作下来就可以。 第四章至第六章：面向研发人员的【Oinone的高级特性】、【Oinone的CDM】、【Oinone的通用能力】。这三篇章节重点介绍了Oinone的高级技术特性、提供的通用数据模型和通用基础能力。它们能够帮助我们更快地进行业务开发，从容应对业务的特殊场景要求，比较适合进阶的研发人员。 第七章：面向非研发人员的【Oinone的设计器们】和【Oinone的低无一体】。如果您并不是专业的研发人员，本章可以帮助您通过使用Oinone的无代码可视化设计器轻松自主解决业务需求，并且当可视化设计器满足不了的时候，您还可以在【Oinone的低无一体】中找到方式，并寻求研发帮助

    史, 昂的头像 史, 昂
    Oinone 7天入门到精通 2024年5月23日
    1.2K00
  • 5.4 基础支撑之商业关系域 Oinone 7天入门到精通

    5.4 基础支撑之商业关系域

    PamirsPartner作为商业关系与商业行为的主体，那么PamirsPartner间的关系如何描述，本文将介绍两种常见的设计思路，从思维和实现两方面进行对比，给出oinone为啥选择关系设计模式的原因。 一、两种设计模式对比 设计模式思路介绍 角色设计模式思路介绍 从产品角度枚举所有商业角色，每个商业角色对应一个派生的商业主体，并把主体间的关系类型进行整理。 图5-4-1 角色设计模式 关系设计模式思路介绍 从产品角度枚举所有商业角色，每个商业角色对应一个派生的主体间商业关系 图5-4-2 关系设计模式 设计模式对应实现介绍 角色设计模式实现介绍 不单商业主体需要扩展，关系也要额外维护，可以是字段或是关系表。一般M2O和O2M字段维护，M2M关系表维护。 创建合同场景中甲方选择【商业主体A】，乙方必须是【商业主体A】有关联的经销商、分销商、零售商、供应商等，则在角色设计模式下就非常麻烦，因为关系都是独立维护的 图5-4-3 角色设计模式实现介绍 关系设计模式实现介绍 只需维护商业关系扩展 同时在设计上收敛了商业关系，统一管理应对不同场景都比较从容 图5-4-4 关系设计模式实现介绍 二、oinone商业关系的默认实现 首先oinone的商业关系选择关系设计模式 其次模型上采用多表继承模式，父模型上维护核心字段，子模型维护个性化字段。

    史, 昂的头像 史, 昂
    2024年5月23日
    1.0K00

Leave a Reply

登录后才能评论