Using Amazon ElastiCache for Valkey as a Queue
Amazon ElastiCache for Valkey can serve as an efficient message queue, leveraging its list and set data structures to manage tasks for immediate or scheduled processing.
Gopinath Srinivasan
Amazon Employee
Published Nov 5, 2024
In this tutorial, I'll build a simple event processing system using Valkey as a message queue that we set up in my previous post : "Using CloudFormation to set up Amazon Elasticache Serverless for Valkey". We'll create two Python programs: a producer that generates mock events and a consumer that processes these events in real-time. I use Amazon Elasticache Serverless for Valkey. However, the steps outlined below are applicable for Amazon Elasticache for Valkey and Redis OSS.
- Amazon ElastiCache for Valkey (set up using the instructions from my previous blog)
- Python 3.x
- Valkey Python library
Install the required python package:
1
pip install valkeyMy example consists of three main components:
- Amazon ElastiCache for Valkey acting as a queue
- A producer program generating mock events
- A consumer program processing the events
The producer and consumer communicate through Valkey, which provides FIFO (First-In-First-Out) queue functionality.
The producer generates mock events simulating a real-world application. Each event includes:
- Event ID
- Event type (user_signup, purchase, etc.)
- User ID
- Timestamp
- Additional metadata
Key features of the producer:
- Generates random mock data, with random delays to simulate real-world scenarios
- Includes random delays to simulate real-world scenarios
- Handles errors and graceful shutdown
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
while True:
# Generate and publish event
event = generate_mock_event()
event_json = json.dumps(event)
# Push to Valkey list
valkey_client.lpush('event_queue', event_json)
counter += 1
print(f"Published event {counter}: {event['event_type']} for user {event['user_id']}")
# Random delay between 1-5 seconds
delay = random.uniform(1, 5)
time.sleep(delay)
except KeyboardInterrupt:
print("\nStopping event producer...")
except valkey.ValkeyError as e:
print(f"Valkey error: {e}")
except Exception as e:
print(f"Unexpected error: {e}")The consumer runs continuously, waiting for new events to process. Key features:
- Blocking read from Valkey queue
- JSON parsing and validation
- Error handling and failed event management
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
while True:
# Block until an event is available (timeout after 1 second)
event = valkey_client.brpop('event_queue', timeout=1)
if event is not None:
# event is a tuple of (queue_name, value)
_, event_data = event
success = process_event(event_data)
if success:
# Implement successful processing and move to a processed queue
pass
else:
# Handle failed processing
# Maybe move to a dead letter queue
valkey_client.lpush('event_queue_failed', event_data)
except KeyboardInterrupt:
print("\nStopping event consumer...")
except valkey.ValkeyError as e:
print(f"Valkey error: {e}")
except Exception as e:
print(f"Unexpected error: {e}")This simple implementation demonstrates the basics of using Amazon ElastiCache for Valkey as a message queue. While this example uses mock data, the same pattern can be applied to real-world scenarios such as:
- Processing user activities
- Handling IoT device data
- Managing background tasks
Happy building!
Any opinions in this post are those of the individual author and may not reflect the opinions of AWS.
Comments
Log in to comment