基于python的socket实现单机五子棋到双人对战

Python  /  管理员 发布于 7年前   240

基于python的socket实现单机五子棋到双人对战，供大家参考，具体内容如下

本次实验使用python语言。通过socket进行不同机器见的通信，具体可以分为以下四步：1.创建ServerSocket和Socket；2.打开链接到Socket的输入/输出流；3.按照协议对Socket进行读/写操作；4.关闭输入输出流、关闭Socket。

由于是双人对战，服务器必须应对多人及以上的客户端的连接，因此本实验还引入了python的threading多线程模块，通过监听实时监控网络状态，同时利用socket.listen(2)引入排队等待机制。

chess类

#五子棋类import osclass chessboard(object):  def __init__(self):    self.size = 16    #初始化棋盘    self.__board=[[' ' for n in range(self.size)] for m in range(self.size)]    n = 0    #添加桌面标签    while n < self.size:      ntr=str(n)      self.__board[0][n] = ntr.zfill(2)      self.__board[n][0] = ntr.zfill(2)      n=n+1    self.id=0  #胜利条件  def is_end(self):    ch_stack=[]    #行检查    for i in range(self.size):      for j in range(self.size):        #判断是否结束        chess=self.__board[i][j]        if len(ch_stack)==5 and ch_stack[-1]=='* ':          print('winner=id 1')          return 1        elif len(ch_stack) == 5 and ch_stack[-1] == '@ ':          print('winner=id 2')          return 2        if chess==' ':          ch_stack.clear()        else:          if (not ch_stack) or ch_stack[-1] == chess:ch_stack.append(chess)          else:ch_stack.clear()ch_stack.append(chess)      ch_stack.clear()    ch_stack.clear()    #列检查    for j in range(self.size):      for i in range(self.size):        #判断是否结束        if len(ch_stack)==5 and ch_stack[-1]=='* ':          print('winner=id 1')          return 1        elif len(ch_stack) == 5 and ch_stack[-1] == '@ ':          print('winner=id 2')          return 2        chess=self.__board[i][j]        if chess==' ':          ch_stack.clear()        else:          if (not ch_stack) or ch_stack[-1] == chess:ch_stack.append(chess)          else:ch_stack.clear()ch_stack.append(chess)      ch_stack.clear()    ch_stack.clear()    #左斜检查    #下三角    for i in range(self.size):      for j in range(1,self.size):        #判断是否结束        if len(ch_stack)==5 and ch_stack[-1]=='* ':          print('winner=id 1')          return 1        elif len(ch_stack) == 5 and ch_stack[-1] == '@ ':          print('winner=id 2')          return 2        if i+j<self.size:          chess=self.__board[i+j][j]          if chess==' ':ch_stack.clear()          else:if (not ch_stack) or ch_stack[-1] == chess:  ch_stack.append(chess)else:  ch_stack.clear()  ch_stack.append(chess)        else:          break      ch_stack.clear()    ch_stack.clear()    #上三角    for i in range(self.size):      for j in range(1,self.size):        #判断是否结束        if len(ch_stack)==5 and ch_stack[-1]=='* ':          print('winner=id 1')          return 1        elif len(ch_stack) == 5 and ch_stack[-1] == '@ ':          print('winner=id 2')          return 2        if i+j<self.size:          chess=self.__board[j][j+i]          if chess==' ':ch_stack.clear()          else:if (not ch_stack) or ch_stack[-1] == chess:  ch_stack.append(chess)else:  ch_stack.clear()  ch_stack.append(chess)        else:          break      ch_stack.clear()    ch_stack.clear()    #右斜检查    #上三角    for i in range(self.size):      for j in range(1,self.size):        #判断是否结束        if len(ch_stack)==5 and ch_stack[-1]=='* ':          print('winner=id 1')          return 1        elif len(ch_stack) == 5 and ch_stack[-1] == '@ ':          print('winner=id 2')          return 2        if self.size-i-j+1>0:          chess=self.__board[self.size-i-j][j]          if chess==' ':ch_stack.clear()          elif not chess:break          else:if (not ch_stack) or ch_stack[-1] == chess:  ch_stack.append(chess)else:  ch_stack.clear()  ch_stack.append(chess)        else:          break      ch_stack.clear()    ch_stack.clear()    #下三角    for i in range(self.size):      for j in range(1,self.size):        # 判断是否结束        if len(ch_stack) == 5 and ch_stack[-1] == '* ':          print('winner=id 1')          return 1        elif len(ch_stack) == 5 and ch_stack[-1] == '@ ':          print('winner=id 2')          return 2        if self.size-i-j> 0:          chess = self.__board[j][self.size-i-j]          if chess == ' ':ch_stack.clear()          elif not chess:break          else:if (not ch_stack) or ch_stack[-1] == chess:  ch_stack.append(chess)else:  ch_stack.clear()  ch_stack.append(chess)        else:          break      ch_stack.clear()    ch_stack.clear()    return 0  def draw(self):    #clear()    for x in self.__board:      print(x)    return 0  def drop_chess(self,x,y,id):    if id==1 and self.__board[x][y]==' ':      self.__board[x][y]='* '      return 1    elif id==2 and self.__board[x][y]==' ':      self.__board[x][y]='@ '      return 1    else:      return 0

然后是用while循环实现的单机版五子棋

# -*- coding: utf-8 -*-#单机版五子棋from chess import chessboarddef changeid(id):  if id==1:    return 2  elif id==2:    return 1  else:    return 0t=chessboard()id=1#初始化idt.draw()while (not t.is_end()):#end函数  print('your id is %d,input your next drop(x,y)'% id)  x=input()  y=input()  x=int(x)  y=int(y)  if t.drop_chess(x,y,id):    t.draw()  else:    print('_________Illegal Input,Please Check Again_________')    continue  id=changeid(id)

―――――――分割线―――――――

由于要实现双人对战，所以服务器端必须要用多线程使其服务多个客户端，因此使用threading

服务器端

# -*- coding: utf-8 -*-#服务器import osimport socketimport jsonimport threadingimport timeimport sysfrom chess import chessboardt=chessboard()id=1#初始化iddef handle():  while (not t.is_end()):    for c in socks:      global id      json_string0 = json.dumps(t._chessboard__board)      c.sendto(json_string0.encode('utf-8'), address)      msg1 = 'Your id is %d,input your next drop' % id + "\r\n"      c.send(msg1.encode('utf-8'))      msg2x = c.recv(1024)      msg2y = c.recv(1024)      x = int(msg2x.decode('utf-8'))      y = int(msg2y.decode('utf-8'))      print('processing......\n')      if t.drop_chess(x, y, id):        json_string = json.dumps(t._chessboard__board)        c.sendto(json_string.encode('utf-8'), address)      else:        msg3 = '_________Illegal Input,Please Check Again_________'        c.send(msg3.encode('utf-8'))        continue      id = changeid(id)def clear():  os.system('cls')def changeid(id):  if id==1:    return 2  elif id==2:    return 1  else:    return 0# 创建 socket 对象s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)# 获取本地主机名host = socket.gethostname()port = 9999# 绑定端口号s.bind((host, port))address=(host, port)# 设置最大连接数，超过后排队s.listen(2)socks=[]th = threading.Thread(target=handle)th.start()while 1:  c, addr = s.accept()  print  'connected from:', addr  socks.append(c)s.close()

然后是客户端

# -*- coding: utf-8 -*-#客户端import socketimport time# 创建 socket 对象c = socket.socket(socket.AF_INET, socket.SOCK_STREAM)# 获取本地主机名#host = socket.gethostname()host='10.41.114.198'# 设置端口号port = 9999# 连接服务，指定主机和端口c.connect((host, port))address=(host, port)while 1:  #s=c.accept()  print('')  print('__________________wait__________________')  msg0 = c.recv(2048).decode('utf-8') # 棋盘大于1024  for x in msg0:    if x == '[':      print('')    else:      print(x, end='')  print('')  msg1 = c.recv(1024)#接收输入提示  print (msg1.decode('utf-8'))  time.sleep(1)  x = input('x=')  y = input('y=')  c.send(x.encode('utf-8'))  c.send(y.encode('utf-8'))  msg3 = c.recv(2048).decode('utf-8')#棋盘大于1024  if msg3=='_________Illegal Input,Please Check Again_________':    print(msg3)    continue  else:    #print(msg3)    for x in msg3:      if x=='[':        print('')      else:        print(x, end='')    print('')    print('__________________wait__________________')  print('')c.close()

注意socket传输时只能传送bytes，因此list先用json转成str，再encode编码

使用方法：先更改客户端host为自己地址，然后先打开服务端，然后打开多个客户端（大于2个开始排队），然后开始输入X,Y坐标开始游戏。

由于时间紧急，暂时未处理单个客户端退出后的程序初始化问题，可能过几日会补上。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。