spring-batch - ItemReaderAdapter と ItemStream

Question

<bean id="itemReader" class="org.springframework.batch.item.adapter.ItemReaderAdapter">
  <property name="targetObject" ref="fooService" />
  <property name="targetMethod" value="generateFoo" />
</bean>

この単純なセットアップを考えると、fooService は単純な pojo Spring Bean です。

fooService が ItemStream を実装し、open メソッドと update メソッドを正しく実装している場合、ItemReader は再起動可能になりますか?

よろしく

score 4 · Accepted Answer

わかりました、私はついに自分の質問に答えました。

ほとんどの場合、必要なデータへのアクセスを提供するある種の DAO またはサービスが既にあるため、ItemReaderAdapter は非常に役立ちます。

しかし、私のテストでは、ItemReaderAdapter impl を使用するとそれが示されました。ItemStream を実装していないため、すぐに再起動できませんでした。

したがって、再起動可能な機能を備えた ItemReaderAdapter を使用したい場合は、ここに私の解決策があります。

テスト済みで動作しています;-)

1) ItemReaderAdapter の独自の実装を作成します。

package xxx.readers.adapters;

import java.math.BigDecimal;

import org.apache.log4j.Logger;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.adapter.AbstractMethodInvokingDelegator;
import org.springframework.beans.factory.annotation.Autowired;


/**
* Invokes a custom method on a delegate plain old Java object which itself
* provides an item.
* 
* overriden to implements the ItemStream interface
* 
* @author Benoit Campeau
*/
public class MyItemReaderAdapter<T> extends AbstractMethodInvokingDelegator<T>  implements ItemReader<T>, ItemStream {

private static final Logger log = Logger.getLogger(MyItemReaderAdapter.class);

private long currentCount = 0;

private final String CONTEXT_COUNT_KEY = "ReglementAdapter.count"; 

/**
 * @return return value of the target method.
 */
public T read() throws Exception {

    super.setArguments(new Long[]{currentCount++});
    return invokeDelegateMethod();
}
@Override
public void open(ExecutionContext executionContext)
        throws ItemStreamException {
    currentCount = executionContext.getLong(CONTEXT_COUNT_KEY,0);
    log.info("Open Stream current count : " + currentCount);

}


@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
    executionContext.putLong(CONTEXT_COUNT_KEY, currentCount);
    log.info("Update Stream current count : " + currentCount);
}


@Override
public void close() throws ItemStreamException {
    // TODO Auto-generated method stub

}

}

2) 次に、impl を使用してリーダーをアダプターとして構成します。(MyItemReaderAdapter)。

<bean id="MyReader" class="xxx.readers.adapters.MyItemReaderAdapter">
    <property name="targetObject" ref="someAdapter" />
    <property name="targetMethod" value="next" />       
</bean>

3) 最後に、アダプタデリゲートクラスとして機能するコンポーネントを作成します。

package fcdq.iemt.batch.validation.reglement.readers.adapters;

import java.util.List;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;


@Component("someAdapter")
public class SomeAdapter {

private static final Logger log = Logger.getLogger(SomeAdapter .class);

@Autowired
private SomeService srv1;


private List<Transaction> listTrx;



public void init() {
    log.info("Initializing " + SomeAdapter.class.toString() );
    listTrx = srv1.findByTimestampAndStatus(context.getBeginTSCutoff(), context.getEndTSCutoff(), TransactionTypeEnum.TRANSFER_COMPLETE);
        }

/**
 * read method delegate  
 * @return
 */
public Transaction next(Long index) {
    if (listTrx != null && listTrx.size() > index ) {
        return listTrx.get(index.intValue());
    } else {
        return null;
    }

}

}

注意事項:

MyItemReaderAdapter の setArguments に注目してください。これは、execution_context に保存されている currentCount の値をデリゲートの read() メソッドに渡すために必須です。
MyItemReaderAdapter は InitializingBean インターフェースを実装していないことに注意してください。代わりに stepListener を使用します。これは、アイテムのリストを初期化してジャストインタイムで読み取れるようにするためです。

それが他の誰かを助けることを願っています。

よろしく

score 0 · Accepted Answer

同じことを達成する別の方法は、拡張することですAbstractItemCountingItemStreamItemReader

ドキュメントから::

ExecutionContext にアイテム数を格納することで再起動をサポートする ItemReaders の抽象スーパークラス (したがって、実行間でアイテムの順序を保持する必要があります)。サブクラスは本質的にスレッドセーフではありません

サンプルコード:

package com.***.batch.reader;

import java.util.List;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.mivim.campaignmanager.data.model.custom.CustomCampaignSubscriberEmail;
import com.mivim.campaignmanager.service.CampaignSubscriberService;


@Component
public class CampaignSubscriberItemReader extends
        AbstractItemCountingItemStreamItemReader<CustomCampaignSubscriberEmail> {

    private Logger logger = LogManager.getLogger(CampaignSubscriberItemReader.class);

    @Autowired
    CampaignSubscriberService campaignSubscriberService;

    List<CustomCampaignSubscriberEmail> customCampaignSubscriberEmails;

    final String ecName = "csItemReaderContext";

    public CampaignSubscriberItemReader() {
        setName(ecName);
    }

    @Override
    protected CustomCampaignSubscriberEmail doRead() throws Exception {
        CustomCampaignSubscriberEmail customCampaignSubscriberEmail = customCampaignSubscriberEmails
                .get(getCurrentItemCount() - 1);
        return customCampaignSubscriberEmail;
    }

    @Override
    protected void doOpen() throws Exception {
        customCampaignSubscriberEmails = campaignSubscriberService.getPendingCampaignSubscriber();

        setMaxItemCount(customCampaignSubscriberEmails.size());

    }

    @Override
    protected void doClose() throws Exception {
        customCampaignSubscriberEmails.clear();
        setMaxItemCount(0);
        setCurrentItemCount(0);
    }


}

spring-batch - ItemReaderAdapter と ItemStream

2 に答える 2

Related

Reference