<jupyter_start><jupyter_text>Throughput- publisher와 subsriber의 throughput 각각 동시에 측정<jupyter_code>import sys
import time
import redis
import datetime
from multiprocessing import Process
import sys

def pub(myredis):     #pubsliher
    start = datetime.datetime.now()
    cnt = 0
    while True:
        myredis.publish('channel', time.time())
        if datetime.datetime.now() > start + datetime.timedelta(seconds=10):  #pubulisher가 10초 동안 msg 보냄
            break
        cnt+=1
    print(f"published {cnt / 10} msgs")  #publisher throughput 측정

def sub(myredis, name):  #subscriber
    pubsub = myredis.pubsub()
    pubsub.subscribe(['channel'])

    start = datetime.datetime.now()
    cnt = 0

    for _ in pubsub.listen():
        cnt+=1
        if datetime.datetime.now() > start + datetime.timedelta(seconds=10):  #10초 동안 subsriber 가 listen
            break

    print(f"sub throughput {cnt / 10}")  #subsriber throughput 측정

if __name__ == '__main__':
    # timer()
    myredis = redis.StrictRedis(socket_timeout=15)  #sub이 for 문에 갇힐 수 있을므로 socket_timeout 설정

    sub = Process(target=sub, kwargs={'myredis': myredis, 'name': 'reader1'}) 
    sub.start()

    pub = Process(target=pub , args=[myredis, ])
    pub.start()

    procs = [sub,pub]

    for proc in procs:
        proc.join()<jupyter_output><empty_output><jupyter_text>Latency 1. publisher<jupyter_code>import time
import redis

def pub(myredis,s=1000):
    for n in range(s):
        myredis.publish('channel',time.time())


if __name__ == '__main__':
    myredis = redis.StrictRedis()
    pub(myredis)<jupyter_output><empty_output><jupyter_text>2. subscriber<jupyter_code>import sys
import time
import redis
import datetime


def sub(myredis, name):
    pubsub = myredis.pubsub()
    pubsub.subscribe(['channel'])

    latency_list = []
    Append = latency_list.append  #append 를 더 빠르게 하기 위함

    try:
        for item in pubsub.listen():
            latency = time.time()- float(item['data'])   #latency 계산
            print(f"arrived time {time.time()} departure time {item['data']}, latency : {latency}")
            Append(latency)    
    except:  #10초 지나면 timeout error 발생
        latency_list= latency_list[1:]  #초기값 제외-> 초기값은 publisher에서 보낸 값이 아니라 다른 값
        print("latency average:" , sum(latency_list) / len(latency_list))   #latency의 평균을 구함


if __name__ == '__main__':

    myredis = redis.StrictRedis(socket_timeout=10)  #subscriber가 10초간 listen, sub이 for문에 갇히는 것 방지
    sub(myredis, 'reader1')<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11858336649423, 'fraction_numerical': 0.013131715081575806, 'mean_word_length': 3.3344827586206898, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 16, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '48149', 'license': 'MIT', 'n_tokens_mistral': 985, 'n_tokens_neox': 991, 'n_words': 212, 'path': 'Redis throughput and latency.ipynb', 'repo_name': 'jsrimr/redis-kafka-rabbit-bench', 'size': '2513'}