pub.py

# pub.py

import json
import redis
import time


r = redis.StrictRedis()
channels = range(1000)
messages_per_channel = 1000

print 'sending...'

for num in range(messages_per_channel):
    for c in channels:
        message = {'signature': 'content', 'time': time.time(), 'num': num}
        r.publish('test-%s' % c, json.dumps(message))

r.publish('test-0', json.dumps({'signature': 'stop', 'time': time.time()}))

print 'sent %s messages' % (len(channels) * messages_per_channel)

sub.py

import json
import redis
import signal
import time


r = redis.StrictRedis()
pubsub = r.pubsub()
pubsub.psubscribe('test-*')

loop = True
tot = 0
avg = 0
sum = 0

with open('logs/sub', 'w') as logfile:
    while loop:
        for m in pubsub.listen():
            if m['type'] != 'pmessage':
                continue
            current_time = time.time()
            decoded = json.loads(m['data'])
            if decoded['signature'] == 'stop':
                loop = False
                break
            sent_time = float(decoded['time'])
            dur = current_time - sent_time
            tot += 1
            sum += dur
            avg = sum / tot

            logfile.write('processing no. %s it took %s current avg: %s\n' % (
                tot, dur, avg))

    print 'run completed, avg latency: %s' % avg