使用Tkinter写一个发送kafka消息的工具

背景

公司是做AR实景产品的，近几年无人机特别的火，一来公司比较关注低空经济这个新型领域，二来很多政企、事业单位都采购了无人机用于日常工作。那么就衍生了将无人机接入到AR实景项目中的需求。

本工具的作用就是在AR实景项目中模拟无人机飞行时候产生的飞控数据。

工具界面展示

功能

• 用户填写kafka连接地址，
• 用户参考消息示例模板编写消息内容，并发送到kafka 的 topic 中。
• 所有操作的log输出到用户界面

代码讲解

运行环境

• python：3.9.7
• pip：pip3(pip 21.2.3 )
• OS：win11
• GUI：TKinter
• kafka-client：confluent_kafka

创建GUI程序

• 通过 tk.Tk() 实例化一个GUI程序。
• 通过 ttk.Style() 实例化主题并通过style.configure 设置公共样式。
• 通过 KafkaMessageSenderApp(root) 的构造函数将root实例传入业务代码中
• 最后通过root.mainloop()让程序保持前台运行

if __name__ == "__main__":# 设置样式root = tk.Tk()style = ttk.Style()# 尝试配置主题try:if "clam" in style.theme_names():style.theme_use("clam")except:pass# 自定义样式style.configure("TLabel", font=("SimHei", 10))style.configure("TButton", font=("SimHei", 10))style.configure("TEntry", font=("SimHei", 10))style.configure("TFrame", background="#f0f0f0")style.configure("TLabelframe", background="#f0f0f0")style.configure("TLabelframe.Label", background="#f0f0f0", font=("SimHei", 10, "bold"))# 创建强调按钮样式style.configure("Accent.TButton", foreground="white", background="#4a86e8")app = KafkaMessageSenderApp(root)root.mainloop()

搭建前端样式

• 在KafkaMessageSenderApp 的构造函数中，通过create_widgets函数完成前端样式的编写，例如 label 的位置，button 的名字及绑定的回调处理函数
• 通过 start_log_consumer 处理日志
• 通过generate_example_message生成模板消息

self.create_widgets()self.start_log_consumer()# 生成示例消息self.generate_example_message()

编写功能实现代码

通过为 前端各种控件绑定回调函数来实现业务代码的编写。例如通过为 发送消息 按钮绑定回调处理消息的发送。

按钮样式和回调函数绑定：

# 按钮区域button_frame = ttk.Frame(main_frame)button_frame.pack(fill=tk.X, pady=5)# 创建按钮并保存引用self.send_button = ttk.Button(button_frame,text="发送消息",command=self.send_message,width=20,style="Accent.TButton",)

发送消息：

def send_message(self):"""发送消息到Kafka"""if self.producer is None:messagebox.showerror("错误", "请先连接到Kafka服务器")returntopic = self.topic_var.get()if not topic:messagebox.showerror("错误", "请输入Topic")returnmessage_text = self.message_text.get(1.0, tk.END).strip()if not message_text:messagebox.showerror("错误", "消息内容不能为空")return# 解析JSONtry:message_data = json.loads(message_text)except json.JSONDecodeError as e:messagebox.showerror("JSON解析错误", f"无效的JSON格式: {str(e)}")return# 在新线程中发送消息，避免阻塞UIthreading.Thread(target=self._send_message_thread, args=(topic, message_data), daemon=True).start()