python - 套接字实现上的连接被拒绝
问题描述
我已经尝试过在书中找到的代码，但我无法运行它。实际上，此代码引发的错误是：[Errno 111] Connection denied when the run_clientfunction running the line with socket.create_connection(address) as connection。有人知道原因吗？
代码如下：
class EOFError(Exception):
pass
class ConnectionBase:
def __init__(self, connection):
self.connection = connection
self.file = connection.makefile('rb')
def send(self, command):
line = command + '\n'
data = line.encode()
self.connection.send(data)
def receive(self):
line = self.file.readline()
if not line:
raise EOFError('Connection closed')
return line[:-1].decode()
import random
WARMER = 'Warmer'
COLDER = 'Colder'
UNSURE = 'Unsure'
CORRECT = 'Correct'
class UnknowCommandError(Exception):
pass
class Session(ConnectionBase):
def __init__(self, *args):
super().__init__(*args)
self._clear_state(None, None)
def _clear_state(self, lower, upper):
self.lower = lower
self.upper = upper
self.secret = None
self.guesses = []
def loop(self):
while command := self.receive():
match parts := command.split():
case ['PARAM', lower, upper]:
self.set_params(lower, upper)
case ['NUMBER', *_]:
self.send_number()
case ['REPORT', decision]:
self.receive_report(decision)
case _:
raise UnknowCommandError(command)
def set_params(self, lower, upper):
self._clear_state(lower, upper)
def next_guess(self):
if self.secret is not None:
return self.secret
while True:
guess = random.randint(self.lower, self.upper)
if guess not in self.guesses:
return guess
def send_number(self):
guess = self.next_guess()
self.guesses.append(guess)
self.send(format(guess))
def receive_report(self, decision):
last = self.guesses[-1]
if decision == CORRECT:
self.secret = last
print(f'Server: {last} is {decision}')
import contextlib
import math
class Client(ConnectionBase):
def __init__(self, *args):
super().__init__(*args)
self._clear_state()
def _clear_state(self):
self.secret = None
self.last_distance = None
@contextlib.contextmanager
def session(self, lower, upper, secret):
print(f'Guess a number between {lower} and {upper}!'
f'Shhhhhh, it is {secret}.')
self.secret = secret
self.send('PARAM {lower} {upper}')
try:
yield
finally:
self._clear_state()
self.send('PARAM 0 -1')
def request_numbers(self, count):
for _ in range(count):
self.send('NUMBER')
data = self.receive()
yield int(data)
if self.last_distance == 0:
return
def report_outcome(self, number):
new_distance = math.fabs(number - self.secret)
decision = UNSURE
if new_distance == 0:
decision = CORRECT
elif self.last_distance is None:
pass
elif new_distance < self.last_distance:
decision = WARMER
elif new_distance > self.last_distance:
decision = COLDER
self.last_distance = new_distance
self.send(f'REPORT {desicion}')
return decision
import socket
from threading import Thread
def handle_connection(connection):
with connection:
session = Session(connection)
try:
session.loop()
except EOFError:
pass
def run_server(address):
with socket.socket() as listener:
listener.bind(address)
listener.listen()
while True:
connection, _ = listener.accept()
thread = Thread(target=handle_connection,
args=(connection,),
daemon=True)
thread.start()
def run_client(address):
with socket.create_connection(address) as connection:
client = Client(connection)
with client.session(1, 5, 3):
results = [(x, client.report_outcome(x))
for x in client.request_numbers(5)]
with client.session(10, 15, 12):
for number in client.request_numbers(5):
outcome = client.report_outcome(number)
results.append((number, outcome))
return results
def main():
address = ('127.0.0.1', 1234)
server_thread = Thread(target=run_server,
args=(address,),
daemon=True)
server_thread.start()
results = run_client(address)
for number, outcome in results:
print(f'Client: {number} is {outcome}')
main()
此代码旨在创建一个基于 TCP 的服务器，用于玩涉及猜数字的游戏。服务器采用确定要考虑的数字范围的lower和upper参数。然后，服务器根据客户端的请求返回对该范围内整数值的猜测。最后，服务器从客户端收集关于这些数字中的每一个是更接近（更温暖）还是更远离（更冷）客户端的秘密号码的报告。
解决方案
好像哪里有错别字，我已经更正了。谢谢你的帮助@SteffenUllrich
这是工作代码：
class EOFError(Exception):
pass
class ConnectionBase:
def __init__(self, connection):
self.connection = connection
self.file = connection.makefile('rb')
def send(self, command):
line = command + '\n'
data = line.encode()
self.connection.send(data)
def receive(self):
line = self.file.readline()
if not line:
raise EOFError('Connection closed')
return line[:-1].decode()
import random
WARMER = 'Warmer'
COLDER = 'Colder'
UNSURE = 'Unsure'
CORRECT = 'Correct'
class UnknowCommandError(Exception):
pass
class Session(ConnectionBase):
def __init__(self, *args):
super().__init__(*args)
self._clear_state(None, None)
def _clear_state(self, lower, upper):
self.lower = lower
self.upper = upper
self.secret = None
self.guesses = []
def loop(self):
while command := self.receive():
match command.split():
case ['PARAM', lower, upper]:
self.set_params(int(lower), int(upper))
case ['NUMBER', *_]:
self.send_number()
case ['REPORT', decision]:
self.receive_report(decision)
case _:
raise UnknowCommandError(command)
def set_params(self, lower, upper):
self._clear_state(lower, upper)
def next_guess(self):
if self.secret is not None:
return self.secret
while True:
guess = random.randint(self.lower, self.upper)
if guess not in self.guesses:
return guess
def send_number(self):
guess = self.next_guess()
self.guesses.append(guess)
self.send(format(guess))
def receive_report(self, decision):
last = self.guesses[-1]
if decision == CORRECT:
self.secret = last
print(f'Server: {last} is {decision}')
import contextlib
import math
class Client(ConnectionBase):
def __init__(self, *args):
super().__init__(*args)
self._clear_state()
def _clear_state(self):
self.secret = None
self.last_distance = None
@contextlib.contextmanager
def session(self, lower, upper, secret):
print(f'Guess a number between {lower} and {upper}!'
f'Shhhhhh, it is {secret}.')
self.secret = secret
self.send(f'PARAM {lower} {upper}')
try:
yield
finally:
self._clear_state()
self.send('PARAM 0 -1')
def request_numbers(self, count):
for _ in range(count):
self.send('NUMBER')
data = self.receive()
yield int(data)
if self.last_distance == 0:
return
def report_outcome(self, number):
new_distance = math.fabs(number - self.secret)
decision = UNSURE
if new_distance == 0:
decision = CORRECT
elif self.last_distance is None:
pass
elif new_distance < self.last_distance:
decision = WARMER
elif new_distance > self.last_distance:
decision = COLDER
self.last_distance = new_distance
self.send(f'REPORT {decision}')
return decision
import socket
from threading import Thread
def handle_connection(connection):
with connection:
session = Session(connection)
try:
session.loop()
except EOFError:
pass
def run_server(address):
with socket.socket() as listener:
listener.bind(address)
listener.listen()
while True:
connection, _ = listener.accept()
thread = Thread(target=handle_connection,
args=(connection,),
daemon=True)
thread.start()
def run_client(address):
with socket.create_connection(address) as connection:
client = Client(connection)
with client.session(1, 5, 3):
results = [(x, client.report_outcome(x))
for x in client.request_numbers(5)]
with client.session(10, 15, 12):
for number in client.request_numbers(5):
outcome = client.report_outcome(number)
results.append((number, outcome))
return results
def main():
address = ('127.0.0.1', 1234)
server_thread = Thread(target=run_server,
args=(address,),
daemon=True)
server_thread.start()
results = run_client(address)
for number, outcome in results:
print(f'Client: {number} is {outcome}')
main()
PS0：要使其正常工作，您需要 python3.10 或更改loop,set_params和receive_report. PS1：在运行代码之前，请确保tcp:1234localhost 的端口未被使用。
推荐阅读
- php - 使用静态属性构建的子类上的 PHP 7.1.3 Foreach 打印出意外的输出
- perl - 如果我想获得命令的输出，如何修复错误“命令行太长”？
- vba - 访问 VBA 抛出“运行时错误 7 - 内存不足”
- javascript - 返回 jQuery 元素的函数返回未定义
- django - 无法模拟遍历 for 循环的 django 查询对象
- macos - 如何在 Xamarin.Mac 中的应用程序级别捕获按键事件
- angular - Angular - PrimeNg - 文件上传 - PDF 文件未在 AWS S3 存储桶中正确上传
- nativescript - 如何在 NativeScript 中调试/显示所有网络请求？
- python - 有什么方法可以测试并给出真/假语句以查看文件是否存在，Python
- unity3d - 玩家的跳跃限制