RabbitMQ DeDup как сообщения

Question

На беговой RabbitMQ пользователя сервера PHP для рабочих,

У меня есть событие, которое будет срабатывать от сообщения «Приведите пользователь X Счет до $ 5» в очередь «заряд». Тем не менее, многие события могут отправить это сообщение, и если в системе работает более одного рабочего, я вижу проблему, когда есть условие гонки, которое происходит, когда пользователь получает двойную плату.

$accountBalance = -25.00 // starts 
    event 1: "Bring User X Account to $5" 
    event 1: check balance, $balance = -25.00 
    event 2: "Bring User X Account to $5" 
    event 1: $chargeAmount = 30 
    event 2: check balance, $balance = -25.00 
    event 1: Charge Card $30 // takes 3s 
    event 2: $chargeAmount = 30 
    event 2: Charge Card $30 // takes 6s 
    event 3: "Bring User X Account to $5" 
    event 3: check balance, $balance = -25.00 
    event 1: add transaction $30 // $accountBalance = 5.00 
    event 1: END 
    event 4: "Bring User X Account to $5" 
    event 4: check balance, $balance = 5.00 
    event 4: END 
    event 2: add transaction $30 // $accountBalance = 35.00 
    event 2: END 
    event 3: $chargeAmount = 30 
    event 3: Charge Card $30 // takes 3s 
    event 3: add transaction $30 // $accountBalance = 65.00 
    event 2: END 
    $accountBalance = 65.00 

Как вы можете видеть, произошло 4 события, и вместо того, чтобы заряжать 30 долларов США один раз, он взимает три раза. Только событие 4 вышло, но потому, что событие 1 закончилось в ближайшее время.

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

Есть ли какие-либо проблемы с продуктами, из которых легко работать?

Answer 1

Вам необходимо использовать распределенный замок, чтобы одновременно предотвратить одновременное выполнение двух задач, а также сделать свою задачу идемпотентной. Даже если у вас есть блокировки, задачи должны быть в состоянии запускать и обнаруживать, что работа была выполнена и проигнорирована. Эти две концепции работают вместе.

Вы можете использовать что-то вроде memcached для распределенной блокировки. Например, вы можете попробовать что-то вроде LockBundle.

$memcached = new Memcached(); 
$this->adapter = new MemcachedAdapter($memcached); 
$lock = new Lock("test", $this->adapter); 
$lock->acquire(); 
// Do stuff here // 
$lock->release(); 

Просто сделайте ключ блокировки любой гранулярностью, которую вы хотите защитить. Есть некоторые проблемы вокруг HA, хотя, если вы пытаетесь сделать что-то устойчивое к авариям memcached.

Вы также можете создать нечто похожее, которое использует таблицу базы данных, в которой у вас есть PK в поле, и попросите процесс вставить значение в таблицу. ПК блокирует любую вторую попытку, и как только первая совершит, последующие будут разблокированы и не будут выполнены из-за нарушения ПК.

Возможно, что-то похожее на Zookeeper, хотя я этого никогда не делал. Я бы подумал, что это будет более устойчивым, чем memcached.