Flowable异步历史消息处理-----RabbitMQ

一.介绍

flowable相比activity,增加了一个新的特性，即异步处理历史数据。

我们在操作流程的过程中都会保存历史数据到不同的历史表中，如果流程数量很多的情况下，同步保存历史数据便会带来一定的性能压力。

flowable给我们提供了一种异步处理的方法，其中这种方式的原理其实就是开启历史定时器，然后定期轮训历史作业，然后解析并将其插入到不同的历史表。

这种方式本质还是使用的线程池以及数据库的轮训处理方式，flowable在6.1.2版本之后引入了消息队列机制来处理历史作业。

JMS被用作传输协议，当然了任何协议都是有效的。

这里我采用RabbitMQ，通过异步消息处理历史数据。

1.不使用轮训数据库的方式，而是采用将消息发到到消息队列，并指示消息作业已经准备ok。

2.历史作业数据与运行数据是在同一个事务中的。这样就可以避免脏数据。

3.Flowable提供了一个消息监听器，这样监听器就可以监听到指定的队列消息并进行处理。比如作业的获取，将其存储到历史表中。

4.如果我们实际项目开发中，不希望将历史数据放置到flowable引擎表中，那我们可以扩展相应的代码，将历史作业以及历史数据存放到其他介质，比如mongodb或者es等搜索引擎。

如果使用这种方式，那么消息监听器可以完全与流程引擎解耦，也就是各做各的事情。

架构示意图如下所示：

一.flowable服务端（消息的提供端）

1.引入依赖

<!--flowable工作流依赖-->
<dependency>
 <groupId>org.flowable</groupId>
 <artifactId>flowable-spring-boot-starter-process</artifactId>
 <version>${flowable.version}</version>
</dependency>

<!--jms-->
<dependency>
 <groupId>org.flowable</groupId>
 <artifactId>flowable-jms-spring-executor</artifactId>
 <version>${flowable.version}</version>
</dependency>

<!--mysql依赖-->
<dependency>
 <groupId>mysql</groupId>
 <artifactId>mysql-connector-java</artifactId>
 <version>${mysql-version}</version>
</dependency>

<!-- rabbitmq -->
<dependency>
 <groupId>org.springframework.amqp</groupId>
 <artifactId>spring-rabbit</artifactId>
 <version>${amqp.version}</version>
</dependency>

<!-- flowable-dmn-->
<dependency>
 <groupId>org.flowable</groupId>
 <artifactId>flowable-dmn-spring-configurator</artifactId>
 <version>${flowable.version}</version>
</dependency>

<!-- flowable-form-->
<dependency>
 <groupId>org.flowable</groupId>
 <artifactId>flowable-form-spring-configurator</artifactId>
 <version>${flowable.version}</version>
</dependency>

2.配置Config

@Configuration
public class FlowableRabbitConfig {


    public static final String EXCHANGE = "mbo-flowable";
    public static final String QUEUE = "mbo-flowable-history";


    /**
     * 配置mysql连接
     * @return
     */
    @Bean
    @ConfigurationProperties("spring.datasource")
    public DataSource dataSource() {
        return new DataSource();
    }

    /**
     * 事务配置
     * @return
     */
    @Bean
    public PlatformTransactionManager transactionManager() {
        DataSourceTransactionManager transactionManager = new DataSourceTransactionManager();
        transactionManager.setDataSource(dataSource());
        return transactionManager;
    }

    /**
     * 自定义引擎配置
     * @return
     */
    @Bean
    public SpringProcessEngineConfiguration processEngineConfiguration() {
        SpringProcessEngineConfiguration config = new SpringProcessEngineConfiguration();
        //数据源配置
        config.setDataSource(dataSource());
        //事务配置
        config.setTransactionManager(transactionManager());
        //数据库变化修改
        config.setDatabaseSchemaUpdate(ProcessEngineConfiguration.DB_SCHEMA_UPDATE_TRUE);

        config.addConfigurator(new SpringDmnEngineConfigurator());
        config.addConfigurator(new SpringFormEngineConfigurator());

        // 同步历史
        config.setAsyncHistoryEnabled(true);
        config.setAsyncHistoryExecutorActivate(true);
        config.setAsyncHistoryExecutorMessageQueueMode(true);

        config.setJobManager(jobManager());
        //设置历史级别
        config.setHistory(HistoryLevel.FULL.getKey());
        return config;
    }

    /**
     * 构建自定义流程引擎
     * @return
     */
    @Bean
    public ProcessEngine processEngines() {
        return processEngineConfiguration().buildProcessEngine();
    }

    /**
     * 将历史作业与消息队列关联，发送消息
     * @return
     */
    @Bean
    public RabbitMQMessageBasedJobManager jobManager() {
        RabbitMQMessageBasedJobManager jobManager = new RabbitMQMessageBasedJobManager();
        jobManager.setRabbitTemplate(rabbitTemplate());
        return jobManager;
    }

    /**
     * 配置rabbitmq连接
     * @return
     */
    @Bean
    @ConfigurationProperties("spring.rabbitmq")
    public ConnectionFactory connectionFactory() {
        return new CachingConnectionFactory();
    }

    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory());
        return rabbitTemplate;
    }


    @Bean
    public AmqpAdmin amqpAdmin() {
        return new RabbitAdmin(connectionFactory());
    }

    @Bean
    public TaskExecutor taskExecutor() {
        return new SimpleAsyncTaskExecutor();
    }

    /**
     * 应用启动时创建交换机和队列
     * @param executor
     * @param amqpAdmin
     * @return
     */
    @Bean
    public CommandLineRunner schedulingRunner(final TaskExecutor executor, final AmqpAdmin amqpAdmin) {
        return args -> executor.execute(() -> {

            TopicExchange exchange = new TopicExchange(EXCHANGE);
            amqpAdmin.declareExchange(exchange);
            //队列持久化
            Queue queue = new Queue(QUEUE, true);
            amqpAdmin.declareQueue(queue);
            amqpAdmin.declareBinding(BindingBuilder.bind(queue).to(exchange).with(QUEUE));
        });
    }

}

3.自定义类继承AbstractMessageBasedJobManager

public class RabbitMQMessageBasedJobManager extends AbstractMessageBasedJobManager {


    private RabbitTemplate rabbitTemplate;

    @Override
    protected void sendMessage(JobInfo job) {
        if(job!=null) {
            rabbitTemplate.convertAndSend(FlowableRabbitConfig.EXCHANGE, FlowableRabbitConfig.QUEUE, job.getId());
        }
    }

    public RabbitTemplate getRabbitTemplate() {
        return rabbitTemplate;
    }

    public void setRabbitTemplate(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

}

上面自定义类继承AbstractMessageBasedJobManager，发送工作标识符到指定队列。

二.配置消息监听者

1.依赖同上

2.配置config

@Configuration
public class FlowRabbitConfig {

    public static final String EXCHANGE = "mbo-flowable";
    public static final String QUEUE = "mbo-flowable-history";

    @Bean
    public Queue queue() {
        return new Queue(QUEUE, true);
    }

    @Bean
    public Binding binding() {
        return BindingBuilder.bind(queue()).to(exchange()).with(QUEUE);
    }

    @Bean
    public TopicExchange exchange() {
        return new TopicExchange(EXCHANGE);
    }

    @Bean
    public SimpleMessageListenerContainer container(ConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(QUEUE);
        container.setMessageListener(listenerAdapter);
		//设置消费者数量，flowable异步消息处理，多个消息是一组，一组消息乱序会出错，所以不支持多个消费者消费
        container.setConcurrentConsumers(1);
        container.start();
        return container;
    }

    @Bean
    public Receiver receiver() {
        Receiver receiver = new Receiver();
        receiver.setAsyncHistoryJobMessageReceiver(asyncHistoryJobMessageReceiver());
        return receiver;
    }

    @Bean
    public AsyncHistoryJobMessageReceiver asyncHistoryJobMessageReceiver() {
        AsyncHistoryJobMessageReceiver asyncHistoryJobMessageReceiver = new AsyncHistoryJobMessageReceiver();
        asyncHistoryJobMessageReceiver.setProcessEngineConfiguration(processEngineConfiguration());
        asyncHistoryJobMessageReceiver.setAsyncHistoryJobMessageHandler(myJobMessageHandler());
        return asyncHistoryJobMessageReceiver;
    }

    @Bean
    public MessageListenerAdapter listenerAdapter(Receiver receiver) {
        return new MessageListenerAdapter(receiver, "receiveMessage");
    }

    @Bean
    public MyJobMessageHandler myJobMessageHandler() {
        return new MyJobMessageHandler();
    }

    @Bean
    @ConfigurationProperties("spring.datasource")
    public DataSource dataSource() {
        return new DataSource();
    }

    @Bean
    public PlatformTransactionManager transactionManager() {
        DataSourceTransactionManager transactionManager = new DataSourceTransactionManager();
        transactionManager.setDataSource(dataSource());
        return transactionManager;
    }

    @Bean
    public SpringProcessEngineConfiguration processEngineConfiguration() {
        SpringProcessEngineConfiguration config = new SpringProcessEngineConfiguration();
        config.setDataSource(dataSource());
        config.setTransactionManager(transactionManager());
        config.setDatabaseSchemaUpdate(ProcessEngineConfiguration.DB_SCHEMA_UPDATE_TRUE);

        config.addConfigurator(new SpringDmnEngineConfigurator());
        config.addConfigurator(new SpringFormEngineConfigurator());

        return config;
    }

    @Bean
    public ProcessEngine processEngine() {
        return processEngineConfiguration().buildProcessEngine();
    }
}

3.自定义消息接收端

public class Receiver {

    private AsyncHistoryJobMessageReceiver asyncHistoryJobMessageReceiver;

    public void receiveMessage(byte[] messageBytes) {
        receiveMessage(new String(messageBytes));
    }

    public void receiveMessage(String message) {
        asyncHistoryJobMessageReceiver.messageForJobReceived(message);
    }

    public AsyncHistoryJobMessageReceiver getAsyncHistoryJobMessageReceiver() {
        return asyncHistoryJobMessageReceiver;
    }

    public void setAsyncHistoryJobMessageReceiver(AsyncHistoryJobMessageReceiver asyncHistoryJobMessageReceiver) {
        this.asyncHistoryJobMessageReceiver = asyncHistoryJobMessageReceiver;
    }

}

4.自定义处理器

我这里处理的策略是落库，不过可以根据需要进行历史数据的处理。

落库因为源码里面的方法类型是protectedf，所以我自定义重写了一份。

public class MyJobMessageHandler implements AsyncHistoryJobMessageHandler {


    private static MyHistoryJobHandler myHistoryJobHandler = new MyHistoryJobHandler();

    static {
        myHistoryJobHandler.initDefaultTransformers();
    }
    
    @Override
    public boolean handleJob(HistoryJobEntity historyJobEntity, JsonNode historyData) {
        System.out.println("Handling job " + historyJobEntity.getId() + ", data = " + historyData);
        CommandContext commandContext = Context.getCommandContext();
        myHistoryJobHandler.processHistoryJson(commandContext,historyJobEntity,historyData);
        return true;
    }

    protected byte[] getJobBytes(HistoryJobEntity job) {
        return job.getAdvancedJobHandlerConfigurationByteArrayRef().getBytes();
    }
}

可以通过消息中获取到的历史任务工作标识，获取历史数据，并处理成json格式

5.落库（也可以进行自定义别的处理）

将json格式的历史数据保存到flowable的历史表中，从flowable源码中找到解析json落库的方法，但是是protected类型的方法，继承也无法调用，因为不同包。

所以我Copy出来，修改成public类型，通过自定义类调用

public class MyHistoryJobHandler{

    private static final Logger LOGGER = LoggerFactory.getLogger(AsyncHistoryJobHandler.class);

    public static final String JOB_TYPE = "async-history";

    public List<Integer> list = new ArrayList<>();

    protected Map<String, List<HistoryJsonTransformer>> historyJsonTransformers = new HashMap<>();

    public String getType() {
        return JOB_TYPE;
    }

    public void initDefaultTransformers() {
        addHistoryJsonTransformer(new ProcessInstanceStartHistoryJsonTransformer());
        addHistoryJsonTransformer(new ProcessInstanceEndHistoryJsonTransformer());
        addHistoryJsonTransformer(new ProcessInstanceDeleteHistoryJsonTransformer());
        addHistoryJsonTransformer(new ProcessInstanceDeleteHistoryByProcessDefinitionIdJsonTransformer());
        addHistoryJsonTransformer(new ProcessInstancePropertyChangedHistoryJsonTransformer());
        addHistoryJsonTransformer(new SubProcessInstanceStartHistoryJsonTransformer());
        addHistoryJsonTransformer(new SetProcessDefinitionHistoryJsonTransformer());
        addHistoryJsonTransformer(new UpdateProcessDefinitionCascadeHistoryJsonTransformer());

        addHistoryJsonTransformer(new ActivityStartHistoryJsonTransformer());
        addHistoryJsonTransformer(new ActivityEndHistoryJsonTransformer());
        addHistoryJsonTransformer(new ActivityFullHistoryJsonTransformer());

        addHistoryJsonTransformer(new TaskCreatedHistoryJsonTransformer());
        addHistoryJsonTransformer(new TaskEndedHistoryJsonTransformer());

        addHistoryJsonTransformer(new TaskPropertyChangedHistoryJsonTransformer());
        addHistoryJsonTransformer(new TaskAssigneeChangedHistoryJsonTransformer());
        addHistoryJsonTransformer(new TaskOwnerChangedHistoryJsonTransformer());

        addHistoryJsonTransformer(new IdentityLinkCreatedHistoryJsonTransformer());
        addHistoryJsonTransformer(new IdentityLinkDeletedHistoryJsonTransformer());

        addHistoryJsonTransformer(new VariableCreatedHistoryJsonTransformer());
        addHistoryJsonTransformer(new VariableUpdatedHistoryJsonTransformer());
        addHistoryJsonTransformer(new VariableRemovedHistoryJsonTransformer());
        addHistoryJsonTransformer(new HistoricDetailVariableUpdateHistoryJsonTransformer());
        addHistoryJsonTransformer(new FormPropertiesSubmittedHistoryJsonTransformer());
    }

    public void addHistoryJsonTransformer(HistoryJsonTransformer historyJsonTransformer) {
        String type = historyJsonTransformer.getType();
        if (!historyJsonTransformers.containsKey(type)) {
            historyJsonTransformers.put(type, new ArrayList<HistoryJsonTransformer>());
        }
        historyJsonTransformers.get(historyJsonTransformer.getType()).add(historyJsonTransformer);
    }

    public void processHistoryJson(CommandContext commandContext, HistoryJobEntity job, JsonNode historyNode) {

        String type = historyNode.get(HistoryJsonConstants.TYPE).asText();
        ObjectNode historicalJsonData = (ObjectNode) historyNode.get(HistoryJsonConstants.DATA);

        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("Handling async history job (id={}, type={})", job.getId(), type);
        }
        List<HistoryJsonTransformer> transformers = historyJsonTransformers.get(type);
        if (transformers != null && !transformers.isEmpty()) {
            for (HistoryJsonTransformer transformer : transformers) {
                if (transformer.isApplicable(historicalJsonData, commandContext)) {
                    transformer.transformJson(job, historicalJsonData, commandContext);
                }else {
                    System.err.println("---------------"+job.getId());
                    throw new AsyncHistoryJobNotApplicableException();
                }
            }
        } else {
            LOGGER.debug("Cannot transform history json: no transformers found for type {}", type);
        }
    }

    public Map<String, List<HistoryJsonTransformer>> getHistoryJsonTransformers() {
        return historyJsonTransformers;
    }

    public void setHistoryJsonTransformers(Map<String, List<HistoryJsonTransformer>> historyJsonTransformers) {
        this.historyJsonTransformers = historyJsonTransformers;
    }

}

注意：异步历史消息，消费者不要设置多个，异步历史消息是多个消息为一组，需要一组顺序消费，act_hi_taskinst表的数据需要依赖其他历史表数据先落库。

多个消费者会导致消费消息的乱序，act_hi_taskinst表数据无法落库。

原文链接：https://blog.csdn.net/qq_36418435/article/details/94555187