小猿圈Python游戏之socket实现单机五子棋到双人对战

小猿圈002

前段时间发的飞机大战的游戏很多小伙伴都私聊让再做个游戏，今天小猿圈Python讲师为大家分享的是socket实现单机五子棋到双人对战，想玩的小伙伴记得自己运行一下呦。

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

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

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

    #添加桌面标签

    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[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[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 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函数

  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 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

# 创建 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 socket

import 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坐标开始游戏。

以上就是本文的全部内容，希望对大家的学习有所帮助，希望大家能灵活掌握本文中提到的方法，想要了解更多关于Python开发方面内容的小伙伴可以关注小猿圈官网Python交流群：874680195，如果有什么问题可以留言给小猿圈讲师，遇到的问题可以直接私聊或者提问，看到会尽快帮大家解决的。