from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad

def encrypt_text(text: str, key_hex: str, iv_hex: str) -> str:
    key = bytes.fromhex(key_hex)
    iv = bytes.fromhex(iv_hex)
    text = text.encode('utf-8')
    cipher = AES.new(key, AES.MODE_CBC, iv)
    encrypted_data = cipher.encrypt(pad(text, AES.block_size))
    return encrypted_data.hex()


def decrypt_text(text: str, key_hex: str, iv_hex: str) -> str:
    key = bytes.fromhex(key_hex)
    iv = bytes.fromhex(iv_hex)
    text = bytes.fromhex(text)
    cipher = AES.new(key, AES.MODE_CBC, iv)
    decrypted_data = unpad(cipher.decrypt(text), AES.block_size)
    return decrypted_data.decode('utf-8')

curl -XPOST 'http://127.0.0.1:8000/tag/create' \
-H 'content-type:application/json' \
-d '{"tag":"php"}'

from rest_framework.decorators import api_view
from rest_framework.response import Response

@api_view(["POST"])
def tag_create(request):
    print(request.data)
    tag = request.data.get('tag')

server{
    server_name domain;
    root /var/www/html;
    location / {
        proxy_pass http://127.0.0.1:8080;
        proxy_set_header host $host;
        proxy_set_header X-real-ip $remote_addr;
        proxy_set_header X-forward-for $proxy_add_x_forwarded_for;
    }
}

安装kafka-python

pip install kafka-python

代码

from kafka import KafkaConsumer, TopicPartition

bootstrap_servers = ['localhost:9092']
topic = 'test'
group_id = 'lll'

consumer = KafkaConsumer(
    group_id="my-group",
    bootstrap_servers=bootstrap_servers,
    auto_offset_reset='latest'
)

PARTITIONS = []
for partition in consumer.partitions_for_topic(topic):
    PARTITIONS.append(TopicPartition(topic, partition))

end_offsets = consumer.end_offsets(PARTITIONS)
print(end_offsets)

#输出
#{TopicPartition(topic='test', partition=2): 354915,
#TopicPartition(topic='test', partition=1): 354926,
#TopicPartition(topic='test', partition=0): 354892}

安装kafka库

pip3 install kafka-python

代码

from kafka import KafkaProducer,KafkaConsumer,KafkaAdminClient
import json
import time
topic = 'number'
bootstrap_servers=['127.0.0.1:9092']

消费者

consumer = KafkaConsumer(topic,
                        group_id=group_id,
                         bootstrap_servers=bootstrap_servers)
for message in consumer:
    print ("value=%s" % (message.value))

生产者

producer = KafkaProducer(bootstrap_servers=bootstrap_servers)
for i in range(100):
    producer.send(topic, json.dumps({'number': i}).encode('utf-8'))
producer.flush()

topic增加分区

from kafka.admin import NewPartitions

admin_client = KafkaAdminClient(bootstrap_servers=bootstrap_servers)
topic_partitions = {}
topic_partitions[topic] = NewPartitions(total_count=5)
admin_client.create_partitions(topic_partitions)

列出所有消费者

from kafka import KafkaAdminClient


def list_consumer_groups(broker):
    admin_client = KafkaAdminClient(bootstrap_servers=broker)

    # List all the consumer groups
    groups = admin_client.list_consumer_groups()

    return [group[0] for group in groups]


broker_address = "localhost:9092"
print(list_consumer_groups(broker_address))