python 自建kafka消息生成和消费小工具
要将 Kafka 的消息生产和消费转换为 API 接口,我们可以使用 Python 的 Web 框架。其中 Flask 是一个轻量级且易于使用的选择。下面是一个简单的例子,使用 Flask 创建 API 来生成和消费 Kafka 消息。
1. 安装所需的库:
pip install kafka-python flask
2. 创建 Flask API:
from flask import Flask, request, jsonify
from kafka import KafkaProducer, KafkaConsumerapp = Flask(__name__)# 配置 Kafka
KAFKA_BROKER_URL = 'localhost:9092'
TOPIC = 'test_topic'
producer = KafkaProducer(bootstrap_servers=KAFKA_BROKER_URL)@app.route('/send', methods=['POST'])
def send_message():message = request.json.get('message')if message:producer.send(TOPIC, value=message.encode('utf-8'))return jsonify({"status": "success", "message": "Message sent!"}), 200else:return jsonify({"status": "error", "message": "Message cannot be empty!"}), 400@app.route('/receive', methods=['GET'])
def receive_message():consumer = KafkaConsumer(TOPIC, bootstrap_servers=KAFKA_BROKER_URL, auto_offset_reset='earliest')messages = []for message in consumer:messages.append(message.value.decode('utf-8'))if len(messages) > 5: # 只收集最近的5条消息,可以根据需要调整breakreturn jsonify(messages)if __name__ == '__main__':app.run(debug=True, port=5000)
这个 Flask 应用程序定义了两个端点:
/send: 它接受 POST 请求并发送消息到 Kafka。/receive: 它返回 Kafka 主题中的最近消息。
3. 使用 API:
- 发送消息:
curl -X POST http://localhost:5000/send -H "Content-Type: application/json" -d '{"message": "Hello, Kafka!"}'
- 接收消息:
curl http://localhost:5000/receive
这只是一个简单的示例,您可能需要添加错误处理、日志记录、认证、消息序列化和反序列化等功能,以满足更复杂的需求。
相关文章:
python 自建kafka消息生成和消费小工具
要将 Kafka 的消息生产和消费转换为 API 接口,我们可以使用 Python 的 Web 框架。其中 Flask 是一个轻量级且易于使用的选择。下面是一个简单的例子,使用 Flask 创建 API 来生成和消费 Kafka 消息。 1. 安装所需的库: pip install kafka-py…...
Prim算法:经过图中所有节点的最短路径
题目链接:53. 寻宝(第七期模拟笔试) #include<bits/stdc.h> using namespace std;// v为节点数量,e为边数量 int v, e;// 最小生成树 void prim(vector<vector<int>>& adj) {vector<int> dist(v1, I…...
Linux 信号捕捉函数 signal sigaction
signal函数 #include <signal.h> typedef void (*sighandler_t)(int); sighandler_t signal(int signum, sighandler_t handler); 功能:设置某个信号的捕捉行为 参数: -signum:要捕捉的信号 handler:对捕捉到的信号怎么处理…...
StarRocks操作笔记
最近在使用starRocks,记录一些临时的操作技巧,防止遗忘。 1. 创建表 CREATE TABLE IF NOT EXISTS ODS.T_TEST( pk_day date, pool_address string, code string comment 唯一主键, test1 string, test2 string, test3 string, pk_year varchar(4), pk_m…...
Linux的ls -ld命令产生的信息怎么看
2023年9月24日,周日上午 目录 ls -ld列出的目录或文件的信息含义文件硬链接什么是文件硬链接为什么新建目录的文件硬链接为2举例说明例一例二例三 ls -ld列出的目录或文件的信息含义 第一个字符表示文件类型: d: 目录 -: 普通文件 l: 软链接 b: 块设备文件 c:…...
Linux- 内存映射文件(Memory-Mapped File)
内存映射文件(Memory-Mapped File)是⼀种将文件内容映射到内存中的机制,允许程序直接访问文件数据,就好像这些数据已经被加载到了内存⼀样。这个机制允许文件的内容被映射到⼀个进程的地址空间,从而允许程序以⼀种更高…...
李航老师《统计学习方法》第五章阅读笔记
决策树(decision tree)是一种基本的分类与回归方法。本章主要讨论用于分类的决策树。决策树模型呈树形结构,在分类问题中,表示基于特征对实例进行分类的过程。 以下是关于分类决策树的一些基本概念和特点: 树形结构&am…...
iOS16新特性:实时活动-在锁屏界面实时更新APP消息 | 京东云技术团队
简介 之前在 《iOS16新特性:灵动岛适配开发与到家业务场景结合的探索实践》 里介绍了iOS16新的特性:实时更新(Live Activity)中灵动岛的适配流程,但其实除了灵动岛的展示样式,Live Activity还有一种非常实用的应用场景…...
使用 Elasticsearch、OpenAI 和 LangChain 进行语义搜索
在本教程中,我将引导您使用 Elasticsearch、OpenAI、LangChain 和 FastAPI 构建语义搜索服务。 LangChain 是这个领域的新酷孩子。 它是一个旨在帮助你与大型语言模型 (LLM) 交互的库。 LangChain 简化了与 LLMs 相关的许多日常任务,例如从文档中提取文本…...
NIFI集群_队列Queue中数据无法清空_清除队列数据报错_无法删除queue_解决_集群中机器交替重启删除---大数据之Nifi工作笔记0061
今天发现,有两个处理器,启动以后,数据流不过去,后来,锁定问题在,queue队列上面,因为别的队列都可以通过,右键,empty queue清空,就是 这个队列不行,这个队列无法被删除,至于为什么导致这样的, 猜测是因为之前,流程设计好以后,队列没有设置背压,也没有设置队列中的内容大小和fl…...
leetcode20. 有效的括号 [简单题]
题目 给定一个只包括 (,),{,},[,] 的字符串 s ,判断字符串是否有效。 有效字符串需满足: 左括号必须用相同类型的右括号闭合。左括号必须以正确的顺序闭合。每个右括号都有一个对应的相同类型…...
ubuntu20.04下源码编译colmap
由于稠密重建需要CUDA,因此先安装CUDA,我使用的是3050GPU,nvidia-smi显示最高支持CUDA11.4。 不要用sudo apt安装,版本较低,30系显卡建议安装CUDA11.0以上,这里安装了11.1版本。 下载: cuda_1…...
Jumpserver堡垒机
一、堡垒机概述 1、堡垒机的基本概念 堡垒机也是一台服务器,在一个特定的网络环境下,为了保障网络和数据不受来自外部和内部用户的入侵和破坏,而运用各种技术手段实时收集、监控网络环境中每一个组成部分(服务器)的系…...
第一百五十三回 如何实现滑动窗口
文章目录 概念介绍实现方法示例代码 我们在上一章回中介绍了自定义组件实现游戏摇杆相关的内容,本章回中将介绍 如何实现滑动窗口.闲话休提,让我们一起Talk Flutter吧。 概念介绍 我们在本章回中介绍的滑动窗口表示在屏幕底部向上滑动时弹出一个窗口&a…...
Oracle 12c自动化管理特性的新进展:自动备份、自动恢复和自动维护功能的优势|oracle 12c相对oralce 11g的新特性(3)
一、前言: 前面几期讲解了oracle 12c多租户的使用、In-Memory列存储来提高查询性能以及数据库的克隆、全局数据字典和共享数据库资源的使用 今天我们讲讲oracle 12c的另外的一个自动化管理功能新特性:自动备份、自动恢复、自动维护的功能 二、自动备份、自动恢复、自动维护…...
Redis——Jedis中hash类型使用
hset 和 hget hset可以逐一添加key和value,也可以通过map类型来直接添加多组fields 而hget则返回string类型,如果元素不存在则返回null private static void hsetAndHget(Jedis jedis) {jedis.flushAll();jedis.hset("key", "f1"…...
肖sir__项目实战讲解__004
项目实战讲解 一、项目的类型 金融类: 保险(健康险理财险)、证券、基金(股票型基金、混合型基金、指数型基金、债券型基金、 天天基金网(ETF基金、货币型基金、量化基金)、银行、贷款、信用卡、外汇、二元期权、期货原油、blockchain、 数字货币、黄金白…...
数据库数据恢复-ORACLE常见故障有哪些?恢复数据的可能性高吗?
ORACLE数据库常见故障: 1、ORACLE数据库无法启动或无法正常工作。 2、ORACLE数据库ASM存储破坏。 3、ORACLE数据库数据文件丢失。 4、ORACLE数据库数据文件部分损坏。 5、ORACLE数据库DUMP文件损坏。 ORACLE数据库数据恢复可能性分析: 1、ORACLE数据库无…...
合规性管理如何帮助产品团队按时交付?
成功的产品和产品发布背后通常需要经过一个涉及多个监督机构、多功能团队和利益相关者的复杂流程。在组织的治理、风险管理和合规性(GRC)框架下,产品团队不仅需要追求市场创新,还需要确保符合所有适用的法规、标准和合同要求。由于…...
从平均数到排名算法
平均数用更少的数字,概括一组数字。属于概述统计量、集中趋势测度、位置测度。中位数是第二常见的概述统计量。许多情况下比均值更合适。算术平均数是3中毕达哥拉斯平均数之一,另外两种毕达哥拉斯平均数是几何平均数和调和平均数。 算术平均 A M 1 n ∑…...
centos 7 部署awstats 网站访问检测
一、基础环境准备(两种安装方式都要做) bash # 安装必要依赖 yum install -y httpd perl mod_perl perl-Time-HiRes perl-DateTime systemctl enable httpd # 设置 Apache 开机自启 systemctl start httpd # 启动 Apache二、安装 AWStats࿰…...
家政维修平台实战20:权限设计
目录 1 获取工人信息2 搭建工人入口3 权限判断总结 目前我们已经搭建好了基础的用户体系,主要是分成几个表,用户表我们是记录用户的基础信息,包括手机、昵称、头像。而工人和员工各有各的表。那么就有一个问题,不同的角色…...
Linux --进程控制
本文从以下五个方面来初步认识进程控制: 目录 进程创建 进程终止 进程等待 进程替换 模拟实现一个微型shell 进程创建 在Linux系统中我们可以在一个进程使用系统调用fork()来创建子进程,创建出来的进程就是子进程,原来的进程为父进程。…...
(一)单例模式
一、前言 单例模式属于六大创建型模式,即在软件设计过程中,主要关注创建对象的结果,并不关心创建对象的过程及细节。创建型设计模式将类对象的实例化过程进行抽象化接口设计,从而隐藏了类对象的实例是如何被创建的,封装了软件系统使用的具体对象类型。 六大创建型模式包括…...
Cilium动手实验室: 精通之旅---13.Cilium LoadBalancer IPAM and L2 Service Announcement
Cilium动手实验室: 精通之旅---13.Cilium LoadBalancer IPAM and L2 Service Announcement 1. LAB环境2. L2公告策略2.1 部署Death Star2.2 访问服务2.3 部署L2公告策略2.4 服务宣告 3. 可视化 ARP 流量3.1 部署新服务3.2 准备可视化3.3 再次请求 4. 自动IPAM4.1 IPAM Pool4.2 …...
pgsql:还原数据库后出现重复序列导致“more than one owned sequence found“报错问题的解决
问题: pgsql数据库通过备份数据库文件进行还原时,如果表中有自增序列,还原后可能会出现重复的序列,此时若向表中插入新行时会出现“more than one owned sequence found”的报错提示。 点击菜单“其它”-》“序列”,…...
【若依】框架项目部署笔记
参考【SpringBoot】【Vue】项目部署_no main manifest attribute, in springboot-0.0.1-sn-CSDN博客 多一个redis安装 准备工作: 压缩包下载:http://download.redis.io/releases 1. 上传压缩包,并进入压缩包所在目录,解压到目标…...
Java中栈的多种实现类详解
Java中栈的多种实现类详解:Stack、LinkedList与ArrayDeque全方位对比 前言一、Stack类——Java最早的栈实现1.1 Stack类简介1.2 常用方法1.3 优缺点分析 二、LinkedList类——灵活的双端链表2.1 LinkedList类简介2.2 常用方法2.3 优缺点分析 三、ArrayDeque类——高…...
Python异步编程:深入理解协程的原理与实践指南
💝💝💝欢迎莅临我的博客,很高兴能够在这里和您见面!希望您在这里可以感受到一份轻松愉快的氛围,不仅可以获得有趣的内容和知识,也可以畅所欲言、分享您的想法和见解。 持续学习,不断…...
linux设备重启后时间与网络时间不同步怎么解决?
linux设备重启后时间与网络时间不同步怎么解决? 设备只要一重启,时间又错了/偏了,明明刚刚对时还是对的! 这在物联网、嵌入式开发环境特别常见,尤其是开发板、树莓派、rk3588 这类设备。 解决方法: 加硬件…...
