시행착오 기술 블로그

https://youtu.be/XTsTbqjJhjE

https://dataq.goorm.io/exam/3/%EC%B2%B4%ED%97%98%ED%95%98%EA%B8%B0/quiz/1

> 1유형 

  - https://youtu.be/XWnSx21eZDA?si=o-pf2dCWHwY4TzPv

  - https://youtu.be/lKrdDyr8_rQ?si=rFcCCqgHV3esF50r

> 2유형

  - https://youtu.be/ePid-6QSoYc?si=xlpsG7QN7M7j6L0I

https://youtu.be/EtPuy__yP1E?si=gawMn0gmU4Yo2gam

> 3유형

 -  https://youtu.be/vPx2li_OBrw?si=4FiwIOYx2uCbgb7k

유형1

https://youtu.be/3oyVIIA28vs?si=TlQi9lqpDAWfHZMU

유형2

https://youtu.be/NUY53PbYDDw?si=-rWnCf_Mb0h2-i90

유형1

https://youtu.be/IdP7NAxMxoc?si=1leKxOVfDpGz1u9X

유형2

https://youtu.be/RHFQJZjqfFA?si=AVg-5YLy00nZZ8kz

#!/usr/bin/env python

import sys
import time
from random import choice
from argparse import ArgumentParser, FileType
from configparser import ConfigParser
from confluent_kafka import Producer

if __name__ == '__main__':
    # Record the start time
    start_time = time.time()

    # Parse the command line.
    parser = ArgumentParser()
    parser.add_argument('config_file', type=FileType('r'))
    args = parser.parse_args()

    # Parse the configuration.
    # See https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
    config_parser = ConfigParser()
    config_parser.read_file(args.config_file)
    config = dict(config_parser['default'])

    # Create Producer instance
    producer = Producer(config)

    # Optional per-message delivery callback (triggered by poll() or flush())
    # when a message has been successfully delivered or permanently
    # failed delivery (after retries).
    def delivery_callback(err, msg):
        if err:
            print('ERROR: Message failed delivery: {}'.format(err))
        else:
            print("Produced event to topic {topic}: key = {key:12} value = {value:12}".format(
                topic=msg.topic(), key=msg.key().decode('utf-8'), value=msg.value().decode('utf-8')))

    # Produce data by selecting random values from these lists.
    topic = "purchases"
    user_ids = ['eabara', 'jsmith', 'sgarcia', 'jbernard', 'htanaka', 'awalther']
    products = ['book', 'alarm clock', 't-shirts', 'gift card', 'batteries']

    count = 0
    for _ in range(100):  # Change the range to 100 for 100 messages
        user_id = choice(user_ids)
        product = choice(products)
        producer.produce(topic, product, user_id, callback=delivery_callback)

        # Introduce a delay of 0.1 seconds
        time.sleep(0.1)
        count += 1

    # Block until the messages are sent.
    producer.flush()

    # Record the end time
    end_time = time.time()

    # Calculate and print the total execution time
    total_time = end_time - start_time
    print(f'Total execution time: {total_time:.2f} seconds')

#!/usr/bin/env python

import sys
import time
from argparse import ArgumentParser, FileType
from configparser import ConfigParser
from confluent_kafka import Consumer, OFFSET_BEGINNING

if __name__ == '__main__':
    # Record the start time
    start_time = time.time()
    start_receive_time = time.time()

    # Parse the command line.
    parser = ArgumentParser()
    parser.add_argument('config_file', type=FileType('r'))
    parser.add_argument('--reset', action='store_true')
    args = parser.parse_args()

    # Parse the configuration.
    # See https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
    config_parser = ConfigParser()
    config_parser.read_file(args.config_file)
    config = dict(config_parser['default'])
    config.update(config_parser['consumer'])

    # Create Consumer instance
    consumer = Consumer(config)

    # Set up a callback to handle the '--reset' flag.
    def reset_offset(consumer, partitions):
        if args.reset:
            for p in partitions:
                p.offset = OFFSET_BEGINNING
            consumer.assign(partitions)

    # Subscribe to topic
    topic = "purchases"
    consumer.subscribe([topic], on_assign=reset_offset)
    wait = False

    # Poll for new messages from Kafka and print them.
    try:
        while True:
            msg = consumer.poll(1.0)
            if msg is None:
                # Initial message consumption may take up to
                # `session.timeout.ms` for the consumer group to
                # rebalance and start consuming
                print("Waiting...")

                if not wait:
                    # Calculate and print the waiting time
                    end_receive_time = time.time()
                    transfer_duration = end_receive_time - start_receive_time
                    print(f'Receive duration: {transfer_duration-1:.2f} seconds')
                    wait = True
            elif msg.error():
                print("ERROR: {}".format(msg.error()))
            else:
                # Extract the (optional) key and value, and print.
                start_waiting_time = time.time()  # Record the start time of waiting
                print("Consumed event from topic {topic}: key = {key:12} value = {value:12}".format(
                    topic=msg.topic(), key=msg.key().decode('utf-8'), value=msg.value().decode('utf-8')))
                time.sleep(0.1)
                wait = False

    except KeyboardInterrupt:
        pass
    finally:
        # Leave group and commit final offsets
        consumer.close()

        # Record the end time
        end_time = time.time()

        # Calculate and print the total execution time
        total_time = end_time - start_time
        print(f'Total execution time: {total_time:.2f} seconds')

#!/usr/bin/env python

import sys
import time
from argparse import ArgumentParser, FileType
from configparser import ConfigParser
from confluent_kafka import Consumer, KafkaError

if __name__ == '__main__':
    # Record the start time
    start_time = time.time()
    start_receive_time = time.time()

    # Parse the command line.
    parser = ArgumentParser()
    parser.add_argument('config_file', type=FileType('r'))
    args = parser.parse_args()

    # Parse the configuration.
    config_parser = ConfigParser()
    config_parser.read_file(args.config_file)
    config = dict(config_parser['default'])
    config.update(config_parser['consumer'])

    # Set unique group.id for each consumer group
    config['group.id'] = f"{config['group.id']}_{time.time()}"

    # Create Consumer instance
    consumer = Consumer(config)

    # Subscribe to the topic
    topic = "purchases"
    consumer.subscribe([topic])
    wait = False

    # Poll for new messages from Kafka and print them.
    try:
        while True:
            msg = consumer.poll(1.0)
            if msg is None:
                # No new messages within the timeout
                print("Waiting...")

                if not wait:
                    # Calculate and print the waiting time
                    end_receive_time = time.time()
                    transfer_duration = end_receive_time - start_receive_time
                    print(f'Receive duration: {transfer_duration - 1:.2f} seconds')
                    wait = True
            elif msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    # End of partition event (not an error)
                    continue
                else:
                    print("ERROR: {}".format(msg.error()))
            else:
                # Extract the (optional) key and value, and print.
                print("[Consumer 1] Consumed event from topic {topic}, partition {partition}, offset {offset}: key = {key:12} value = {value:12}".format(
                    topic=msg.topic(), partition=msg.partition(), offset=msg.offset(),
                    key=msg.key().decode('utf-8'), value=msg.value().decode('utf-8')))
                time.sleep(0.1)
                if wait:
                    wait = False
                    start_receive_time = time.time()

    except KeyboardInterrupt:
        pass
    finally:
        # Leave group and commit final offsets
        consumer.close()

        # Record the end time
        end_time = time.time()

        # Calculate and print the total execution time
        total_time = end_time - start_time
        print(f'Total execution time: {total_time:.2f} seconds')

#!/usr/bin/env python

import sys
import time
from argparse import ArgumentParser, FileType
from configparser import ConfigParser
from confluent_kafka import Consumer, KafkaError

if __name__ == '__main__':
    # Record the start time
    start_time = time.time()
    start_receive_time = time.time()

    # Parse the command line.
    parser = ArgumentParser()
    parser.add_argument('config_file', type=FileType('r'))
    args = parser.parse_args()

    # Parse the configuration.
    config_parser = ConfigParser()
    config_parser.read_file(args.config_file)
    config = dict(config_parser['default'])
    config.update(config_parser['consumer'])

    # Set unique group.id for each consumer group
    config['group.id'] = f"{config['group.id']}_{time.time()}"

    # Create Consumer instance
    consumer = Consumer(config)

    # Subscribe to the topic
    topic = "purchases"
    consumer.subscribe([topic])
    wait = False

    # Poll for new messages from Kafka and print them.
    try:
        while True:
            msg = consumer.poll(1.0)
            if msg is None:
                # No new messages within the timeout
                print("Waiting...")

                if not wait:
                    # Calculate and print the waiting time
                    end_receive_time = time.time()
                    transfer_duration = end_receive_time - start_receive_time
                    print(f'Receive duration: {transfer_duration - 1:.2f} seconds')
                    wait = True
            elif msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    # End of partition event (not an error)
                    continue
                else:
                    print("ERROR: {}".format(msg.error()))
            else:
                # Extract the (optional) key and value, and print.
                print("[Consumer 2] Consumed event from topic {topic}, partition {partition}, offset {offset}: key = {key:12} value = {value:12}".format(
                    topic=msg.topic(), partition=msg.partition(), offset=msg.offset(),
                    key=msg.key().decode('utf-8'), value=msg.value().decode('utf-8')))
                time.sleep(0.1)
                if wait:
                    wait = False
                    start_receive_time = time.time()
    except KeyboardInterrupt:
        pass
    finally:
        # Leave group and commit final offsets
        consumer.close()

        # Record the end time
        end_time = time.time()

        # Calculate and print the total execution time
        total_time = end_time - start_time
        print(f'Total execution time: {total_time:.2f} seconds')

#!/usr/bin/env python

import sys
import time
from random import choice
from argparse import ArgumentParser, FileType
from configparser import ConfigParser
from confluent_kafka import Producer
import threading

def produce_messages(producer, topic, user_ids, products, count):
    for _ in range(count):
        user_id = user_ids
        product = choice(products)
        producer.produce(topic, product, user_id, callback=delivery_callback)
        time.sleep(0.1)

def delivery_callback(err, msg):
    if err:
        print('ERROR: Message failed delivery: {}'.format(err))
    else:
        print("Produced event to topic {topic}: key = {key:12} value = {value:12}".format(
            topic=msg.topic(), key=msg.key().decode('utf-8'), value=msg.value().decode('utf-8')))

if __name__ == '__main__':
    # Record the start time
    start_time = time.time()

    # Parse the command line.
    parser = ArgumentParser()
    parser.add_argument('config_file', type=FileType('r'))
    args = parser.parse_args()

    # Parse the configuration.
    # See https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
    config_parser = ConfigParser()
    config_parser.read_file(args.config_file)
    config = dict(config_parser['default'])

    # Create Producer instances
    producer1 = Producer(config)
    producer2 = Producer(config)

    # Produce data by selecting random values from these lists.
    topic = "purchases"
    data_1 = ['APPLE', 'BANANA', 'COCONUT']
    data_2 = ['dog', 'cat', 'giraffe', 'pig']

    count = 50  # Each producer will produce 50 messages

    # Create threads for each producer
    thread1 = threading.Thread(target=produce_messages, args=(producer1, topic, 'producer_1', data_1, count))
    thread2 = threading.Thread(target=produce_messages, args=(producer2, topic, 'producer_2', data_2, count))

    # Start the threads
    thread1.start()
    thread2.start()

    # Wait for the threads to finish
    thread1.join()
    thread2.join()

    # Block until the messages are sent.
    producer1.flush()
    producer2.flush()

    # Record the end time
    end_time = time.time()

    # Calculate and print the total execution time
    total_time = end_time - start_time
    print(f'Total execution time: {total_time:.2f} seconds')

#!/usr/bin/env python

import sys
import time
from random import choice
from argparse import ArgumentParser, FileType
from configparser import ConfigParser
from confluent_kafka import Producer
import threading

def produce_messages(producer, topic, user_ids, products, count, partition):
    for _ in range(count):
        user_id = user_ids
        product = choice(products)
        producer.produce(topic, key=user_id, value=product, partition=partition, callback=delivery_callback)
        time.sleep(0.1)

def delivery_callback(err, msg):
    if err:
        print('ERROR: Message failed delivery: {}'.format(err))
    else:
        print("Produced event to topic {topic}: key = {key:12} value = {value:12}".format(
            topic=msg.topic(), key=msg.key().decode('utf-8'), value=msg.value().decode('utf-8')))

if __name__ == '__main__':
    # Record the start time
    start_time = time.time()

    # Parse the command line.
    parser = ArgumentParser()
    parser.add_argument('config_file', type=FileType('r'))
    args = parser.parse_args()

    # Parse the configuration.
    # See https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
    config_parser = ConfigParser()
    config_parser.read_file(args.config_file)
    config = dict(config_parser['default'])

    # Create Producer instances
    producer1 = Producer(config)
    producer2 = Producer(config)

    # Produce data by selecting random values from these lists.
    topic = "purchases_partition"

    data_1 = ['APPLE', 'BANANA', 'COCONUT']
    data_2 = ['dog', 'cat', 'giraffe', 'pig']

    count = 50  # Each producer will produce 50 messages

    # Create threads for each producer
    thread1 = threading.Thread(target=produce_messages, args=(producer1, topic, 'producer_1', data_1, count, 0))
    thread2 = threading.Thread(target=produce_messages, args=(producer2, topic, 'producer_2', data_2, count, 1))

    # Start the threads
    thread1.start()
    thread2.start()

    # Wait for the threads to finish
    thread1.join()
    thread2.join()

    # Block until the messages are sent.
    producer1.flush()
    producer2.flush()

    # Record the end time
    end_time = time.time()

    # Calculate and print the total execution time
    total_time = end_time - start_time
    print(f'Total execution time: {total_time:.2f} seconds')

#!/usr/bin/env python

import sys
import time
from argparse import ArgumentParser, FileType
from configparser import ConfigParser
from confluent_kafka import Consumer, KafkaError, TopicPartition

if __name__ == '__main__':
    # Record the start time
    start_time = time.time()
    start_receive_time = time.time()

    # Parse the command line.
    parser = ArgumentParser()
    parser.add_argument('config_file', type=FileType('r'))
    args = parser.parse_args()

    # Parse the configuration.
    config_parser = ConfigParser()
    config_parser.read_file(args.config_file)
    config = dict(config_parser['default'])
    config.update(config_parser['consumer'])

    # Set unique group.id for each consumer group
    config['group.id'] = f"{config['group.id']}_{time.time()}"

    # Create Consumer instance
    consumer = Consumer(config)

    # Specify the topic and partition to consume from
    topic = "purchases_partition"
    partition = 0  # Replace with the desired partition

    # Assign the consumer to the specified topic and partition
    consumer.assign([TopicPartition(topic, partition)])

    wait = False

    # Poll for new messages from Kafka and print them.
    try:
        while True:
            msg = consumer.poll(1.0)
            if msg is None:
                # No new messages within the timeout
                print("Waiting...")

                if not wait:
                    # Calculate and print the waiting time
                    end_receive_time = time.time()
                    transfer_duration = end_receive_time - start_receive_time
                    print(f'Receive duration: {transfer_duration - 1:.2f} seconds')
                    wait = True
            elif msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    # End of partition event (not an error)
                    continue
                else:
                    print("ERROR: {}".format(msg.error()))
            else:
                # Extract the (optional) key and value, and print.
                print("[Consumer 1] Consumed event from topic {topic}, partition {partition}, offset {offset}: key = {key:12} value = {value:12}".format(
                    topic=msg.topic(), partition=msg.partition(), offset=msg.offset(),
                    key=msg.key().decode('utf-8'), value=msg.value().decode('utf-8')))
                time.sleep(0.1)
                if wait:
                    wait = False
                    start_receive_time = time.time()

    except KeyboardInterrupt:
        pass
    finally:
        # Leave group and commit final offsets
        consumer.close()

        # Record the end time
        end_time = time.time()

        # Calculate and print the total execution time
        total_time = end_time - start_time
        print(f'Total execution time: {total_time:.2f} seconds')

#!/usr/bin/env python

import sys
import time
from argparse import ArgumentParser, FileType
from configparser import ConfigParser
from confluent_kafka import Consumer, KafkaError, TopicPartition

if __name__ == '__main__':
    # Record the start time
    start_time = time.time()
    start_receive_time = time.time()

    # Parse the command line.
    parser = ArgumentParser()
    parser.add_argument('config_file', type=FileType('r'))
    args = parser.parse_args()

    # Parse the configuration.
    config_parser = ConfigParser()
    config_parser.read_file(args.config_file)
    config = dict(config_parser['default'])
    config.update(config_parser['consumer'])

    # Set unique group.id for each consumer group
    config['group.id'] = f"{config['group.id']}_{time.time()}"

    # Create Consumer instance
    consumer = Consumer(config)

    # Specify the topic and partition to consume from
    topic = "purchases_partition"
    partition = 1  # Replace with the desired partition

    # Assign the consumer to the specified topic and partition
    consumer.assign([TopicPartition(topic, partition)])

    wait = False

    # Poll for new messages from Kafka and print them.
    try:
        while True:
            msg = consumer.poll(1.0)
            if msg is None:
                # No new messages within the timeout
                print("Waiting...")

                if not wait:
                    # Calculate and print the waiting time
                    end_receive_time = time.time()
                    transfer_duration = end_receive_time - start_receive_time
                    print(f'Receive duration: {transfer_duration - 1:.2f} seconds')
                    wait = True
            elif msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    # End of partition event (not an error)
                    continue
                else:
                    print("ERROR: {}".format(msg.error()))
            else:
                # Extract the (optional) key and value, and print.
                print("[Consumer 2] Consumed event from topic {topic}, partition {partition}, offset {offset}: key = {key:12} value = {value:12}".format(
                    topic=msg.topic(), partition=msg.partition(), offset=msg.offset(),
                    key=msg.key().decode('utf-8'), value=msg.value().decode('utf-8')))
                time.sleep(0.1)
                if wait:
                    wait = False
                    start_receive_time = time.time()

    except KeyboardInterrupt:
        pass
    finally:
        # Leave group and commit final offsets
        consumer.close()

        # Record the end time
        end_time = time.time()

        # Calculate and print the total execution time
        total_time = end_time - start_time
        print(f'Total execution time: {total_time:.2f} seconds')