Kolejka w PHP, Opcje

[Prph]

Dawno mnie tu nie było (IMG:style_emoticons/default/smile.gif)

Forumowicze, problem z kolejką w PHP. System, który tworzę przetwarza dane co minutę (z CRONa). Niestety uruchomienie pojedynczego procesu np, przetwarzaj.php nie kończy się w 1 minucie, ponieważ przetwarzanie danych może trwać np. 2 mninuty dla pojedynczego rekordu bazy danych, a rekordów jest np. 20. Teoretycznie szeregowe uruchomienie przetwarzania zajmie więc 40 min. Każdy przetwarzaj.pho bierze z bazy zatem 1 rekord i tylko taki mieli.

Moje pytanie: w jaki sposób napisać kolejkę, która poprawnie przetworzy takie dane? Wywołać z crona 20x ten sam przetwarzaj.php? Zanim proces zacznie wykonywać swoją pracę, ustawia na rekordzie bazy, że jest on przetwarzany, więc inny proces weźmie kolejne dane.

Spotkaliście się z podobnym problemem, macie pomysły?

[Crozin]

Aż mnie coś wzięło, żeby zobaczyć jak można zrobić to "normalnie" (czyt: bez forkowania procesów, a z użyciem normalnych wątków):

[JAVA] pobierz, plaintext 
package com.crozin.concurrent;
 
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
 
public class RecordsProcessor implements Runnable {
	private static final int NUMBER_OF_PROCESSORS = 8;
 
	private final BlockingQueue<Integer> processingQueue = new LinkedBlockingQueue<Integer>();
 
	private class ProcessorThread extends Thread {
		public ProcessorThread(String name) {
			super(name);
		}
 
		@Override
		public void run() {
			try {
				while (true) {
					Integer record = processingQueue.take();
 
					log("Processing record #" + record);
 
					Thread.sleep(new java.util.Random().nextInt(1000) + 1000);
				}
			} catch (InterruptedException e) {
				return;
			}
		}
	}
 
	@Override
	public void run() {
		for (int i = 0; i < NUMBER_OF_PROCESSORS; i++) {
			new ProcessorThread("Processor #" + i).start();
		}
 
		int c = 1;
 
		while (true) {
			if (processingQueue.size() <= 10) {
				log("Refilling the queue");
 
				for (int i = 0; i < 100; i++) {
					processingQueue.add(c++);
				}
			}
 
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
				return;
			}
		}
	}
 
	public static void main(String[] args) throws Exception {
		new RecordsProcessor().run();
	}
 
	public static void log(String msg) {
		System.out.println(Thread.currentThread().getName() + ": " + msg);
	}
}
[JAVA] pobierz, plaintext 

Minusy:
- konieczność zrezygnowania z PHP dla tego elementu systemu.

Plusy:
- będzie działać na każdej platformie,
- gwarantuje, że rekordy zostaną przetworzone w odpowiedniej kolejności oraz tylko i wyłącznie jeden raz,
- brak niepotrzebnych przerw w działaniu,
- kontrola nad ilością przetwarzanych rekordów w danym momencie,
- łatwiej to wszystko ogarnąć, wprowadzać zmiany itp.