Python基於socket編程（UDP版聊天）

时间 2021-01-06 标签 python socket編程 UDP

説明：

多臺電腦連同一網絡可通信；
有界面版來自辛勤的她，但是不完善；

一、UDP聊天（無界面）

1. 架构

2. 代碼

UDP服务端代码：

import socket

# 创建套接字
udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

# 绑定固定端口
udp_socket.bind(('', 6665))

# 回射接收到的消息
while True:
    # 接收消息
    recv_data, remote_address = udp_socket.recvfrom(4096)
    print("接收到了%s发来的消息%s" % (str(remote_address), recv_data.decode()))

    # 回射
udp_socket.sendto(recv_data, remote_address)

客戶端代碼：

import socket

def send_message(udp_socket):
    # 输入的内容
    data = input("请输入您要发送的信息内容：")
    # ip
    ip = input("请输入对方的ip地址：")
    # 端口
    port = int(input("请输入端口："))

    # 发送数据
    udp_socket.sendto(data.encode(), (ip, port))

def receive_message(udp_socket):
    recv_data, romote_address = udp_socket.recvfrom(4096)
    print("收到了来自%s的回信%s" % (str(romote_address), recv_data.decode()))

def main():
    # 创建套接字
    udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

    # 绑定固定端口
    udp_socket.bind(("", 6667))

    while True:
        # 输入指示
        command = input("请输入您要执行的功能 1.发送消息 2.接收消息 3.退出")
        # 发送数据
        if command == '1':
            send_message(udp_socket)
        # 接收数据
        elif command == '2':
            receive_message(udp_socket)
        # 退出
        elif command == '3':
            break
        else:
            print("输入有误，请重新输入")

if __name__ == '__main__':
    main()

3. 測試

運行服務器端等待

運行客戶端

輸入1 選擇發送消息

向本機發送了消息

與同學連上同一個網絡
互相發送消息

二、UDP聊天（可视界面）

1. 架构

服務端界面：

客戶端界面：

2. 代碼

服務器（server）

import wx
import socket
import threading
from server import noname


class udp_server_ui(noname.MyFrame1):
    def __init__(self, parent):
        noname.MyFrame1.__init__(self, parent)

    def btn_login(self, event):
        ipStr = self.ip_text.GetValue()  # 从输入端中获取ip
        # 从输入端中获取端口，注意端口取得时候不能被占用（可以取12345）
        portStr = int(self.port_text.GetValue())
        printStr = "服务器启动成功\n"  # 是否连接成功
        self.showmessage.AppendText(printStr)  # 显示在信息窗口中
        # 创建服务器端


def main():
    app = wx.App(False)
    frame = udp_server_ui(None)
    frame.Show(True)
    # 启动窗口
    app.MainLoop()


if __name__ == "__main__":
    try:
        main()
    finally:
        pass

noname（server）

# -*- coding: utf-8 -*- 

###########################################################################
## Python code generated with wxFormBuilder (version Jan 23 2018)
## http://www.wxformbuilder.org/
##
## PLEASE DO *NOT* EDIT THIS FILE!
###########################################################################

import wx
import wx.xrc

###########################################################################
## Class MyFrame1
###########################################################################

class MyFrame1 ( wx.Frame ):
	
	def __init__( self, parent ):
		wx.Frame.__init__ ( self, parent, id = wx.ID_ANY, title = wx.EmptyString, pos = wx.DefaultPosition, size = wx.Size( 472,466 ), style = wx.DEFAULT_FRAME_STYLE|wx.TAB_TRAVERSAL )
		
		self.SetSizeHints( wx.DefaultSize, wx.DefaultSize )
		
		bSizer1 = wx.BoxSizer( wx.VERTICAL )
		
		bSizer2 = wx.BoxSizer( wx.VERTICAL )
		
		bSizer8 = wx.BoxSizer( wx.HORIZONTAL )
		
		self.m_staticText1 = wx.StaticText( self, wx.ID_ANY, u"IP：", wx.DefaultPosition, wx.DefaultSize, 0 )
		self.m_staticText1.Wrap( -1 )
		bSizer8.Add( self.m_staticText1, 0, wx.ALIGN_CENTER_VERTICAL|wx.TOP|wx.BOTTOM|wx.LEFT, 5 )
		
		self.ip_text = wx.TextCtrl( self, wx.ID_ANY, wx.EmptyString, wx.DefaultPosition, wx.DefaultSize, 0 )
		bSizer8.Add( self.ip_text, 1, wx.ALL|wx.ALIGN_CENTER_VERTICAL, 5 )
		
		self.m_staticText3 = wx.StaticText( self, wx.ID_ANY, u" Port：", wx.DefaultPosition, wx.DefaultSize, 0 )
		self.m_staticText3.Wrap( -1 )
		bSizer8.Add( self.m_staticText3, 0, wx.ALIGN_CENTER_VERTICAL|wx.TOP|wx.BOTTOM|wx.LEFT, 5 )
		
		self.port_text = wx.TextCtrl( self, wx.ID_ANY, wx.EmptyString, wx.DefaultPosition, wx.DefaultSize, 0 )
		bSizer8.Add( self.port_text, 0, wx.ALIGN_CENTER_VERTICAL|wx.ALL, 5 )
		
		self.enter_btn = wx.Button( self, wx.ID_ANY, u"启动", wx.DefaultPosition, wx.DefaultSize, 0 )
		bSizer8.Add( self.enter_btn, 0, wx.ALL|wx.ALIGN_CENTER_VERTICAL, 5 )
		
		
		bSizer2.Add( bSizer8, 1, wx.EXPAND, 5 )
		
		
		bSizer1.Add( bSizer2, 1, wx.EXPAND, 5 )
		
		bSizer3 = wx.BoxSizer( wx.VERTICAL )
		
		self.showmessage = wx.TextCtrl( self, wx.ID_ANY, wx.EmptyString, wx.DefaultPosition, wx.DefaultSize, wx.TE_MULTILINE )
		bSizer3.Add( self.showmessage, 1, wx.ALL|wx.EXPAND, 5 )
		
		
		bSizer1.Add( bSizer3, 5, wx.EXPAND, 5 )
		
		
		self.SetSizer( bSizer1 )
		self.Layout()
		
		self.Centre( wx.BOTH )
		
		# Connect Events
		self.enter_btn.Bind( wx.EVT_BUTTON, self.btn_login )
	
	def __del__( self ):
		pass
	
	
	# Virtual event handlers, overide them in your derived class
	def btn_login( self, event ):
		event.Skip()

客戶端（client）

import wx
import socket
import threading
from client import noname


udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)  # 创建套接字


class ClientFrame(noname.MyFrame1):
    def __init__(self, parent):
        noname.MyFrame1.__init__(self, parent)

    def btn_login(self, event):
        ipStr = self.ip_text.GetValue()
        portStr = int(self.port_text.GetValue())
        # 创建udp套接字进行连接
        udp_socket.bind((ipStr, portStr))  # 服务器绑定ip和端口
        self.showmessage.AppendText('----已上线!----\n')  # 显示连接成功

    def btn_send(self, event):
        # while True:
        send_ip = self.send_ip.GetValue()
        send_port = int(self.send_port.GetValue())
        send_address = (send_ip, send_port)  # 发送地址
        send_msg = self.sendmessage.GetValue()  # 获得发送内容
        send_data = send_msg.encode('utf-8')  # 转码
        self.showmessage.AppendText('\n 说:%s' % send_msg)
        # sendto（发送数据，发送地址）
        udp_socket.sendto(send_data, send_address)
        self.sendmessage.SetValue("")  # 清空输入栏
        # 接收信息
        recv_data = udp_socket.recv(1024)
        recv_data = recv_data.decode('utf-8')
        self.showmessage.AppendText('\n 收到信息为:%s' % recv_data)
        print('收到信息为:%s' % recv_data)

    def btn_out(self, event):
        self.Close(True)  # 关闭窗口


def main():
    app = wx.App(False)
    frame = ClientFrame(None)
    frame.Show(True)
    # 启动窗口
    app.MainLoop()


if __name__ == "__main__":
    try:
        main()
    finally:
        udp_socket.close()

# -*- coding: utf-8 -*- 

###########################################################################
## Python code generated with wxFormBuilder (version Jan 23 2018)
## http://www.wxformbuilder.org/
##
## PLEASE DO *NOT* EDIT THIS FILE!
###########################################################################

import wx
import wx.xrc

###########################################################################
## Class MyFrame1
###########################################################################

class MyFrame1 ( wx.Frame ):
	
	def __init__( self, parent ):
		wx.Frame.__init__ ( self, parent, id = wx.ID_ANY, title = wx.EmptyString, pos = wx.DefaultPosition, size = wx.Size( 619,466 ), style = wx.DEFAULT_FRAME_STYLE|wx.TAB_TRAVERSAL )
		
		self.SetSizeHints( wx.DefaultSize, wx.DefaultSize )
		
		bSizer1 = wx.BoxSizer( wx.VERTICAL )
		
		bSizer2 = wx.BoxSizer( wx.VERTICAL )
		
		bSizer8 = wx.BoxSizer( wx.HORIZONTAL )
		
		self.m_staticText1 = wx.StaticText( self, wx.ID_ANY, u"创建我的IP：", wx.DefaultPosition, wx.DefaultSize, 0 )
		self.m_staticText1.Wrap( -1 )
		bSizer8.Add( self.m_staticText1, 0, wx.ALIGN_CENTER_VERTICAL|wx.TOP|wx.BOTTOM|wx.LEFT, 5 )
		
		self.ip_text = wx.TextCtrl( self, wx.ID_ANY, wx.EmptyString, wx.DefaultPosition, wx.DefaultSize, 0 )
		bSizer8.Add( self.ip_text, 1, wx.ALL|wx.ALIGN_CENTER_VERTICAL, 5 )
		
		self.m_staticText3 = wx.StaticText( self, wx.ID_ANY, u" Port：", wx.DefaultPosition, wx.DefaultSize, 0 )
		self.m_staticText3.Wrap( -1 )
		bSizer8.Add( self.m_staticText3, 0, wx.ALIGN_CENTER_VERTICAL|wx.TOP|wx.BOTTOM|wx.LEFT, 5 )
		
		self.port_text = wx.TextCtrl( self, wx.ID_ANY, wx.EmptyString, wx.DefaultPosition, wx.DefaultSize, 0 )
		bSizer8.Add( self.port_text, 0, wx.ALIGN_CENTER_VERTICAL|wx.ALL, 5 )
		
		self.m_staticText7 = wx.StaticText( self, wx.ID_ANY, u"(1024~65535)", wx.DefaultPosition, wx.DefaultSize, 0 )
		self.m_staticText7.Wrap( -1 )
		bSizer8.Add( self.m_staticText7, 0, wx.ALL|wx.ALIGN_CENTER_VERTICAL, 5 )
		
		
		bSizer8.Add( ( 0, 0), 1, wx.EXPAND, 5 )
		
		self.enter_btn = wx.Button( self, wx.ID_ANY, u"进入", wx.DefaultPosition, wx.DefaultSize, 0 )
		bSizer8.Add( self.enter_btn, 0, wx.ALL|wx.ALIGN_CENTER_VERTICAL, 5 )
		
		
		bSizer2.Add( bSizer8, 1, wx.EXPAND, 5 )
		
		
		bSizer1.Add( bSizer2, 1, wx.EXPAND, 5 )
		
		bSizer3 = wx.BoxSizer( wx.VERTICAL )
		
		self.showmessage = wx.TextCtrl( self, wx.ID_ANY, wx.EmptyString, wx.DefaultPosition, wx.DefaultSize, wx.TE_MULTILINE )
		bSizer3.Add( self.showmessage, 1, wx.ALL|wx.EXPAND, 5 )
		
		
		bSizer1.Add( bSizer3, 3, wx.EXPAND, 5 )
		
		bSizer4 = wx.BoxSizer( wx.VERTICAL )
		
		bSizer81 = wx.BoxSizer( wx.HORIZONTAL )
		
		self.m_staticText4 = wx.StaticText( self, wx.ID_ANY, u"接收者的ip：", wx.DefaultPosition, wx.DefaultSize, 0 )
		self.m_staticText4.Wrap( -1 )
		bSizer81.Add( self.m_staticText4, 0, wx.ALL, 5 )
		
		self.send_ip = wx.TextCtrl( self, wx.ID_ANY, wx.EmptyString, wx.DefaultPosition, wx.DefaultSize, 0 )
		bSizer81.Add( self.send_ip, 0, wx.ALL, 5 )
		
		self.m_staticText6 = wx.StaticText( self, wx.ID_ANY, u"Port：", wx.DefaultPosition, wx.DefaultSize, 0 )
		self.m_staticText6.Wrap( -1 )
		bSizer81.Add( self.m_staticText6, 0, wx.ALL, 5 )
		
		self.send_port = wx.TextCtrl( self, wx.ID_ANY, wx.EmptyString, wx.DefaultPosition, wx.DefaultSize, 0 )
		bSizer81.Add( self.send_port, 0, wx.ALL, 5 )
		
		self.m_staticText8 = wx.StaticText( self, wx.ID_ANY, u"(1024~65535)", wx.DefaultPosition, wx.DefaultSize, 0 )
		self.m_staticText8.Wrap( -1 )
		bSizer81.Add( self.m_staticText8, 0, wx.ALL|wx.ALIGN_CENTER_VERTICAL, 5 )
		
		
		bSizer81.Add( ( 0, 0), 1, wx.EXPAND, 5 )
		
		
		bSizer4.Add( bSizer81, 0, wx.EXPAND, 5 )
		
		bSizer5 = wx.BoxSizer( wx.VERTICAL )
		
		self.me = wx.StaticText( self, wx.ID_ANY, u"发送的消息：", wx.DefaultPosition, wx.DefaultSize, 0 )
		self.me.Wrap( -1 )
		bSizer5.Add( self.me, 0, wx.ALL, 5 )
		
		self.sendmessage = wx.TextCtrl( self, wx.ID_ANY, wx.EmptyString, wx.DefaultPosition, wx.DefaultSize, 0 )
		bSizer5.Add( self.sendmessage, 1, wx.ALL|wx.EXPAND, 5 )
		
		
		bSizer4.Add( bSizer5, 2, wx.EXPAND, 5 )
		
		bSizer7 = wx.BoxSizer( wx.HORIZONTAL )
		
		
		bSizer7.Add( ( 0, 0), 1, wx.EXPAND, 5 )
		
		self.send_btn = wx.Button( self, wx.ID_ANY, u"发送", wx.DefaultPosition, wx.DefaultSize, 0 )
		bSizer7.Add( self.send_btn, 0, wx.ALL|wx.ALIGN_RIGHT, 5 )
		
		self.exit_btn = wx.Button( self, wx.ID_ANY, u"退出", wx.DefaultPosition, wx.DefaultSize, 0 )
		bSizer7.Add( self.exit_btn, 0, wx.ALL|wx.ALIGN_RIGHT, 5 )
		
		
		bSizer4.Add( bSizer7, 0, wx.EXPAND|wx.ALIGN_CENTER_VERTICAL, 5 )
		
		
		bSizer1.Add( bSizer4, 3, wx.EXPAND, 5 )
		
		
		self.SetSizer( bSizer1 )
		self.Layout()
		
		self.Centre( wx.BOTH )
		
		# Connect Events
		self.enter_btn.Bind( wx.EVT_BUTTON, self.btn_login )
		self.send_btn.Bind( wx.EVT_BUTTON, self.btn_send )
		self.exit_btn.Bind( wx.EVT_BUTTON, self.btn_out )
	
	def __del__( self ):
		pass
	
	
	# Virtual event handlers, overide them in your derived class
	def btn_login( self, event ):
		event.Skip()
	
	def btn_send( self, event ):
		event.Skip()
	
	def btn_out( self, event ):
		event.Skip()

3. 運行測試

啓動服務器

啓動客戶端，並發送消息

Python基於socket編程（UDP版聊天）

目錄

一、UDP聊天（無界面）

1. 架构

2. 代碼

3. 測試

二、UDP聊天（可视界面）

1. 架构

2. 代碼

服務器（server）

客戶端（client）

3. 運行測試