本文實(shí)例為大家分享了python實(shí)現(xiàn)簡(jiǎn)單聊天室的具體代碼，供大家參考，具體內(nèi)容如下

剛剛接觸python編程，又從接觸java開(kāi)始一直對(duì)socket模塊感興趣，所以就做了一個(gè)聊天室的小程序。

該程序由客戶(hù)端與服務(wù)器構(gòu)成，使用UDP服務(wù)，服務(wù)器端綁定本地IP和端口，客戶(hù)端由系統(tǒng)隨機(jī)選擇端口。

實(shí)現(xiàn)了群發(fā)、私發(fā)、點(diǎn)對(duì)點(diǎn)文件互傳功能。

客戶(hù)端自建了一個(gè)類(lèi)繼承了Cmd模塊，使用自定義的命令command進(jìn)行操作，調(diào)用相應(yīng)的do_command方法。

使用json模塊進(jìn)行消息的封裝序列化，在接收方進(jìn)行解析。

客戶(hù)端代碼如下：

import socketimport threadingimport jsonimport osfrom cmd import Cmd  class Client(Cmd):  """  客戶(hù)端  """  prompt = '>>>'  intro = '[Welcome] 簡(jiǎn)易聊天室客戶(hù)端(Cli版)/n' + '[Welcome] 輸入help來(lái)獲取幫助/n'  buffersize = 1024   def __init__(self, host):    """    構(gòu)造    """    super().__init__()    self.__socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)    # self.__id = None    self.__nickname = None    self.__host = host    self.thread_recv = None    self.threadisalive = False    # 是否在接收文件    self.recvfile = False    # 是否在發(fā)送文件    self.sendfile = False    self.filesize = None    self.sendfilesize = None     # 接收文件包計(jì)數(shù)    self.filecount = None    # 接收文件名    self.filename = None    # 發(fā)送文件名    self.sendfilename = None     # 發(fā)送者    self.filefrom = None    # 接收者    self.fileto = None     # 接收文件流    self.file_recv = None    # 發(fā)送文件流    self.file_send = None     # 接收文件地址    self.filefrom_addr = None    # 發(fā)送文件地址    self.fileto_addr = None   def __receive_message_thread(self):    """    接受消息線程    """    while self.threadisalive:      # noinspection PyBroadException      try:        buffer, addr = self.__socket.recvfrom(1024)        '''        文件流由發(fā)送端直接發(fā)送，不經(jīng)過(guò)服務(wù)器，故當(dāng)發(fā)送端發(fā)來(lái)的消息時(shí)，將收到的數(shù)據(jù)存入文件        '''        if (addr != self.__host) & (addr == self.filefrom_addr) & self.recvfile:          self.file_recv.write(buffer)          self.filecount += 1          if self.filecount * 1024 >= self.filesize:            self.file_recv.close()            print(self.filename, 'is received.')            self.recvfile = False          continue         js = json.loads(buffer.decode())         # 若接收的數(shù)據(jù)為消息信息，則顯示        if js['type'] == 'message':          print(js['message'])         # 若接收的數(shù)據(jù)為文件發(fā)送請(qǐng)求，則存儲(chǔ)文件信息，并顯示        elif js['type'] == 'filequest':          if self.recvfile:            self.__socket.sendto(json.dumps({              'type': 'fileres',              'fileres': 'no',              'nickname': self.__nickname,              'who': js['nickname'],              'errormessage': 'is transfroming files.',            }).encode(), self.__host)            continue          filename = js['filename']          who = js['nickname']          filesize = js['filesize']          self.recvfile = True          self.filesize = filesize          self.filename = filename          self.filecount = 0          self.filefrom = who          self.filefrom_addr = (js['send_ip'], js['send_port'])           print('[system]:', who, ' send a file(',             filename, ') to you. receive? ')         # 接受的數(shù)據(jù)為請(qǐng)求回復(fù)，若同意接收則存儲(chǔ)服務(wù)器發(fā)來(lái)的接收方的地址，并開(kāi)啟發(fā)送線程        elif js['type'] == 'fileres':          if js['fileres'] == 'yes':            print(js['recv_ip'], js['recv_port'])            self.fileto_addr = (js['recv_ip'], js['recv_port'])            thread = threading.Thread(              target=self.__send_file_thread)            thread.start()          else:            print(js['nickname'], js['errormessage'])            self.sendfile = False       except Exception as e:        print(e)        print('[Client] 無(wú)法從服務(wù)器獲取數(shù)據(jù)')   def __send_broadcast_message_thread(self, message):    """    發(fā)送廣播消息線程    :param message: 消息內(nèi)容    """    self.__socket.sendto(json.dumps({      'type': 'broadcast',      'nickname': self.__nickname,      'message': message,    }).encode(), self.__host)   def __send_file_thread(self):    """    發(fā)送文件線程    :param message: 消息內(nèi)容    """    filecount = 0    print('[system]', 'sending the file...')    while filecount * 1024 <= self.sendfilesize:      self.__socket.sendto(        self.file_send.read(1024), self.fileto_addr)      filecount += 1    self.file_send.close()    self.sendfile = False    print('[system]', 'the file is sended.')   def __send_whisper_message_thread(self, who, message):    """    發(fā)送私發(fā)消息線程    :param message: 消息內(nèi)容    """    self.__socket.sendto(json.dumps({      'type': 'sendto',      'who': who,      'nickname': self.__nickname,      'message': message    }).encode(), self.__host)   def send_exit(self):    self.__socket.sendto(json.dumps({      'type': 'offline',      'nickname': self.__nickname,    }).encode(), self.__host)    def start(self):    """    啟動(dòng)客戶(hù)端    """    self.cmdloop()   def do_login(self, args):    """    登錄聊天室    :param args: 參數(shù)    """    nickname = args.split(' ')[0]     # 將昵稱(chēng)發(fā)送給服務(wù)器，獲取用戶(hù)id    self.__socket.sendto(json.dumps({      'type': 'login',      'nickname': nickname,    }).encode(), self.__host)    # 嘗試接受數(shù)據(jù)     buffer = self.__socket.recvfrom(1300)[0].decode()    obj = json.loads(buffer)    if obj['login'] == 'success':      self.__nickname = nickname      print('[Client] 成功登錄到聊天室')      self.threadisalive = True      # 開(kāi)啟子線程用于接受數(shù)據(jù)      self.thread_recv = threading.Thread(        target=self.__receive_message_thread)      self.thread_recv.setDaemon(True)      self.thread_recv.start()    else:      print('[Client] 無(wú)法登錄到聊天室', obj['errormessage'])   def do_send(self, args):    """    發(fā)送消息    :param args: 參數(shù)    """    if self.__nickname is None:      print('請(qǐng)先登錄！login nickname')      return    message = args    # 開(kāi)啟子線程用于發(fā)送數(shù)據(jù)    thread = threading.Thread(      target=self.__send_broadcast_message_thread, args=(message, ))    thread.setDaemon(True)    thread.start()   def do_sendto(self, args):    """    發(fā)送私發(fā)消息    :param args: 參數(shù)    """    if self.__nickname is None:      print('請(qǐng)先登錄！login nickname')      return    who = args.split(' ')[0]    message = args.split(' ')[1]    # # 顯示自己發(fā)送的消息    # print('[' + str(self.__nickname) + '(' + str(self.__id) + ')' + ']', message)    # 開(kāi)啟子線程用于發(fā)送數(shù)據(jù)    thread = threading.Thread(      target=self.__send_whisper_message_thread, args=(who, message))    thread.setDaemon(True)    thread.start()   def do_catusers(self, arg):    if self.__nickname is None:      print('請(qǐng)先登錄！login nickname')      return    catmessage = json.dumps({'type': 'catusers'})    self.__socket.sendto(catmessage.encode(), self.__host)   def do_catip(self, args):    if self.__nickname is None:      print('請(qǐng)先登錄！login nickname')      return    who = args    catipmessage = json.dumps({'type': 'catip', 'who': who})    self.__socket.sendto(catipmessage.encode(), self.__host)   def do_help(self, arg):    """    幫助    :param arg: 參數(shù)    """    command = arg.split(' ')[0]    if command == '':      print('[Help] login nickname - 登錄到聊天室，nickname是你選擇的昵稱(chēng)')      print('[Help] send message - 發(fā)送消息，message是你輸入的消息')      print('[Help] sendto who message - 私發(fā)消息，who是用戶(hù)名，message是你輸入的消息')      print('[Help] catusers - 查看所有用戶(hù)')      print('[Help] catip who - 查看用戶(hù)IP，who為用戶(hù)名')      print('[Help] sendfile who filedir - 向某用戶(hù)發(fā)送文件，who為用戶(hù)名，filedir為文件路徑')      print('[Help] getfile filename who yes/no - 接收文件，filename 為文件名,who為發(fā)送者，yes/no為是否接收')    elif command == 'login':      print('[Help] login nickname - 登錄到聊天室，nickname是你選擇的昵稱(chēng)')    elif command == 'send':      print('[Help] send message - 發(fā)送消息，message是你輸入的消息')    elif command == 'sendto':      print('[Help] sendto who message - 發(fā)送私發(fā)消息，message是你輸入的消息')    else:      print('[Help] 沒(méi)有查詢(xún)到你想要了解的指令')   def do_exit(self, arg): # 以do_*開(kāi)頭為命令    print("Exit")    self.send_exit()    try:      self.threadisalive = False      self.thread_recv.join()    except Exception as e:      print(e)    # self.__socket.close()   def do_sendfile(self, args):    who = args.split(' ')[0]    filepath = args.split(' ')[1]    filename = filepath.split('//')[-1]    # 判斷是否在發(fā)送文件    if self.sendfile:      print('you are sending files, please try later.')      return    if not os.path.exists(filepath):      print('the file is not exist.')      return    filesize = os.path.getsize(filepath)    # print(who, filename, filesize)     self.sendfile = True    self.fileto = who    self.sendfilename = filename    self.sendfilesize = filesize    self.file_send = open(filepath, 'rb')     self.__socket.sendto(json.dumps({      'type': 'filequest',      'nickname': self.__nickname,      'filename': self.sendfilename,      'filesize': self.sendfilesize,      'who': self.fileto,      'send_ip': '',      'send_port': '',    }).encode(), self.__host)     print('request send...')     # fileres = self.__socket.recvfrom(1024)[0].decode()    # js = json.loads(fileres)   def do_getfile(self, args):    filename = args.split(' ')[0]    who = args.split(' ')[1]    ch = args.split(' ')[2]    # print(self.filename is not None, filename, self.filename, who, self.filefrom)    if (self.filename is not None) & (filename == self.filename) & (who == self.filefrom):      if ch == 'yes':        self.file_recv = open(self.filename, 'wb')        self.__socket.sendto(json.dumps({          'type': 'fileres',          'fileres': 'yes',          'nickname': self.__nickname,          'who': who,          'recv_ip': '',          'recv_port': '',        }).encode(), self.__host)        print('you agree to reveive the file(', filename, ') from', who)       else:        self.__socket.sendto(json.dumps({          'type': 'fileres',          'fileres': 'no',          'nickname': self.__nickname,          'errormessage': 'deny the file.',          'who': who,          'recv_ip': '',          'recv_port': '',        }).encode(), self.__host)        print('you deny to reveive the file(', filename, ') from', who)        self.recvfile = False    else:      print('the name or sender of the file is wrong.')  c = Client(('127.0.0.1', 12346))c.start()