Spring Poller работает с нескольких серверов приложений

Question

Мы используем Spring Poller в нашем приложении для опроса данных из базы данных MySQL и отправки третьим лицам. Функциональность как таковая работает, но когда мы перешли на PRODUCTION, поскольку у нас есть несколько серверов приложений, задание запускается со всех серверов, и нам нужно, чтобы оно выполнялось на всех серверах для обработки входящего запроса.

Время опроса настроено для запуска @ каждые 5 секунд.

Но даже если мы добавим оператор обновления, определенная запись будет выбрана на нескольких серверах, так как оба работают одновременно.

У нас есть следующие конфигурации

<int-jdbc:inbound-channel-adapter id="datachannel"
        query="${sql}"
        data-source="dbDataSource" max-rows-per-poll="1" row-mapper="pollerdatamapper"
        update="update <table> set flag=1 where id =:Id">

        <int:poller fixed-rate="${pollerinterval}">
        <int:transactional/>
        </int:poller>

    </int-jdbc:inbound-channel-adapter>

Класс PollerService будет вызываться, как показано ниже, для каждой записи из указанного выше опроса

<int:service-activator input-channel="datachannel"
        output-channel="executerchannel" ref="pollerservice" method="getRecordFromPoller">
</int:service-activator>

открытый класс PollerService {

private static final Logger LOGGER = Logger.getLogger(PollerService.class);


public PollerDataBO getRecordFromPoller(PollerDataBO pollerDataBO)
{
  LOGGER.info("call for the Id " + Id); 
 }

Не могли бы вы подтвердить, есть ли какие-либо транзакционные настройки, которые мы можем ограничить выбор одной и той же записи на других серверах.

Answer 1

Правильно, SELECT ... FOR UPDATE Должно быть отличным решением для вас:

https://docs.oracle.com/cd/E17952_01/mysql-5.1-en/innodb-locking-reads.html

Вы также можете попытаться играть с isolation="SERIALIZABLE" для <int:transactional/>, но я не слишком уверен в этом.

Также я думаю, что мы могли бы улучшить JdbcPollingChannelAdapter в коде вроде:

if (this.updatePerRow) {
    for (Object row : payload) {
            executeUpdateQuery(row);
    }
}

И пропустить те строки, которые не были обновлены.

Не стесняйтесь поднять JIRA по этому вопросу.