Spring Batch Partition Handler错误

我是Spring框架和Spring Batch的新手

我正在尝试设置示例 spring 批远程分区示例 .

我正在使用这个堆栈Spring Boot Spring Batch Spring Integration AWS SQS

事后,我成功了 .

1.创建所有配置,包括渠道,工作,队列和其他东西 .

2.Ran主进程,我能够对表进行分区,并将分区元数据推送到AWS SQS .

但是在运行slave进程时我得到了错误,在slave进程中,我能够从队列中提取消息但是在StepExecutionRequestHandler的handle()方法中获取错误

org.springframework.messaging.MessageHandlingException:嵌套异常是org.springframework.expression.spel.SpelEvaluationException:EL1004E:方法调用:在org.springframework.batch.integration.partition上找不到方法句柄(java.lang.String) . StepExecutionRequestHandler类型,failedMessage = GenericMessage [payload = StepExecutionRequest:[jobExecutionId = 2,stepExecutionId = 3,stepName = slaveStep],headers = {sequenceNumber = 2,aws_messageId =“”,SentTimestamp = 1523215624042,sequenceSize = 4,SenderId =“”, aws_receiptHandle =“”,ApproximateReceiveCount = 2,correlationId = 2:slaveStep,id =“”,lookupDestination = master,aws_queue = master,ApproximateFirstReceiveTimestamp = 1523215634470,timestamp = 1523215864910}]

@Configuration
public class JobConfiguration implements ApplicationContextAware
{

  @Autowired
  public JobBuilderFactory jobBuilderFactory;

  @Autowired
  public StepBuilderFactory stepBuilderFactory;

  @Autowired
  public DataSource dataSource;

  @Autowired
  public JobExplorer jobExplorer;

  @Autowired
  public JobRepository jobRepository;

  private ApplicationContext applicationContext;

  private static final int GRID_SIZE = 4;

  @Bean
  public PartitionHandler partitionHandler(MessagingTemplate messagingTemplate) throws Exception
  {
    MessageChannelPartitionHandler partitionHandler = new MessageChannelPartitionHandler();
    partitionHandler.setStepName("slaveStep");
    partitionHandler.setGridSize(GRID_SIZE);
    partitionHandler.setMessagingOperations(messagingTemplate);
    partitionHandler.setPollInterval(5000l);
    partitionHandler.setJobExplorer(this.jobExplorer);

    partitionHandler.afterPropertiesSet();

    return partitionHandler;
  }

  @Bean
  public ColumnRangePartitioner partitioner()
  {
    ColumnRangePartitioner columnRangePartitioner = new ColumnRangePartitioner();

    columnRangePartitioner.setColumn("id");
    columnRangePartitioner.setDataSource(this.dataSource);
    columnRangePartitioner.setTable("customer");
    return columnRangePartitioner;
  }

  @Bean
  @Profile("slave")
  @ServiceActivator(inputChannel = "inboundRequests", outputChannel = "outboundStaging")
  public StepExecutionRequestHandler stepExecutionRequestHandler()
  {
    StepExecutionRequestHandler stepExecutionRequestHandler = new StepExecutionRequestHandler();

    BeanFactoryStepLocator stepLocator = new BeanFactoryStepLocator();
    stepLocator.setBeanFactory(this.applicationContext);
    stepExecutionRequestHandler.setStepLocator(stepLocator);
    stepExecutionRequestHandler.setJobExplorer(this.jobExplorer);

    return stepExecutionRequestHandler;
  }

  @Bean(name = PollerMetadata.DEFAULT_POLLER)
  public PollerMetadata defaultPoller()
  {
    PollerMetadata pollerMetadata = new PollerMetadata();
    pollerMetadata.setTrigger(new PeriodicTrigger(10));
    return pollerMetadata;
  }

  @Bean
  @StepScope
  public JdbcPagingItemReader<Customer> pagingItemReader(@Value("#{stepExecutionContext['minValue']}") Long minValue,
      @Value("#{stepExecutionContext['maxValue']}") Long maxValue)
  {
    System.out.println("reading " + minValue + " to " + maxValue);
    JdbcPagingItemReader<Customer> reader = new JdbcPagingItemReader<>();

    reader.setDataSource(this.dataSource);
    reader.setFetchSize(100);
    reader.setRowMapper(new CustomerRowMapper());

    MySqlPagingQueryProvider queryProvider = new MySqlPagingQueryProvider();
    queryProvider.setSelectClause("id, firstName, lastName, birthdate");
    queryProvider.setFromClause("from customer");
    queryProvider.setWhereClause("where id >= " + minValue + " and id <= " + maxValue);

    Map<String, Order> sortKeys = new HashMap<>(1);

    sortKeys.put("id", Order.ASCENDING);

    queryProvider.setSortKeys(sortKeys);

    reader.setQueryProvider(queryProvider);

    return reader;
  }

  @Bean
  @StepScope
  public JdbcBatchItemWriter<Customer> customerItemWriter()
  {
    JdbcBatchItemWriter<Customer> itemWriter = new JdbcBatchItemWriter<>();

    itemWriter.setDataSource(this.dataSource);
    itemWriter.setSql("INSERT INTO new_customer VALUES (:id, :firstName, :lastName, :birthdate)");
    itemWriter.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider());
    itemWriter.afterPropertiesSet();

    return itemWriter;
  }

  @Bean
  public Step step1() throws Exception
  {
    return stepBuilderFactory.get("step1").partitioner(slaveStep().getName(), partitioner()).step(slaveStep())
        .partitionHandler(partitionHandler(null)).build();
  }

  @Bean
  public Step slaveStep()
  {
    return stepBuilderFactory.get("slaveStep").<Customer, Customer>chunk(1000).reader(pagingItemReader(null, null)).writer(customerItemWriter())
        .build();
  }

  @Bean
  @Profile("master")
  public Job job() throws Exception
  {
    return jobBuilderFactory.get("job").start(step1()).build();
  }

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

@Configuration
public class IntegrationConfiguration
{

  @Autowired
  private AmazonSQSAsync amazonSqs;

  @Bean
  public MessagingTemplate messageTemplate()
  {
    MessagingTemplate messagingTemplate = new MessagingTemplate(outboundRequests());

    messagingTemplate.setReceiveTimeout(60000000l);

    return messagingTemplate;
  }

  @Bean
  public DirectChannel outboundRequests()
  {
    return new DirectChannel();
  }


  @Bean
  @Profile("slave")
  public MessageProducer sqsMessageDrivenChannelAdapter()
  {
    SqsMessageDrivenChannelAdapter adapter = new SqsMessageDrivenChannelAdapter(this.amazonSqs, "master");
    adapter.setOutputChannel(inboundRequests());
    adapter.afterPropertiesSet();
    return adapter;
  }

  @Bean
  @ServiceActivator(inputChannel = "outboundRequests")
  public MessageHandler sqsMessageHandler()
  {
    SqsMessageHandler messageHandler = new SqsMessageHandler(amazonSqs);
    messageHandler.setQueue("master");
    return messageHandler;
  }

  @Bean
  public PollableChannel outboundStaging()
  {
    return new NullChannel();
  }

  @Bean
  public QueueChannel inboundRequests()
  {
    return new QueueChannel();
  }
}

谢谢

回答(1)

2 years ago

你应该记住 StepExecutionRequestHandler 的 Contract 如下:

public StepExecution handle(StepExecutionRequest request)

根据您的异常和SQS的性质, inboundRequests 中消息的有效负载是一个字符串 . 我相信它是在JSON中 . 所以,考虑在 StepExecutionRequestHandler 之前使用 JsonToObjectTrabsformer .

UPDATE

有效载荷不是JSON格式 . 它是一个字符串,从StepExecutionRequest类的toString()创建 . 格式为StepExecutionRequest:[jobExecutionId = 2,stepExecutionId = 3,stepName = slaveStep]

好!我明白你的意思了 . SQS Message 只能有 String 正文 . 发送消息到SQS的 SqsMessageHandler 默认使用 GenericMessageConverter 将传入对象转换为String .

我认为您需要考虑使用 MappingJackson2MessageConverter 配置 SqsMessageHandler 以将 StepExecutionRequest 真正序列化为正确的JSON并让它通过SQS传输 .

在调用 StepExecutionRequestHandler 之前的另一个(从属)侧,你真的应该在 JsonToObjectTransformer 之后为 JsonToObjectTransformer 放置一个 @Transformer .