且构网

分享程序员开发的那些事...
且构网 - 分享程序员编程开发的那些事

通过python进行多人游戏

更新时间:2023-02-02 18:11:17

这是一个非常模糊的问题,但我可以给你一些模糊的答案.

This is a very vague question, but I can give you some vague answers.

首先,您需要设计一个简单的协议.一个非常简单的基于行的协议应该可以正常工作:UTF-8 文本、换行符分隔消息、空格分隔参数.例如,您可以有这些客户端->服务器消息:

First, you need to design a simple protocol. A very simple line-based protocol should work fine: UTF-8 text, newlines separating messages, spaces separating parameters. For example, you could have these client->server messages:

JOIN name
SAY message with spaces
FOLD
RAISE amount
# ...

...以及这些服务器->客户端消息:

… and these server->client messages:

OK
ERROR error message
JOINED player name with spaces
LEFT player
SAID player message with spaces
NEWHAND player player player player…
DEALT player face suit
ANTED player amount
CHECKED player
# ...

这样的协议的好处在于,您可以使用 telnetnc 手动键入它,因此您甚至不需要客户端进行测试.

The nice thing about a protocol like this is that you can type it manually with telnet or nc, so you don't even need a client for testing.

现在您需要构建一个实现该协议的服务器,并将游戏逻辑构建到服务器中.

Now you need to build a server that implements that protocol, and build the game logic into the server.

线程服务器可能是这里最简单的事情.然后主线程启动一个游戏线程,该线程将大部分时间花在等待玩家行动的Condition 上.它还在 accept 上阻塞,为每个连接启动一个新的客户端线程,它大部分时间都在 self.sock.makefile(): 中的 for 行阻塞.在客户端对象中添加一个 Lock 以允许其他线程安全地发送消息.然后你只需要一组带有锁的客户端对象,就大功告成了.

A threaded server may be the simplest thing here. Then main thread kicks off a game thread, which spends most of its time blocking on a Condition waiting for players to act. It also blocks on accept, kicking off a new client thread for each connection, which spends most of its time blocking on for line in self.sock.makefile():. Add a Lock inside the client object to allow other threads to safely send messages. Then you just need a collection of client objects with a lock around it, and you're done.

因为我有一个类似设计的聊天服务器,让我修改它的一些部分来给你一个骨架.

Since I've got a chat server sitting around with a similar design, let me adapt some bits out of it to give you a skeleton.

首先,这是整个主线程:

First, here's the entire main thread:

lock = threading.Lock()
clients = []
game = Game()

ssock = socket.socket()
ssock.bind(('', 12345))
ssock.listen(5)
while True:
    sock, addr = ssock.accept()
    with lock:
        clients.append(Client(addr, sock, len(clients))

Client 对象是一个标准的调度器:

The Client object is a standard dispatcher:

class Client(object):
    def __init__(self, addr, sock, number):
        self.sock = sock
        self.name = '<{}> (not logged in)'.format(addr)
        self.number = number
        self.lock = threading.Lock()
        self.thread = threading.Thread(target=self.serve)
        self.thread.start()

    def send(self, msg):
        with self.lock:
            self.sock.send(msg)

    def run(self):
        for line in self.sock.makefile():
            args = line.rstrip().split()
            cmd = args.pop().upper()
            method = getattr(self, 'do_{}'.format(cmd), None)
            if method is none:
                self.write('ERROR unknown command {}\n'.format(cmd))
            else:
                try:
                    method(*args)
                except Exception as e:
                    self.send('ERROR in {}: {}\n'.format(cmd, e))
                else:
                    self.send('OK\n')

您可能还需要一个 broadcast 功能:

You probably also want a broadcast function:

def broadcast(msg):
    with lock:
        for client in clients:
            client.send(msg)

然后在 Client 上为每个命令编写方法.基本上,菜单代码中的每个 elif response == 'FOO' 都变成了 do_FOO 方法,而每个 print 变成了 广播,然后……就是这样.稍后我会展示一个更复杂的,但大多数看起来像这样:

Then you write methods on Client for each command. Basically, each elif response == 'FOO' you had in your menu code becomes a do_FOO method, and each print becomes a broadcast, and… that's about it. I'll show a more complicated one later, but here's what most of them will look like:

def do_SAY(self, *msg):
    broadcast('SAID {} {}'.format(self.number, ' '.join(msg)))

最后是 Game 对象.这在它自己的线程上运行,就像每个 Client 一样.在大多数情况下,它的 run 方法与您的顺序、非网络游戏中的逻辑相同.当然,您必须调用 broadcast 而不是 print,但这很容易.唯一棘手的是你需要一点同步.

Finally, there's the Game object. This runs on its own thread, just like each Client. For the most part, its run method is the same logic as in your sequential, non-networked game. Of course you have to call broadcast instead of print, but that's easy. The only tricky bit is that you need a bit of synchronization.

例如,在开始新手之前,您必须复制玩家列表(可能还有其他一些相关的游戏状态),以便其他线程可以在不影响当前游戏的情况下对其进行修改,并且您还需要等到有足够多的玩家,所以你不会在 1 个人玩自己的情况下开始手牌.所以:

For example, before starting a new hand, you have to copy the list of players (and maybe some other related game state) so other threads can modify it without affecting the current game, and you also need to wait until there are enough players so you don't go starting hands with 1 person playing himself. So:

def new_hand(self):
    with self.condition:
        while len(self.players) < 2:
            self.condition.wait()
        players = self.players
    # all your existing sequential logic

并且您需要添加一个 join 方法供客户端从自己的线程调用:

And you need to add a join method for the clients to call from their own threads:

def join(self, player):
    with self.condition:
        self.players.append(self)
        self.condition.notify()

因此,在 Client 对象中:

def do_JOIN(self, name):
    self.name = name
    game.join(self)
    broadcast('JOINED {} {}'.format(self.number, self.name)

让我们让等待投注尽可能复杂,看看即使在最坏的情况下也能轻松.如果你想下注,你可以.每个人都可以看到你的下注,如果情况发生变化,你就会下注(例如,如果你跟注,那么前面的人加注,你跟注他的新下注).所以,这就是我们要做的:

Let's make waiting for bets as complicated as possible, just to see how easy it is even in the worst case. If you want to bet out of turn, you can. Everyone can see your bet, and if circumstances change, you're committed (e.g., if you call, then the guy ahead of you raises, you're calling his new bet). So, here's what we do:

def wait_for_bets(self, bettor):
    with self.condition:
        while self.bets[self.bettor] is None:
            self.condition.wait()
        bettor, bet = self.bettor, self.bets[self.bettor]
        self.bets[self.bettor] = None
    # handle the bet

以下是 Client 提交投注的方式:

And here's how a Client submits a bet:

def bet(self, player, bet):
    with self.condition:
        self.bets[player] = bet
        self.condition.notify()

例如,在Client中:

def do_FOLD(self):
    game.bet(self, 'fold')

显然有一堆代码要写.但关键是,除了上面已经显示的内容,或者您​​现有的游戏中已经存在的内容之外,没有什么复杂的.

Obviously there's a bunch of code to write. But the point is that there's nothing complicated beyond what's already shown above, or already in your existing game.