本文最后更新于 864 天前，其中的信息可能已经有所发展或是发生改变。
Python选修课的课程设计为网络五子棋，此案例运行环境为Python 3.9.9，使用VSCode编写。
客户端大部分代码是老师给的(老师也是从网上扒的)
1. 案例展示
2. 服务器端
2.1 简介
服务器端的主要作用就是进行端口转发，即与两个客户端分别建立tcp连接，然后将两个客户端发送来的数据分别发给另一个客户端，两个没有静态公网ip的客户端就可以通过有静态公网ip的服务器端进行对战。
2.2 函数解释
2.2.1 启动服务
这段主要目的是在31500端口（可自行指定）开启tcp监听，与两个客户端建立连接之后即可开始转发数据。还有一个将连接数限制在2个的功能，若连接数大于2则断开额外的连接并告知服务器已满。
def start():
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
localaddr = ("localhost",31500) #填写本机信息，运行在服务器上时需填写内网ip
server.bind(localaddr)
server.listen(5) #开启端口监听
print('服务器启动成功')
while True:
ss, addr = server.accept()
tip1 = 'success|' #连接成功告知
tip2 = 'full|' #已满告知
if len(portlist) < 2:
ss.send(tip1.encode())
elif len(portlist) >= 2:
ss.send(tip2.encode())
ss.close() #若已建立连接数为2则断开新建立的连接并告知已满
return
print(str(addr) + "已连接！")
t = threading.Thread(target=run, args=(ss, addr)) #每连接一个客户端就开启一个线程
t.start()2.2.2 进行转发
这段的主要目的是获取一个客户端发送的数据并将其转发给另一个客户端。区分客户端的方法是建立一个字典users{}，因为tcp连接并不会建立在监听端口，而是会分配一个新的端口作为通道，所以端口号是唯一的，因此键为端口号，值为对应的套接字。
def run(ss, addr):
port = addr[1]
portlist.append(port) #将端口号加入portlist
users[port] = ss #存储端口号对应的套接字
while True:
rData = ss.recv(1024) #接受客户端发送的信息
dataStr = rData.decode("utf-8")
ifExit(dataStr, ss) #判断游戏是否已经结束
print('源地址：{0}, 消息：{1}'.format(addr, dataStr))
for i in portlist: #将信息转发给另一个客户端
if port != i:
users[i].send(rData)2.2.3 断开连接
当有玩家退出时清空列表portlist，以便新玩家加入。
def ifExit(text, ss): #断开连接并清空portlist
global portlist
status = text.split('|')
if status[0] == 'exit':
print('玩家已断开连接')
for i in portlist: #断开已存入users的所有连接
users[i].close()
portlist = [] #清空portlist2.3 服务器端完整代码
# -- coding:UTF-8 --
import socket
import threading
def ifExit(text, ss): #断开连接并清空portlist
global portlist
status = text.split('|')
if status[0] == 'exit':
print('玩家已断开连接')
for i in portlist: #断开已存入users的所有连接
users[i].close()
portlist = [] #清空portlist
def run(ss, addr):
port = addr[1]
portlist.append(port) #将端口号加入portlist
users[port] = ss #存储端口号对应的套接字
while True:
rData = ss.recv(1024) #接受客户端发送的信息
dataStr = rData.decode("utf-8")
ifExit(dataStr, ss) #判断游戏是否已经结束
print('源地址：{0}, 消息：{1}'.format(addr, dataStr))
for i in portlist: #将信息转发给另一个客户端
if port != i:
users[i].send(rData)
def start():
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
localaddr = ("10.0.8.7",31500) #填写本机信息
server.bind(localaddr)
server.listen(5) #开启端口监听
print('服务器启动成功')
while True:
ss, addr = server.accept()
tip1 = 'success|' #连接成功告知
tip2 = 'full|' #已满告知
if len(portlist) < 2:
ss.send(tip1.encode())
elif len(portlist) >= 2:
ss.send(tip2.encode())
ss.close() #若已建立连接数为2则断开新建立的连接并告知已满
return
print(str(addr) + "已连接！")
t = threading.Thread(target=run, args=(ss, addr)) #每连接一个客户端就开启一个线程
t.start()
def startSever():
s = threading.Thread(target=start) #启用一个线程开启服务器
s.start()
users = {} #用户字典，存储{端口号：套接字}
portlist = [] #存储已连接设备的端口号
startSever()3. 客户端
3.1 简介
客户端使用tk构建图形化界面，有关tk编程可参考此链接：tkinter —— Tcl/Tk 的 Python 接口 — Python 3.10.1 文档
3.2 函数解释
3.2.1 图形界面
首先创建一个主界面
window = Tk() #创建一个窗口
window.title('网络五子棋') #设置窗口标题
cv = Canvas(window, bg = 'green', width = 610, height = 610) #创建背景画布
cv.create_image(305, 305, image = bgds[0]) #打印背景图片（bgd[]为存储了图片地址的列表）接下来打印棋盘线
def drawChessboard(): #打印棋盘
for i in range(0, 15):
cv.create_line(20, 20 + 40 * i, 580, 20 + 40 * i,width = 2)
for i in range(0, 15):
cv.create_line(20 + 40 * i, 20, 20 + 40 * i, 580, width = 2)
cv.pack() #使用pack方法创建组件现在创建各种按钮
#创建gui组件
cv.bind('<Button-1>', callpos) #将按下鼠标左键绑定至callpos函数（报告落子坐标）
cv.pack()
label1 = Label(window,text = '五子棋')
label1.pack()
f = Frame(window) #将下列四个按钮打包成一个框架
button0 = Button(f,text="连接服务器")
button0.bind("<Button-1>", calljoin)
button0.pack(side='left')
button2 = Button(f, text = ' 悔棋 ')
button2.bind("<Button-1>", callWithdraw)
button2.pack(side='left')
button1 = Button(f, text = ' 退出游戏 ')
button1.bind('<Button-1>', callexit)
button1.pack(side='left')
button1 = Button(f, text = ' 切换背景 ')
button1.bind('<Button-1>', changeBackground)
button1.pack(side='left')
f.pack()
一个主界面就这样创建完毕了。
但是现在只能看，什么都干不了，接下来要写的是运行需要的功能代码。
3.2.2 走棋功能
棋盘信息
首先创建一个二维列表，用来存放棋子信息，也就是棋盘上哪里有棋子，哪里没有棋子，这些信息都存在二维列表map[ ][ ]里面。没有棋子为空格‘ ’，有黑色棋子为‘0’，有白色棋子为‘1’。
map = [[' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ']for i in range(15)]
回合判断
接下来定义两个变量，turn意为当前回合，Myturn在第一次赋值为0或1后就不会再改变，用于判断当前回合属于哪一方。
turn = 0 #黑方为0，白方为1
Myturn = -1落子
当第一个落子的客户端令turn = 0，即黑方，后落子的客户端为白方，当turn == Myturn时说明此回合走棋。
若鼠标点击处map的值为‘ ’则可以落子，并向服务器发送一个报文报告落子坐标，报文结构为move| x，y；当另一个客户端收到后会将报文以分隔符 | 分隔，| 左边为执行动作，右边为其他信息，在这里右边为落子坐标。
每次落子后使用judge函数判断游戏是否结束，若已结束则发送报文报告胜负信息，若未结束则换对手走棋。
def callpos(event): #落子
global turn
global Myturn, myChess
global map
global xtmp, ytmp
if Myturn == -1: #确认自己角色
Myturn = turn
else:
if(Myturn != turn):
showinfo(title = '提示', message = '还没轮到您走棋')
return
x = (event.x)//40 #将鼠标点击坐标转换到棋盘线交点上
y = (event.y)//40
xtmp = x
ytmp = y
if map[x][y] != ' ': #判断落子处是否已有棋子
pass
else:
img1 = imgs[turn] #创建棋子
myChess = cv.create_image((x * 40 + 20, y * 40 + 20), image = img1)
cv.pack()
map[x][y] = str(turn)
pos = str(x) + ',' + str(y) #编辑落子坐标报文
sendMessage('move|' + pos)
label1['text'] = '对手落子坐标: ' + pos
if judge() == True: #判断输赢
if turn == 0:
showinfo(title = '提示', message = '黑方获胜')
sendMessage('over|黑方获胜')
else:
showinfo(title = '提示', message = '白方获胜')
sendMessage('over|白方获胜')
#若游戏未结束则换方落子
if turn == 0:
turn = 1
else:
turn = 0悔棋
设置一个全局变量withdrawFlag判断是否已悔过棋。稍微修改一下可以更改悔棋次数。
如果未悔过棋且不在自己回合就删除自己刚下的棋子并转到自己回合，同时向服务器发送报文报告悔棋，我这里写的是要在对手落子之前才能悔棋，还不够完善（懒得改了）。
def callWithdraw(event): #悔棋(仅限一次)
global withdrawFlag
global turn, Myturn
global myChess
global xtmp, ytmp
if withdrawFlag == 0 and turn != Myturn: #未悔过棋且不在自己回合
pos = 'withdraw|{0}|{1}'.format(xtmp, ytmp)
sendMessage(pos)
cv.delete(myChess) #删除自己棋子
map[xtmp][ytmp] = ' '
withdrawFlag = 1
turn = Myturn #切换为自己回合
elif withdrawFlag == 1 and turn != Myturn:
showinfo(title = '提示', message = '悔棋次数已用完！')
else:
showinfo(title = '提示', message = '无法在您的回合悔棋！')结束判断
判断棋盘上四个轴有没有五颗棋子连在一起的，如果有就返回True，否则返回False。
原理很简单，就是把列表map的每一维都遍历一次。
def judge(): #判断输赢
#扫描整个棋盘，判断是否连成五颗
a = str(turn)
for i in range(0,11): # 判断X = Y轴
for j in range(0,11):
if map[i][j] == a and map[i + 1][j + 1] == a and map[i + 2][j + 2] == a and map[i + 3][j + 3] == a and map[i + 4][j + 4] == a :
print("X= Y轴上形成五子连珠")
return True
for i in range(4,15): # 判断X = -Y轴
for j in range(0,11):
if map[i][j] == a and map[i - 1][j + 1] == a and map[i - 2][j + 2] == a and map[i - 3][j + 3] == a and map[i - 4][j + 4] == a :
print("X= -Y轴上形成五子连珠")
return True
for i in range(0,15): # 判断Y轴
for j in range(4,15):
if map[i][j] == a and map[i][j - 1] == a and map[i][j - 2] == a and map[i][j - 3] == a and map[i][j - 4] == a :
print("Y轴上形成五子连珠")
return True
for i in range(0,11): # 判断X轴
for j in range(0,15):#0--14
if map[i][j] == a and map[i + 1][j] == a and map[i + 2][j] == a and map[i + 3][j] == a and map[i + 4][j] == a :
print("X轴上形成五子连珠")
return True
return False3.2.3 网络功能
建立连接
用网络发送报文，首先得建立连接。有关socket编程可参考：socket --- 底层网络接口 — Python 3.10.1 文档
#建立连接
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
addr = ('localhost', 31500) #要连接的服务器ip向服务器发送连接请求，服务器收到之后会返回连接成功或服务器已满。发送后创建新进程处理收到的报文。
def calljoin(event): #连接服务器
s.connect(addr)
pos = 'join|'
sendMessage(pos)
startNewThread() #创建新进程处理收到的报文发送报文
使用send方法发送报文pos。
def sendMessage(pos): #发送报文
global s
s.send(pos.encode())处理报文
报文结构总体为“功能 | 信息”，客户端将收到的报文解码并使用split方法以’|‘为分隔符分离出功能和信息，根据不同功能做出不同选择，并将信息显示在程序窗口和弹窗提示。
当收到的是悔棋报文时会删除刚刚生成的对方棋子，并将回合转为对方回合，同时将对方悔棋坐标对应的map值设为‘ ’。
def receiveMessage(): #处理收到的报文
global s, turn
global oppoChess
global map
while True:
data, addr = s.recvfrom(1024)
data = data.decode('utf-8')
status = data.split('|')
if not data:
print('客户端已断开连接')
break
elif status[0] == 'join': #加入信息
print('玩家已加入')
label1['text'] = '玩家已加入，游戏开始'
elif status[0] == 'exit': #断开信息
print('玩家已退出')
label1['text'] = '玩家已退出，游戏结束'
elif status[0] == 'over': #结束信息
print('对方获胜')
label1['text'] = data.split('|')[0]
showinfo(title = '提示', message = data.split('|')[1])
elif status[0] == 'move': #落子信息
print('落子信息：', data)
coordinate = status[1].split(',')
x = int(coordinate[0])
y = int(coordinate[1])
print(coordinate[0], coordinate[1])
label1['text'] = '对手落子坐标：' + coordinate[0] + ',' + coordinate[1]
drawOpponentChess(x, y)
elif status[0] == 'success':#连接成功
print('连接成功')
label1['text'] = data.split('|')[0]
showinfo(title = '提示', message = '服务器连接成功!')
elif status[0] == 'full': #服务器已满
print('服务器已满')
label1['text'] = data.split('|')[0]
showinfo(title = '提示', message = '服务器已满!')
elif status[0] == 'withdraw': #悔棋
cv.delete(oppoChess) #删除对方棋子
showinfo(title = '提示', message = '对方已悔棋！')
label1['text'] = data.split('|')[0]
turn = 1 - Myturn #切换回合
map[int(data.split('|')[1])][int(data.split('|')[2])] = ' '
s.close()3.2.4 其他功能
切换背景
就是创建一个列表存放各个图片的路径，然后设置点击切换背景按钮时循环遍历图片列表即可。
那自然要整点二次元了。
def changeBackground(event): #更换背景图片
global bgdIndex
cv.create_image(305, 305, image = bgds[bgdIndex])
bgdIndex = (bgdIndex + 1) % len(bgds) #在背景图片列表里循环选择
x = y = 0
for x in range(0, 15): #重新绘制棋子
for y in range(0, 15):
if map[x][y] == '0':
cv.create_image((x * 40+20, y * 40 + 20), image = imgs[0])
elif map[x][y] == '1':
cv.create_image((x * 40+20, y * 40 + 20), image = imgs[1])
3.3 客户端完整代码
import threading
import os
import socket
from tkinter import *
from tkinter.messagebox import *
window = Tk()
window.title('网络五子棋')
imgs = [PhotoImage(file = 'bmp\\blackstone.gif'), PhotoImage(file = 'bmp\\whitestone.gif')]
bgds = [PhotoImage(file = 'bmp\\bgd.gif'), PhotoImage(file = 'bmp\\bgd2.gif'), PhotoImage(file = 'bmp\\bgd3.gif'), PhotoImage(file = 'bmp\\bgd4.gif'), PhotoImage(file = 'bmp\\bgd5.gif')]
turn = 0 #黑方为0，白方为1
Myturn = -1
cv = Canvas(window, bg = 'green', width = 610, height = 610) #创建画布
cv.create_image(305, 305, image = bgds[0])
map = [[' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ']for i in range(15)]
withdrawFlag = 0
bgdIndex = 1
def callexit(event): #断开连接
pos = 'exit|'
sendMessage(pos)
os._exit(0)
def calljoin(event): #连接服务器
s.connect(addr)
pos = 'join|'
sendMessage(pos)
startNewThread()
def callpos(event): #落子
global turn
global Myturn, myChess
global map
global xtmp, ytmp
if Myturn == -1: #确认自己角色
Myturn = turn
else:
if(Myturn != turn):
showinfo(title = '提示', message = '还没轮到您走棋')
return
x = (event.x)//40
y = (event.y)//40
xtmp = x
ytmp = y
if map[x][y] != ' ': #判断落子处是否已有棋子
pass
else:
img1 = imgs[turn] #创建棋子
myChess = cv.create_image((x * 40 + 20, y * 40 + 20), image = img1)
cv.pack()
map[x][y] = str(turn)
pos = str(x) + ',' + str(y) #编辑落子坐标报文
sendMessage('move|' + pos)
label1['text'] = '对手落子坐标: ' + pos
if judge() == True: #判断输赢
if turn == 0:
showinfo(title = '提示', message = '黑方获胜')
sendMessage('over|黑方获胜')
else:
showinfo(title = '提示', message = '白方获胜')
sendMessage('over|白方获胜')
#若游戏未结束则换方落子
if turn == 0:
turn = 1
else:
turn = 0
def callWithdraw(event): #悔棋(仅限一次)
global withdrawFlag
global turn, Myturn
global myChess
global xtmp, ytmp
if withdrawFlag == 0 and turn != Myturn: #未悔过棋且不在自己回合
pos = 'withdraw|{0}|{1}'.format(xtmp, ytmp)
sendMessage(pos)
cv.delete(myChess) #删除自己棋子
map[xtmp][ytmp] = ' '
withdrawFlag = 1
turn = Myturn #切换为自己回合
elif withdrawFlag == 1 and turn != Myturn:
showinfo(title = '提示', message = '悔棋次数已用完！')
else:
showinfo(title = '提示', message = '无法在您的回合悔棋！')
def changeBackground(event): #更换背景图片
global bgdIndex
cv.create_image(305, 305, image = bgds[bgdIndex])
bgdIndex = (bgdIndex + 1) % len(bgds) #在背景图片列表里循环选择
x = y = 0
for x in range(0, 15): #重新绘制棋子
for y in range(0, 15):
if map[x][y] == '0':
cv.create_image((x * 40+20, y * 40 + 20), image = imgs[0])
elif map[x][y] == '1':
cv.create_image((x * 40+20, y * 40 + 20), image = imgs[1])
def drawOpponentChess(x, y): #打印对手棋子
global turn
global oppoChess
img1 = imgs[turn]
oppoChess = cv.create_image((x * 40 + 20, y * 40 + 20), image = img1)
cv.pack()
map[x][y] = str(turn)
#换方落子
if turn == 0:
turn = 1
else:
turn = 0
def drawChessboard(): #打印棋盘
for i in range(0, 15):
cv.create_line(20, 20 + 40 * i, 580, 20 + 40 * i,width = 2)
for i in range(0, 15):
cv.create_line(20 + 40 * i, 20, 20 + 40 * i, 580, width = 2)
cv.pack()
def judge(): #判断输赢
#扫描整个棋盘，判断是否连成五颗
a = str(turn)
for i in range(0,11): # 判断X = Y轴
for j in range(0,11):
if map[i][j] == a and map[i + 1][j + 1] == a and map[i + 2][j + 2] == a and map[i + 3][j + 3] == a and map[i + 4][j + 4] == a :
print("X= Y轴上形成五子连珠")
return True
for i in range(4,15): # 判断X = -Y轴
for j in range(0,11):
if map[i][j] == a and map[i - 1][j + 1] == a and map[i - 2][j + 2] == a and map[i - 3][j + 3] == a and map[i - 4][j + 4] == a :
print("X= -Y轴上形成五子连珠")
return True
for i in range(0,15): # 判断Y轴
for j in range(4,15):
if map[i][j] == a and map[i][j - 1] == a and map[i][j - 2] == a and map[i][j - 3] == a and map[i][j - 4] == a :
print("Y轴上形成五子连珠")
return True
for i in range(0,11): # 判断X轴
for j in range(0,15):#0--14
if map[i][j] == a and map[i + 1][j] == a and map[i + 2][j] == a and map[i + 3][j] == a and map[i + 4][j] == a :
print("X轴上形成五子连珠")
return True
return False
def printMap(): #打印map地图
for j in range(0, 15):
for i in range(0, 15):
print(map[i][j], end=' ')
print('w')
def receiveMessage(): #处理收到的报文
global s, turn
global oppoChess
global map
while True:
data, addr = s.recvfrom(1024)
data = data.decode('utf-8')
status = data.split('|')
if not data:
print('客户端已断开连接')
break
elif status[0] == 'join': #加入信息
print('玩家已加入')
label1['text'] = '玩家已加入，游戏开始'
elif status[0] == 'exit': #断开信息
print('玩家已退出')
label1['text'] = '玩家已退出，游戏结束'
elif status[0] == 'over': #结束信息
print('对方获胜')
label1['text'] = data.split('|')[0]
showinfo(title = '提示', message = data.split('|')[1])
elif status[0] == 'move': #落子信息
print('落子信息：', data)
coordinate = status[1].split(',')
x = int(coordinate[0])
y = int(coordinate[1])
print(coordinate[0], coordinate[1])
label1['text'] = '对手落子坐标：' + coordinate[0] + ',' + coordinate[1]
drawOpponentChess(x, y)
elif status[0] == 'success':#连接成功
print('连接成功')
label1['text'] = data.split('|')[0]
showinfo(title = '提示', message = '服务器连接成功!')
elif status[0] == 'full': #服务器已满
print('服务器已满')
label1['text'] = data.split('|')[0]
showinfo(title = '提示', message = '服务器已满!')
elif status[0] == 'withdraw': #悔棋
cv.delete(oppoChess) #删除对方棋子
showinfo(title = '提示', message = '对方已悔棋！')
label1['text'] = data.split('|')[0]
turn = 1 - Myturn #切换回合
map[int(data.split('|')[1])][int(data.split('|')[2])] = ' '
s.close()
def sendMessage(pos): #发送报文
global s
s.send(pos.encode())
def startNewThread(): #启动新线程接收客户端消息
thread = threading.Thread(target = receiveMessage, args=())
thread.setDaemon(True)
thread.start()
drawChessboard()
#创建gui组件
cv.bind('<Button-1>', callpos)
cv.pack()
label1 = Label(window,text = '五子棋')
label1.pack()
f = Frame(window)
button0 = Button(f,text="连接服务器")
button0.bind("<Button-1>", calljoin)
button0.pack(side='left')
button2 = Button(f, text = ' 悔棋 ')
button2.bind("<Button-1>", callWithdraw)
button2.pack(side='left')
button1 = Button(f, text = ' 退出游戏 ')
button1.bind('<Button-1>', callexit)
button1.pack(side='left')
button1 = Button(f, text = ' 切换背景 ')
button1.bind('<Button-1>', changeBackground)
button1.pack(side='left')
f.pack()
#建立连接
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
addr = ('localhost', 31500) #服务器ip，端口可自行选择
window.mainloop()
