如何在Python中创建简单的网络连接?

4 投票
3 回答
1669 浏览
提问于 2025-04-16 22:40

我想知道有没有人能推荐一些好的资源或者建议,帮助我在Python程序中添加网络功能。具体来说,我想了解一下sockets模块是否最适合我的需求,还有没有什么资源是大家觉得特别有帮助的。

背景: 我正在尝试创建一个幻想足球应用程序(在Windows上),其中有一个“服务器”程序负责进行选秀,而一个“客户端”程序(在远程计算机上运行,不通过局域网连接)可以连接到服务器并接收选秀更新。简单来说,我每分钟只需要传输一两条字符串信息。

我做了一些研究,发现很多人使用内置的sockets模块来完成这样的任务。我不太确定这个模块是否对我这个任务来说太复杂或者太简单,所以任何建议都非常欢迎。

附注: 这个程序的图形界面将使用Tkinter创建,所以我想可能需要实现一些线程来处理Tk和socket的循环,但这属于另一个问题,除非你认为这会直接影响到这个问题。

3 个回答

0

Twisted 这个工具可以让这类事情变得非常简单。不过,它的体积不算小。

1

如果你不想使用Twisted这个库,那就可以选择socket

这个链接里也有一些示例可以参考。

3

最简单的解决办法可能就是让你的服务器变成一个非常基础的XMLRPC服务器。Python里有一个类可以做到这一点(叫做SimpleXMLRPCServer)。然后,让你的客户端每隔几分钟就连接一次这个服务器,获取更新的数据。

想了解更多信息,可以查看这个链接:http://docs.python.org/library/simplexmlrpcserver.html

下面是一个简单粗暴的示例:

服务器代码

from SimpleXMLRPCServer import SimpleXMLRPCServer

# for demonstration purposes, just return an ever increasing integer
score = 0
def get_scores():
    global score
    score += 1
    return score

# create server, register get_scores function
server = SimpleXMLRPCServer(("localhost", 8000))
server.register_function(get_scores, "get_scores")

# start the server
server.serve_forever()

客户端图形界面代码

import Tkinter as tk
import xmlrpclib
import socket

class SampleApp(tk.Tk):
    # url of the server
    url = "http://localhost:8000"

    def __init__(self, *args, **kwargs):
        tk.Tk.__init__(self, *args, **kwargs)

        # initialize connection to server
        self.server = xmlrpclib.ServerProxy(self.url)

        # create GUI
        self.status = tk.Label(text="", anchor="w")
        self.label = tk.Label(text="current score: ?")
        self.status.pack(side="bottom", fill="x")
        self.label.pack(side="top", fill="both", expand=True)
        self.wm_geometry("400x200")

        # wait a second to give the GUI a chance to
        # display, then start fetching the scores 
        # every 5 seconds
        self.after(1000, self.get_latest_scores, 2000)

    def update_status(self, message=""):
        '''Update the statusbar with the given message'''
        self.status.configure(text=message)
        self.update_idletasks()

    def get_latest_scores(self, interval):
        '''Retrieve latest scores and update the UI'''
        try:
            self.update_status("connecting...")
            score = self.server.get_scores()
            self.label.configure(text="current score: %s" % score)
            self.update_status()
        except socket.error, e:
            self.update_status("error: %s" % str(e))
        self.after(interval, self.get_latest_scores, interval)

if __name__ == "__main__":
    app = SampleApp()
    app.mainloop()

撰写回答