快捷導(dǎo)航

基于python的socket實(shí)現(xiàn)單機(jī)五子棋到雙人對戰(zhàn)

更新時間：2020年03月24日 10:35:58 作者：rayrayrayk

這篇文章主要為大家詳細(xì)介紹了基于python的socket實(shí)現(xiàn)單機(jī)五子棋到雙人對戰(zhàn)，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價值，感興趣的小伙伴們可以參考一下

基于python的socket實(shí)現(xiàn)單機(jī)五子棋到雙人對戰(zhàn)，供大家參考，具體內(nèi)容如下

本次實(shí)驗(yàn)使用python語言。通過socket進(jìn)行不同機(jī)器見的通信，具體可以分為以下四步：1.創(chuàng)建ServerSocket和Socket；2.打開鏈接到Socket的輸入/輸出流；3.按照協(xié)議對Socket進(jìn)行讀/寫操作；4.關(guān)閉輸入輸出流、關(guān)閉Socket。

由于是雙人對戰(zhàn)，服務(wù)器必須應(yīng)對多人及以上的客戶端的連接，因此本實(shí)驗(yàn)還引入了python的threading多線程模塊，通過監(jiān)聽實(shí)時監(jiān)控網(wǎng)絡(luò)狀態(tài)，同時利用socket.listen(2)引入排隊(duì)等待機(jī)制。

chess類

#五子棋類
import os
class chessboard(object):
 def __init__(self):
  self.size = 16
  #初始化棋盤
  self.__board=[[' ' for n in range(self.size)] for m in range(self.size)]
  n = 0
  #添加桌面標(biāo)簽
  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):
    #判斷是否結(jié)束
    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):
    #判斷是否結(jié)束
    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):
    #判斷是否結(jié)束
    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):
    #判斷是否結(jié)束
    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):
    #判斷是否結(jié)束
    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):
    # 判斷是否結(jié)束
    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循環(huán)實(shí)現(xiàn)的單機(jī)版五子棋

# -*- coding: utf-8 -*-
#單機(jī)版五子棋
from chess import chessboard
def changeid(id):
 if id==1:
  return 2
 elif id==2:
  return 1
 else:
  return 0
t=chessboard()
id=1#初始化id
t.draw()
while (not t.is_end()):#end函數(shù)
 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)

———————分割線———————

由于要實(shí)現(xiàn)雙人對戰(zhàn)，所以服務(wù)器端必須要用多線程使其服務(wù)多個客戶端，因此使用threading

服務(wù)器端

# -*- coding: utf-8 -*-
#服務(wù)器
import os
import socket
import json
import threading
import time
import sys
from chess import chessboard
t=chessboard()
id=1#初始化id
def 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
# 創(chuàng)建 socket 對象
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 獲取本地主機(jī)名
host = socket.gethostname()
port = 9999
# 綁定端口號
s.bind((host, port))
address=(host, port)
# 設(shè)置最大連接數(shù)，超過后排隊(duì)
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 socket
import time

# 創(chuàng)建 socket 對象
c = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 獲取本地主機(jī)名
#host = socket.gethostname()
host='10.41.114.198'
# 設(shè)置端口號
port = 9999
# 連接服務(wù)，指定主機(jī)和端口
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轉(zhuǎn)成str，再encode編碼

使用方法：先更改客戶端host為自己地址，然后先打開服務(wù)端，然后打開多個客戶端（大于2個開始排隊(duì)），然后開始輸入X,Y坐標(biāo)開始游戲。