first commit

zrj

2024-09-24 1216a77e26cb2c788fa6a78991e073e671f0f59d

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
package tech.powerjob.worker.processor;
 
import lombok.extern.slf4j.Slf4j;
import tech.powerjob.common.exception.PowerJobException;
import tech.powerjob.worker.extension.processor.ProcessorBean;
import tech.powerjob.worker.extension.processor.ProcessorDefinition;
import tech.powerjob.worker.extension.processor.ProcessorFactory;
 
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
 
/**
 * PowerJobProcessorLoader
 *
 * @author tjq
 * @since 2023/1/17
 */
@Slf4j
public class PowerJobProcessorLoader implements ProcessorLoader {
 
    private final List<ProcessorFactory> processorFactoryList;
    private final Map<ProcessorDefinition, ProcessorBean> def2Bean = new ConcurrentHashMap<>(128);
 
    public PowerJobProcessorLoader(List<ProcessorFactory> processorFactoryList) {
        this.processorFactoryList = processorFactoryList;
    }
 
    @Override
    public ProcessorBean load(ProcessorDefinition definition) {
        return def2Bean.computeIfAbsent(definition, ignore -> {
            final String processorType = definition.getProcessorType();
            log.info("[ProcessorFactory] start to load Processor: {}", definition);
            for (ProcessorFactory pf : processorFactoryList) {
                final String pfName = pf.getClass().getSimpleName();
                if (!Optional.ofNullable(pf.supportTypes()).orElse(Collections.emptySet()).contains(processorType)) {
                    log.info("[ProcessorFactory] [{}] can't load type={}, skip!", pfName, processorType);
                    continue;
                }
                log.info("[ProcessorFactory] [{}] try to load processor: {}", pfName, definition);
                try {
                    ProcessorBean processorBean = pf.build(definition);
                    if (processorBean != null) {
                        log.info("[ProcessorFactory] [{}] load processor successfully: {}", pfName, definition);
                        return processorBean;
                    }
                } catch (Throwable t) {
                    log.error("[ProcessorFactory] [{}] load processor failed: {}", pfName, definition, t);
                }
            }
            throw new PowerJobException("fetch Processor failed, please check your processorType and processorInfo config");
        });
    }
}