Hazelcast Jet Job无法访问spring上下文

Question

我正在创建一个spring引导应用程序，在我的应用程序中，我尝试在普通的spring应用程序中使用Hazelcast Jet，它工作正常，但是在尝试spring引导应用程序时，它无法访问bean。Bean是动态创建的，因此我可以通过AppBeans类访问它们。

对于依赖性，我有以下3条：

<dependency>
        <groupId>com.hazelcast</groupId>
        <artifactId>hazelcast-all</artifactId>
        <version>4.2</version>
    </dependency>
    <dependency>
        <groupId>com.hazelcast.jet</groupId>
        <artifactId>hazelcast-jet</artifactId>
        <version>4.5</version>
    </dependency>
    <dependency>
        <groupId>com.hazelcast</groupId>
        <artifactId>hazelcast-spring</artifactId>
        <version>4.2</version>
    </dependency>

对于创建Hazelcast实例，我在下面的代码中使用：

@SpringAware
public class DataFactory implements Serializable {
   SpringManagedContext springManagedContext = (SpringManagedContext) AppBeans.getBean("springManagedContext");

public JetInstance buildJetInstance() {
    if (!EtlObjects.jetStart) {
        
    EtlObjects.jetStart = true;
    JetConfig jetConfig = new JetConfig();
    jetConfig.getHazelcastConfig().setProperty( "hazelcast.logging.type", "log4j" );
    jetConfig.getInstanceConfig().setCooperativeThreadCount(5);
    jetConfig.configureHazelcast(c -> {
        c.getNetworkConfig().setReuseAddress(true);
        c.setClusterName(UUID.randomUUID().toString());
        c.setManagedContext(springManagedContext);
        c.getNetworkConfig().setPort(9493);
        c.getNetworkConfig().setPublicAddress("localhost");
        c.getNetworkConfig().setPortAutoIncrement(true);
        c.getNetworkConfig().getJoin().getAutoDetectionConfig().setEnabled(false);
        c.getNetworkConfig().getJoin().getMulticastConfig().setEnabled(false);
        c.getNetworkConfig().getJoin().getTcpIpConfig().setEnabled(true).setMembers(Arrays.asList(new String[] {"localhost"}));
        
    });
    EtlObjects.jetInstance = Jet.newJetInstance(jetConfig);

    }
    return EtlObjects.jetInstance;
 }
}

在工作中，当我试图访问bean时，它给了我null：

@Component
public class JDBCDataSource implements ISourceBatch, Serializable {
        @Override
         public BatchStage<Object> readSource(Pipeline pipeline) {
                 BatchSource<Object> jdbcSource = Sources
                .jdbc(() -> {

                 **> //Here it giving null cannot access datasource Bean**

                    Connection conn = ((DataSource)AppBeans.getBean(thissource.get("datasourceName").toString())).getConnection();
                    return conn;
                },
                    (con, parallelism, index) -> {
                     ..........my code
                   }});
         }
}

AppBeans类如下所示：

@Component("s")
public class AppBeans implements ApplicationContextAware, 
ServletContextListener,BeanDefinitionRegistryPostProcessor,Serializable{
private static ApplicationContext CONTEXT;
private static ServletContext SERVLETCONTEXT;
private static BeanDefinitionRegistry REGISTRY;
private transient AutowireCapableBeanFactory beanFactory;


public void setApplicationContext(ApplicationContext context) throws BeansException {
   
    CONTEXT = context;
    beanFactory = context.getAutowireCapableBeanFactory();
 }

 public static Object getBean(String beanName) {
    return CONTEXT.getBean(beanName);
  }

}

也适用于启动应用程序：

@SpringBootApplication(exclude = {MongoAutoConfiguration.class, MongoDataAutoConfiguration.class, FreeMarkerAutoConfiguration.class})
@ComponentScan(basePackages =  {"com"})
@EnableAutoConfiguration
public class AppApplication extends SpringBootServletInitializer {

    public static void main(String[] args) {
    SpringApplication.run(AppApplication.class, args);
   }
   @Bean
   public SpringManagedContext springManagedContext() {
     return new SpringManagedContext();
   }
}

请让我知道我如何能够喷气源访问春季上下文。如果有什么不清楚的话请告诉我

Answer 1

您需要将Spring托管上下文传递给作业源，然后作业源可以使用Spring的应用程序上下文进行初始化。

试一试

    static Pipeline buildPipeline() {
        return Pipeline.create()
                .readFrom(mySource())
                .writeTo(Sinks.logger())
                .getPipeline();
    }
    
    static BatchSource<String> mySource() {
        return SourceBuilder.batch("72743077", 
                        jobContext -> new MyBatchSource(jobContext.hazelcastInstance().getConfig().getManagedContext()))
                    .fillBufferFn(MyBatchSource::fillBufferFn)
                    .build();
    }

    @SpringAware
    static class MyBatchSource implements ApplicationContextAware {
        private ApplicationContext applicationContext;
        
        MyBatchSource(ManagedContext managedContext) {
            if (managedContext instanceof SpringManagedContext) {
                SpringManagedContext springManagedContext = (SpringManagedContext) managedContext;
                springManagedContext.initialize(this);
            }
        }
        
        void fillBufferFn(SourceBuilder.SourceBuffer<String> buffer) {
            Config config = this.applicationContext.getBean(Config.class);
            buffer.add("HELLO FROM " + config.getClusterName());
            buffer.close();
        }

        @Override
        public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
            this.applicationContext = applicationContext;
        }
    }

要使其工作，确保Spring托管上下文在创建服务器实例之前插入到Hazelcast配置中。

    @Bean
    public SpringManagedContext managedContext() {
        return new SpringManagedContext();
    }
    
    @Bean
    public Config config(ManagedContext managedContext) {
        Config config = new Config();
        config.setManagedContext(managedContext);

此外，如果升级到5.1.2或其他最新版本，则只需要将com.hazelcast:hazelcast-spring:5.1.2作为依赖项。