1Queue::DBI(3pm) User Contributed Perl Documentation Queue::DBI(3pm)
2
6 Queue::DBI - A queueing module with an emphasis on safety, using DBI as
7 a storage system for queued data.
8
10 Version 2.7.0
11
13 This module allows you to safely use a queueing system by preventing
14 backtracking, infinite loops and data loss.
15 An emphasis of this distribution is to provide an extremely reliable
17 dequeueing mechanism without having to use transactions.
18 use Queue::DBI;
20 my $queue = Queue::DBI->new(
21 'queue_name' => $queue_name,
22 'database_handle' => $dbh,
23 'cleanup_timeout' => 3600,
24 'verbose' => 1,
25 );
26 # Store a complex data structure.
28 $queue->enqueue(
29 {
30 values => [ 1, 2, 3 ],
31 data => { key1 => 1, key2 => 2 },
32 }
33 );
34 # Store a scalar, which must be passed by reference.
36 $queue->enqueue( \"Lorem ipsum dolor sit amet" );
37 # Process the queued elements one by one.
39 while ( my $queue_element = $queue->next() )
40 {
41 # Skip elements that cannot be locked.
42 next
43 unless $queue_element->lock();
44 eval {
46 # Do some work
47 process( $queue_element->{'email'} );
48 };
49 if ( $@ )
50 {
51 # Something failed, we clear the lock but don't delete the record in the
52 # queue so that we can try again next time
53 $queue_element->requeue();
54 }
55 else
56 {
57 # All good, remove definitively the element
58 $queue_element->success();
59 }
60 }
61 # Requeue items that have been locked for more than 6 hours
63 $queue->cleanup( 6 * 3600 );
64
66 This distribution currently supports:
67 • SQLite
69 • MySQL
71 • PostgreSQL
73 Please contact me if you need support for another database type, I'm
75 always glad to add extensions if you can help me with testing.
76
78 new()
79 Create a new Queue::DBI object.
80 my $queue = Queue::DBI->new(
82 'queue_name' => $queue_name,
83 'database_handle' => $dbh,
84 'cleanup_timeout' => 3600,
85 'verbose' => 1,
86 'max_requeue_count' => 5,
87 );
88 # Custom table names (optional).
90 my $queue = Queue::DBI->new(
91 'queue_name' => $queue_name,
92 'database_handle' => $dbh,
93 'cleanup_timeout' => 3600,
94 'verbose' => 1,
95 'max_requeue_count' => 5,
96 'queues_table_name' => $custom_queues_table_name,
97 'queue_elements_table_name' => $custom_queue_elements_table_name,
98 );
99 Parameters:
101 • 'queue_name'
103 Mandatory, the name of the queue elements will be added to /
105 removed from.
106 • 'database handle'
108 Mandatory, a DBI object.
110 • 'cleanup_timeout'
112 Optional, if set to an integer representing a time in seconds, the
114 module will automatically make available again elements that have
115 been locked longuer than that time.
116 • 'verbose'
118 Optional, control the verbosity of the warnings in the code. 0 will
120 not display any warning; 1 will only give one line warnings about
121 the current operation; 2 will also usually output the SQL queries
122 performed.
123 • 'max_requeue_count'
125 By default, Queue:::DBI will retrieve again the queue elements that
127 were requeued without limit to the number of times they have been
128 requeued. Use this option to specify how many times an element can
129 be requeued before it is ignored when retrieving elements.
130 • 'queues_table_name'
132 By default, Queue::DBI uses a table named 'queues' to store the
134 queue definitions. This allows using your own name, if you want to
135 support separate queuing systems or legacy systems.
136 • 'queue_elements_table_name'
138 By default, Queue::DBI uses a table named 'queue_elements' to store
140 the queued data. This allows using your own name, if you want to
141 support separate queuing systems or legacy systems.
142 • 'lifetime'
144 By default, Queue:::DBI will fetch elements regardless of how old
146 they are. Use this option to specify how old (in seconds) an
147 element can be and still be retrieved for processing.
148 get_queue_id()
150 Returns the queue ID corresponding to the current queue object.
151 my $queue_id = $queue->get_queue_id();
153 count()
155 Returns the number of elements in the queue.
156 my $elements_count = $queue->count();
158 Optional parameter:
160 • exclude_locked_elements
162 Exclude locked elements from the count. Default 0.
164 my $unlocked_elements_count = $queue->count(
166 exclude_locked_elements => 1
167 );
168 enqueue()
170 Adds a new element at the end of the current queue.
171 # Store a scalar by passing its reference.
173 my $queue_element_id = $queue->enqueue( \$string );
174 my $queue_element_id = $queue->enqueue( \"string" );
175 # Store an array reference.
177 my $queue_element_id = $queue->enqueue( [ 1, 2, 3 ] );
178 # Store a hash reference.
180 my $queue_element_id = $queue->enqueue( { key => 123 } );
181 # Store a complex datastructure.
183 my $queue_element_id = $queue->enqueue(
184 {
185 values => [ 1, 2, 3 ],
186 data => { key1 => 1, key2 => 2 },
187 }
188 );
189 The data passed should be a reference to a scalar or a reference to a
191 complex data structure, but you cannot pass a scalar directly. There is
192 otherwise no limitation on the type of data that can be stored as it is
193 serialized for storage in the database.
194 next()
196 Retrieves the next element from the queue and returns it in the form of
197 a Queue::DBI::Element object.
198 my $queue_element = $queue->next();
200 while ( my $queue_element = $queue->next() )
202 {
203 # [...]
204 }
205 Additionally, for testing purposes, a list of IDs to use when trying to
207 retrieve elements can be specified using 'search_in_ids':
208 my $queue_item = $queue->next( 'search_in_ids' => [ 123, 124, 125 ] );
210 retrieve_batch()
212 Retrieves a batch of elements from the queue and returns them in an
213 arrayref.
214 This method requires an integer to be passed as parameter to indicate
216 the maximum size of the batch to be retrieved.
217 my $queue_elements = $queue->retrieve_batch( 500 );
219 foreach ( @$queue_elements )
221 {
222 # [...]
223 }
224 Additionally, for testing purposes, a list of IDs to use when trying to
226 retrieve elements can be specified using 'search_in_ids':
227 my $queue_items = $queue->retrieve_batch(
229 10,
230 'search_in_ids' => [ 123, 124, 125 ],
231 );
232 get_element_by_id()
234 Retrieves a queue element using a queue element ID, ignoring any lock
235 placed on that element.
236 This method is mostly useful when doing a lock on an element and then
238 calling success/requeue asynchroneously.
239 This method requires a queue element ID to be passed as parameter.
241 my $queue_element = $queue->get_element_by_id( 123456 );
243 cleanup()
245 Requeue items that have been locked for more than the time in seconds
246 specified as parameter.
247 Returns the items requeued so that a specific action can be taken on
249 them.
250 my $elements = $queue->cleanup( $time_in_seconds );
252 foreach my $element ( @$elements )
253 {
254 # $element is a Queue::DBI::Element object
255 }
256 purge()
258 Remove (permanently, caveat emptor!) queue elements based on how many
259 times they've been requeued or how old they are, and return the number
260 of elements deleted.
261 # Remove permanently elements that have been requeued more than 10 times.
263 my $deleted_elements_count = $queue->purge( max_requeue_count => 10 );
264 # Remove permanently elements that were created over an hour ago.
266 my $deleted_elements_count = $queue->purge( lifetime => 3600 );
267 Important: locked elements are not purged even if they match the
269 criteria, as they are presumed to be currently in process and purging
270 them would create unexpected failures in the application processing
271 them.
272 Also note that max_requeue_count and lifetime cannot be combined.
274
276 get_max_requeue_count()
277 Return how many times an element can be requeued before it is ignored
278 when retrieving elements.
279 my $max_requeue_count = $queue->get_max_requeue_count();
281 set_max_requeue_count()
283 Set the number of time an element can be requeued before it is ignored
284 when retrieving elements. Set it to "undef" to disable the limit.
285 # Don't keep pulling the element if it has been requeued more than 5 times.
287 $queue->set_max_requeue_count( 5 );+
288 # Retry without limit.
290 $queue->set_max_requeue_count( undef );
291 get_lifetime()
293 Return how old an element can be before it is ignored when retrieving
294 elements.
295 # Find how old an element can be before the queue will stop retrieving it.
297 my $lifetime = $queue->get_lifetime();
298 set_lifetime()
300 Set how old an element can be before it is ignored when retrieving
301 elements.
302 Set it to "undef" to reset Queue::DBI back to its default behavior of
304 retrieving elements without time limit.
305 # Don't pull queue elements that are more than an hour old.
307 $queue->set_lifetime( 3600 );
308 # Pull elements without time limit.
310 $queue->set_lifetime( undef );
311 get_verbose()
313 Return the verbosity level, which is used in the module to determine
314 when and what type of debugging statements / information should be
315 warned out.
316 See set_verbose() for the possible values this function can return.
318 warn 'Verbose' if $queue->get_verbose();
320 warn 'Very verbose' if $queue->get_verbose() > 1;
322 set_verbose()
324 Control the verbosity of the warnings in the code:
325 • 0 will not display any warning;
327 • 1 will only give one line warnings about the current operation;
329 • 2 will also usually output the SQL queries performed.
331 $queue->set_verbose(1); # turn on verbose information
333 $queue->set_verbose(2); # be extra verbose
335 $queue->set_verbose(0); # quiet now!
337
339 freeze()
340 Serialize an element to store it in a SQL "text" column.
341 my $frozen_data = $queue->freeze( $data );
343 thaw()
345 Deserialize an element which was stored a SQL "text" column.
346 my $thawed_data = $queue->thaw( $frozen_data );
348
350 create_tables()
351 Please use create_tables() in Queue::DBI::Admin instead.
352 Here is an example that shows how to refactor your call to this
354 deprecated function:
355 # Load the admin module.
357 use Queue::DBI::Admin;
358 # Create the object which will allow managing the queues.
360 my $queues_admin = Queue::DBI::Admin->new(
361 database_handle => $dbh,
362 );
363 # Create the tables required by Queue::DBI to store the queues and data.
365 $queues_admin->create_tables(
366 drop_if_exist => $boolean,
367 );
368 lifetime()
370 Please use get_lifetime() and set_lifetime() instead.
371 verbose()
373 Please use get_verbose() and set_verbose() instead.
374 max_requeue_count()
376 Please use get_max_requeue_count() and set_max_requeue_count() instead.
377
379 get_dbh()
380 Returns the database handle used for this queue.
381 my $dbh = $queue->get_dbh();
383 get_queues_table_name()
385 Returns the name of the table used to store queue definitions.
386 my $queues_table_name = $queue->get_queues_table_name();
388 get_queue_elements_table_name()
390 Returns the name of the table used to store queue definitions.
391 my $queue_elements_table_name = $queue->get_queue_elements_table_name();
393
395 Please report any bugs or feature requests through the web interface at
396 <https://github.com/guillaumeaubert/Queue-DBI/issues/new>. I will be
397 notified, and then you'll automatically be notified of progress on your
398 bug as I make changes.
399
401 You can find documentation for this module with the perldoc command.
402 perldoc Queue::DBI
404 You can also look for information at:
406 • GitHub's request tracker
408 <https://github.com/guillaumeaubert/Queue-DBI/issues>
410 • AnnoCPAN: Annotated CPAN documentation
412 <http://annocpan.org/dist/Queue-DBI>
414 • CPAN Ratings
416 <http://cpanratings.perl.org/d/Queue-DBI>
418 • MetaCPAN
420 <https://metacpan.org/release/Queue-DBI>
422
424 Guillaume Aubert <https://metacpan.org/author/AUBERTG>, "<aubertg at
425 cpan.org>".
426
428 I originally developed this project for ThinkGeek
429 (<http://www.thinkgeek.com/>). Thanks for allowing me to open-source
430 it!
431
433 Copyright 2009-2017 Guillaume Aubert.
434 This code is free software; you can redistribute it and/or modify it
436 under the same terms as Perl 5 itself.
437 This program is distributed in the hope that it will be useful, but
439 WITHOUT ANY WARRANTY; without even the implied warranty of
440 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the LICENSE
441 file for more details.
442perl v5.38.0 2023-07-21 Queue::DBI(3pm)