wangli/workflow-engine
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(REQ-3004) - 优化审批实例数据同步 ES 的逻辑

Browse Source
This commit is contained in:
wangli 2024-11-21 09:26:03 +08:00
parent 7f74794c5a
commit fea19159db
3 changed files with 69 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
workflow-engine-core/src/main/java/cn/axzo/workflow/core/conf/SupportRefreshProperties.java
View File

@ -74,4 +74,7 @@ public class SupportRefreshProperties {
*/
@Value("${workflow.useNewToAdminApi:false}")
private Boolean useNewToAdminApi;
@Value("${workflow.esSyncBatchSize:10}")
private Integer esSyncBatchSize;
}

18
workflow-engine-server/src/main/java/cn/axzo/workflow/server/mq/inside/SelfBoradcastRocketConfiguration.java
View File

@ -18,18 +18,18 @@ import javax.annotation.Resource;
* @author wangli
* @since 2024-11-06 15:05
*/
@Configuration(proxyBeanMethods = false)
//@Configuration(proxyBeanMethods = false)
public class SelfBoradcastRocketConfiguration {
public static final String DEFAULT_EVENT = "topic_workflow_engine_";
@Component
@ConditionalOnProperty(name = "rocketmq.name-server")
@RocketMQMessageListener(topic = DEFAULT_EVENT + "${spring.profiles.active}",
consumerGroup = "GID_${spring.application.name}_workflow_engine_${spring.profiles.active}_consumer",
consumeMode = ConsumeMode.ORDERLY,
maxReconsumeTimes = 3,
nameServer = "${rocketmq.name-server}"
)
// @Component
// @ConditionalOnProperty(name = "rocketmq.name-server")
// @RocketMQMessageListener(topic = DEFAULT_EVENT + "${spring.profiles.active}",
// consumerGroup = "GID_${spring.application.name}_workflow_engine_${spring.profiles.active}_consumer",
// consumeMode = ConsumeMode.ORDERLY,
// maxReconsumeTimes = 3,
// nameServer = "${rocketmq.name-server}"
// )
public static class WorkflowEngineBroadcastConsumer extends BaseListener implements RocketMQListener<MessageExt> {
@Resource
private EventConsumer eventConsumer;

57
workflow-engine-server/src/main/java/cn/axzo/workflow/server/mq/inside/consumer/ElasticSearchBatchSyncListener.java Normal file
View File

@ -0,0 +1,57 @@
package cn.axzo.workflow.server.mq.inside.consumer;
import cn.axzo.workflow.core.conf.SupportRefreshProperties;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
/**
* 批量同步 ES
*
* @author wangli
* @since 2024-11-20 18:06
*/
@Component
@Slf4j
public class ElasticSearchBatchSyncListener {
@Resource
private SupportRefreshProperties refreshProperties;
@Resource
private RocketMQProperties rocketMQProperties;
@Value("GID_${spring.application.name}_workflow_engine_${spring.profiles.active}_consumer")
private String consumerGroup;
@Value("topic_workflow_engine_${spring.profiles.active}")
private String topic;
@SneakyThrows
@PostConstruct
public void init() {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.subscribe(topic, "*");
consumer.setConsumeMessageBatchMaxSize(refreshProperties.getEsSyncBatchSize());
consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
log.warn("拉取消息条数：{}", msgs.size());
try {
for (MessageExt msg : msgs) {
String msgBody = new String(msg.getBody(), StandardCharsets.UTF_8);
}
} catch (Exception e) {
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
// consumer.start();
}
}